python进行接口压力测试（1秒1000个请求）

需求

1. 需要对接口进行压力测试，测试其最大并发是多少

python代码

import json
import requests
import time
import concurrent.futures
import threadingSUCCESS = 0
FAIL = 0;def Api(name,pwd):registerUrl = &＃39;http://localhost:8001/api/UserInfo/Login&＃39;data = {&＃39;name&＃39;:name,&＃39;password&＃39;:pwd}paramData = json.dumps(data)headers = {&＃39;Content-Type&＃39;:&＃39;application/json;charset=utf-8&＃39;}try:response = requests.post(registerUrl,data=paramData,headers=headers,timeout=3)resObj = json.loads(response.text)except Exception as ex:global FAILFAIL += 1print(&＃39;失败次数={}&＃39;.format(FAIL))print(str(ex))finally:global SUCCESSSUCCESS += 1# resObj = json.loads(response.text)# print((resObj[&＃39;data&＃39;][&＃39;identifier&＃39;])) class MyThread(threading.Thread):&＃39;&＃39;&＃39;线程类&＃39;&＃39;&＃39;def __init__(self,name,pwd):threading.Thread.__init__(self)self.name = nameself.pwd = pwddef run(self):&＃39;&＃39;&＃39;线程创建后会直接运行run函数&＃39;&＃39;&＃39;Api(self.name,self.pwd)class PressuerTest:def simple(self,num):&＃39;&＃39;&＃39;单线程，不做任何处理&＃39;&＃39;&＃39;for i in range(num):name = "gagag" + str(i)pwd = &＃39;passwrodtest&＃39;Api(name,pwd)def _BatchRegister(self,num):name = "gagag" + str(num)pwd = &＃39;passwrodtest&＃39;Api(name,pwd)def _BatchRegisterAll(self,numList):&＃39;&＃39;&＃39;并发这里创建一个线程池，总共有5个线程可以分配使用executor.map()与map()函数类似，表示对sites中的每一个元素，并发地调用函数&＃39;&＃39;&＃39;with concurrent.futures.ThreadPoolExecutor(max_workers=80) as executor:executor.map(self._BatchRegister, numList)def mult(self,num):&＃39;&＃39;&＃39;线程池&＃39;&＃39;&＃39;numList = []for i in range(num):numList.append(i)self._BatchRegisterAll(numList)def newMultThread(self,num):&＃39;&＃39;&＃39;新建多个线程&＃39;&＃39;&＃39;threadArr = []for i in range(num):name = "newMultThread_" + str(i)pwd = &＃39;pwd&＃39;thread = MyThread(name,pwd)thread.start()threadArr.append(thread)if __name__ == &＃39;__main__&＃39;:startTime = time.strftime(&＃39;%Y-%m-%d %H:%M:%S&＃39;,time.localtime(time.time()))pre = PressuerTest()# pre.simple(100);# pre.mult(100);pre.newMultThread(2000)endTime = time.strftime(&＃39;%Y-%m-%d %H:%M:%S&＃39;,time.localtime(time.time()))print(&＃39;开始时间：{}&＃39;.format(startTime)) print(&＃39;结束时间：{}&＃39;.format(endTime))


总结

代码中一共使用了3种方式取执行压力测试，

1. 单线程明显的会慢很多
2. 多线程的采用线程池的话也达不到1秒1000个请求，会在一个线程执行完才会走到下一个线程
3. 每个请求一个线程可以达到效果，线程会很快启动完，可以达到1秒1000个的高并发量（也可以更高)

额外因素

1. 客户端的设置的超时时间会对并发数量有影响。
2. 服务器的MinRequestBodyDataRate也会有影响（针对.netCore）