需求：用4000用户随机登录调用接口性能压测
思路
1、随机取用户函数
2、将随机取用户函数放到队列中
3、使用的时候从队列依次取出

from locust import HttpUser, task, between,TaskSet
from gevent._semaphore import Semaphore
import queue
from cloudDataFactory.publicCenter.excle_data import ExcelHander
from cloudDataFactory.cloud_tools.operopration import get_client_token
import os
import random
#获取账号信息
filepath = r'C:\Users\WS\Desktop\uc_user_wsid_email.xls'
wsid_dict = ExcelHander(filepath).ExcelDick("uc_user_wsid_email")["uc_user_wsid_email"]def random_wsid():
#随机取用户函数return random.choice(wsid_dict)wsid_email = random_wsid()def GetData():s=queue.Queue(maxsize=0)for i in range(4000):
#将用户信息放到队列s.put_nowait(wsid_email)return s
class wgpTasks(TaskSet):@taskdef messagesV1(self):# wsid_email = random_wsid()
#使用的时候从队列依次取出Authorization = json.loads(get_client_token(username=wsid_email["email"], password="1234qwer"))["data"]["access_token"]headers = {}self.client.get('/v1/wms/rule/messages?pid=846', headers=headers)@taskdef messagesV2(self):# wsid_email = random_wsid()#使用的时候从队列依次取出Authorization = json.loads(get_client_token(username=wsid_email["email"], password="1234qwer"))["data"]["access_token"]headers = {"Authorization": f"Bearer {Authorization}"}self.client.get('/v2/xxxx',headers=headers)@taskdef messagesRuleV1(self):# wsid_email = random_wsid()#使用的时候从队列依次取出Authorization = json.loads(get_client_token(username=wsid_email["email"], password="1234qwer"))["data"]["access_token"]headers = {}self.client.get('/v1xxxxxx',headers=headers)
class websitUser(HttpUser):tasks = [wgpTasks]wait_time = between(1, 3)host = "xxxxxx"#队列取数queueData=GetData()if __name__ == "__main__":os.system(" locust -f .\performace_locust1.py --web-host=10.90.12.148 --web-port=8090")```