python脚本测试-接口高并发场景

python脚本测试——接⼝⾼并发场景⼀、环境准备。1、下载安装python2，安装教程请⾃⾏百度3、IDEA在file–settings–Plugins中安装Python⼆、复⽤需要修改的地⽅：1、url路径2、token信息3、连接数据库的信息4、sql语句5、请求⽅式：get、post、delete、put6、⽤户username的特殊性规则7、并发线程数：concurrent_thread_count8、请求数据在RequestBody中还是在RequestParam中三、添加学⽣。1、设计思路设置线程数，作为循环次数，特殊值+i作为学号，性别取1、2的随机数2、实现代码如下

#并发测试框架//添加学⽣importpymysqlimportrequests,threadingimportrandom#接⼝路径和tokendata={"url":"http://。。。/base/add","header":{"X-Access-Token":"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9."},}#并发线程数concurrent_thread_count=1000#每个线程中循环测试接⼝的次数ONE_WORKER_NUM=1#测试代码deftest(username):random.randint(1,2)#获取试题接⼝#参数在请求体中时，使⽤bodysbodys={#请求体参数"num":username,"name":"随删数据","gender":random.randint(1,2)}print(bodys)response=requests.post(data["url"],json=bodys,headers=data["header"])#response=requests.get(data["url"]+'?userId='+resultSql[0],headers=data["header"])ifresponse.status_code==200:result=response.content.decode('utf-8')else:result="访问失败"print(threading.current_thread(),result)#单个线程，可以设置⼀个线程访问⼏次接⼝defworking(username):globalONE_WORKER_NUMforiinrange(0,ONE_WORKER_NUM):test(username)deft():Threads=[]#根据数据的数量当作循环次数，每次创建⼀个线程foriinrange(concurrent_thread_count):username=111111000+it=threading.Thread(target=working,args=(username,))t.setDaemon(True)Threads.append(t)fortinThreads:t.start()fortinThreads:t.join()if__name__=="__main__":t()四、获取试题。1、设计思路如下1、第⼀步：先执⾏查询数据库操作，查询出来⼀共有多少条符合条件的数据；数据总数作为循环的次数，每次循环起⼀个线程，线程中带有参数username，根据添加⽤户时账号的规律设置username=“”"’%s’"""%(111111000+i)当作参数传递2、第⼆步：每个线程中可以设置循环次数，每次调⽤具体⽅法3、第三步：测试代码部分，根据传递的username作为条件查询⽤户id；获取到⽤户id后当作参数去访问接⼝，打印返回值2、实现代码如下（使⽤多个不同的⽤户id做参数）#并发测试框架importpymysqlimportrequests,threading#接⼝路径和tokendata={"url":"http://。。。/mobile/getAll","header":{"X-Access-Token":"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9"},}#每个线程中循环测试接⼝的次数ONE_WORKER_NUM=1#测试代码deftest(username):#根据username查询idconn=pymysql.connect(host="192.168...",port=3306,user="root",passwd="...",db="jeecg")cur=conn.cursor()sql="SELECTidFROMuserWHEREusername="+username#print(sql)cur.execute(sql)resultSql=cur.fetchone()#获取试题接⼝#参数在请求体中时，使⽤bodys#bodys={##请求体参数#"userId":resultSql[0]#}#response=requests.get(data["url"]+'?userId='+resultSql[0],json=bodys,headers=data["header"])print(data["url"]+'?userId='+resultSql[0])response=requests.get(data["url"]+'?userId='+resultSql[0],headers=data["header"])ifresponse.status_code==200:result=response.content.decode('utf-8')else:result="访问失败"print(threading.current_thread(),result)#单个线程，可以设置⼀个线程访问⼏次接⼝defworking(username):globalONE_WORKER_NUMforiinrange(0,ONE_WORKER_NUM):test(username)deft():#查询符合条件的数据总数conn=pymysql.connect(host="192.168...",port=3306,user="root",passwd="...",db="jeecg")cur=conn.cursor()sql="SELECTCOUNT(*)FROMuserWHEREusernameLIKE'11111%'"sql="SELECTCOUNT(*)FROMuserWHEREusernameLIKE'11111%'"#打印SQL#print(sql)cur.execute(sql)result=cur.fetchone()#打印循环次数print(result[0])Threads=[]#根据数据的数量当作循环次数，每次创建⼀个线程foriinrange(result[0]):username="""'%s'"""%(111111000+i)t=threading.Thread(target=working,args=(username,))#使⽤setDaemon()和守护线程这⽅⾯知识有关，⽐如在启动线程前设置thread.setDaemon(True)，就是设置该线程为守护线程，#表⽰该线程是不重要的进,程退出时不需要等待这个线程执⾏完成。#这样做的意义在于：避免⼦线程⽆限死循环，导致退不出程序，也就是避免楼上说的孤⼉进程。##thread.setDaemon（）设置为True,则设为true的话则主线程执⾏完毕后会将⼦线程回收掉,#设置为false,主进程执⾏结束时不会回收⼦线程#setDaemon()说明：#setDaemon()：设置此线程是否被主线程守护回收。默认False不回收，需要在start⽅法前调⽤；#设为True相当于像主线程中注册守护，主线程结束时会将其⼀并回收。t.setDaemon(True)#python中的append()⽅法⽤于在列表末尾添加新的对象。Threads.append(t)fortinThreads:#开始线程活动。#对每⼀个线程对象来说它只能被调⽤⼀次，它安排对象在⼀个另外的单独线程中调⽤run()⽅法（⽽⾮当前所处线程）。#当该⽅法在同⼀个线程对象中被调⽤超过⼀次时，会引RuntimeError(⼊运⾏时错误)。t.start()fortinThreads:t.join()if__name__=="__main__":t()五、删除学⽣。1、设计思路1、先查询数据库中有多少个需要删除的数据，作为循环次数，特殊值+i作为学号传到创建的线程⾥⾯2、再根据学号查学⽣的id，查到id后去请求删除接⼝2、代码实现如下#并发测试框架//删除学⽣importpymysqlimportrequests,threading#接⼝路径和tokendata={"url":"http://。。。/base/delete","header":{"X-Access-Token":"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9"},}#每个线程中循环测试接⼝的次数ONE_WORKER_NUM=1#测试代码deftest(username):#根据username查询idconn=pymysql.connect(host="。。。",port=3306,user="root",user="root",passwd="。。。",db="jeecg")cur=conn.cursor()sql="SELECTidFROMmes_stuWHEREnum="+username#print(sql)cur.execute(sql)resultSql=cur.fetchone()#获取试题接⼝#参数在请求体中时，使⽤bodys#bodys={##请求体参数#"userId":resultSql[0]#}#response=requests.get(data["url"]+'?userId='+resultSql[0],json=bodys,headers=data["header"])print(data["url"]+'?userId='+resultSql[0])response=requests.delete(data["url"]+'?id='+resultSql[0],headers=data["header"])ifresponse.status_code==200:result=response.content.decode('utf-8')else:result="访问失败"print(threading.current_thread(),result)#单个线程，可以设置⼀个线程访问⼏次接⼝defworking(username):globalONE_WORKER_NUMforiinrange(0,ONE_WORKER_NUM):test(username)deft():#查询符合条件的数据总数conn=pymysql.connect(host="。。。",port=3306,user="root",passwd="。。。",db="jeecg")cur=conn.cursor()sql="SELECTCOUNT(*)FROMmes_stuWHEREnumLIKE'11111%'"#打印SQL#print(sql)cur.execute(sql)result=cur.fetchone()#打印循环次数print(result[0])Threads=[]foriinrange(result[0]):#根据学号规律拼接username="""'%s'"""%(111111000+i)t=threading.Thread(target=working,args=(username,))t.setDaemon(True)Threads.append(t)fortinThreads:t.start()fortinThreads:t.join()if__name__=="__main__":t()六、直接并发（参数需要⾃⼰写）importrequestsfrommultiprocessingimportPooldata={"times":10,#并发量"method":"GET","url":"http://。。。/list","header":{#"X-Access-Token":"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9"},"body":{#参数}}defrun_task(idx):response=requests.get(data["url"],json=data["body"],headers=data["header"])ifresponse.status_code==200:result=response.content.decode('utf-8')else:result="访问失败"print("第%s次执⾏：%s\n"%(idx,result))if__name__=='__main__':p=Pool(data["times"])forindexinrange(data["times"]):p.apply_async(run_task,args=(index+1,))p.close()p.join()print("执⾏结束.")七、多个线程并⾏，每个线程内串⾏执⾏多次（可⼀次），参数需要⾃⼰写importthreadingimporttimeimportrequests#这段脚本是本机执⾏，请求QA100个线程并⾏，每个线程内串⾏执⾏100次，⼀共1W次接⼝调⽤环境，data={"url":"http://。。。/list","header":{#"X-Access-Token":"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9"},"body":{#参数}}#concurrentrequestnumberconcurrent_thread_count=5#loopineverythreaditerate_count=1classRequestThread(threading.Thread):def__init__(self,thread_name):threading.Thread.__init__(self)self.test_success_count=0self.test_fail_count=0defrun(self):i=0i=0whilei<iterate_count:self.test()i+=1deftest(self):try:conn=requests.get(data["url"],json=data["body"],