在工作中遇到要对开发的接口做压力测试,以前没有做过开清楚什么压测工具好用,正好接口不是什么复杂的接口,curl -X post "接口地址" --data-binary @二进制认证文件 OK!(@表示验证数据是文件类型)
既然这样那我就写个脚本好了,脚本内容如下:
======================================================================
#!/usr/bin/evn python
#_coding:utf8_
from multiprocessing import Pool,Queue
import time,subprocess,os
class YaCe(object):
def init(self,api,binfile,maxpool,qu,maxrequest=100000,status="success"):
self.api = api
self.binfile = binfile
self.status = status
self.maxpool = maxpool
self.maxrequest = maxrequest
self.qu = qu
def prorequest(self):
for i in range(self.maxrequest):
self.qu.put(i)
print(i)
for i in range(int(self.maxpool)):
self.qu.put(None)
print("None")
def conumers(self,i):
while True:
data = self.qu.get(True)
if data == None:
print("进程%s任务完成..."%i)
break
else:
command = subprocess.getoutput("time curl -X POST --connect-timeout 10 '%s' --data-binary @%s"%(self.api,os.getcwd()+"/"+self.binfile))
if self.status == "success":
logfile = os.getcwd()+"/"+"headbeat.log"+"_%s"%date_time
if "CgoyMDAwMDAwMDAw" in command:
print("进程%s__%s..."%(str(i),str(data)))
with open(logfile,"a") as f:
f.write(command+"\n")
f.close()
else:
print("进程%s__%s..."%(str(i),str(data)))
with open(logfile,"a") as f:
f.write("Faild\n")
f.write(command+"\n")
f.close()
else:
logfile = os.getcwd()+"/"+"roomlist.log"+"_%s"%date_time
#print("time curl -X POST '%s' --data-binary @%s"%(self.api,os.getcwd()+"/"+self.binfile))
command = subprocess.getoutput("time curl -X POST '%s' --data-binary @%s"%(self.api,os.getcwd()+"/"+self.binfile))
if "CAES+" in command:
print("进程%s__%s..."%(str(i),str(data)))
info = command.split('\n')[-3:]
info1 = "\n".join(info)
with open(logfile,"a") as f:
f.write(info1+"\n")
f.close()
else:
print("进程%s__%s..."%(str(i),str(data)))
with open(logfile,"a") as f:
f.write("Faild\n")
f.write(command+"\n")
f.close()
def multirun(self):
ps = int(int(self.maxpool) - 1)
p = Pool(ps)
for i in range(self.maxpool):
print("开启子进程%s"%i)
p.apply_async(self.conumers,args=(self,i))
print('等待所有添加的进程运行完毕。。。')
p.close()
p.join()
endtime = time.strftime("%Y%m%d_%X",time.localtime())
if self.status == "success":
logfile = os.getcwd()+"/"+"headbeat.log"+"_%s"%date_time
else:
logfile = os.getcwd() + "/" + "roomlist.log"+"_%s"%date_time
with open(logfile,"a") as f:
f.write("============[%s]============\n"%endtime)
f.close()
print('End!!,PID:%s'% os.getpid())
if name == "main":
q = Queue()
Yc = YaCe('压测接口','二进制证认文件',开多少个进程,queue(队列),maxrequest=100(模拟测试多少次访问),status="faild"(这里因为测试的两个接口,返回不一样用status参数区分测试的接口的返回值处理))
Yc.prorequest()
print("++++++")
global date_time
datetime = time.strftime("%Y%m%d%X",time.localtime())
Yc.multirun()
====================================================================================
问题
到这里写完了,测试的问题来了,从脚本来看如果运行成功,会有多进程在处理队列的输出,可是结果的输出确是如下:
01
2
3
4
5
6
7
8
9
None
None
++++++
开启子进程0
开启子进程1
等待所有添加的进程运行完毕。。。
End!!,PID:4819
原因
子进程conumers方法完全没有运行,也没有报错这就尴尬了;查了大量的文档资料;发现这个pool方法都使用了queue.Queue将task传递给工作进程。multiprocessing必须将数据序列化以在进程间传递。方法只有在模块的顶层时才能被序列化,跟类绑定的方法不能被序列化,就会出现上面的异常 ; 那肿么办,我不是一个轻易放弃的人,终于被我找到了方法;
注意
解决方作者是在python3下测试了,python2下用脚本的subprocess要换成value,command = commands.getstatusoutput
解决方法1(亲测)
1.首先要看报错,需要对脚本修改如下:
YaCe类下的multirun方法下修改
for i in range(self.maxpool):
print("开启子进程%s"%i)
p.apply_async(self.conumers,args=(self,i))
为
for i in range(self.maxpool):
print("开启子进程%s"%i)
res = p.apply_async(self.conumers,args=(self,i))
print(res.get)
这就可以看到报错:
cPickle.PicklingError: Can't pickle <type 'instancemethod'>: attribute lookup builtin.instancemethod failed
2.解决方法如下在脚本中加一个新的函数
(1).def conumers_wrapper(cls_instance,i):
return cls_instance.conumers(i)
(2).修改YaCe下multirun方法
for i in range(self.maxpool):
print("开启子进程%s"%i)
res = p.apply_async(self.conumers,args=(self,i))
print(res.get())
为
for i in range(self.maxpool):
print("开启子进程%s"%i)
res = p.apply_async(conumers_wrapper,args=(self,i))
print(res.get)
问题解决了,运行一下脚本结果还有报错:
RuntimeError: Queue objects should only be shared between processes through inheritance
原因
这里不可以用Queue,要改用Manager.Queue;因为进程之前的同共离用Queue会用问题;
完结
最终代码如下:
==================================================================================
#!/usr/bin/evn python
#_coding:utf8_
from multiprocessing import Pool,Queue,Manager
import time,subprocess,os
class YaCe(object):
def init(self,api,binfile,maxpool,qu,maxrequest=100000,status="success"):
self.api = api
self.binfile = binfile
self.status = status
self.maxpool = maxpool
self.maxrequest = maxrequest
self.qu = qu
def prorequest(self):
for i in range(self.maxrequest):
self.qu.put(i)
print(i)
for i in range(int(self.maxpool)):
self.qu.put(None)
print("None")
def conumers(self,i):
while True:
data = self.qu.get(True)
if data == None:
print("进程%s任务完成..."%i)
break
else:
#print("time curl -X POST '%s' --data-binary @%s"%(self.api,os.getcwd()+"/"+self.binfile))
command = subprocess.getoutput("time curl -X POST --connect-timeout 10 '%s' --data-binary @%s"%(self.api,os.getcwd()+"/"+self.binfile))
#command = subprocess.getoutput("time curl -X POST '%s' --data-binary @%s"%(self.api,os.getcwd()+"/"+self.binfile))
if self.status == "success":
logfile = os.getcwd()+"/"+"headbeat.log"+"_%s"%date_time
if "CgoyMDAwMDAwMDAw" in command:
print("进程%s__%s..."%(str(i),str(data)))
with open(logfile,"a") as f:
f.write(command+"\n")
f.close()
else:
with open(logfile,"a") as f:
f.write("Faild\n")
f.write(command+"\n")
f.close()
else:
logfile = os.getcwd()+"/"+"roomlist.log"+"_%s"%date_time
#print("time curl -X POST '%s' --data-binary @%s"%(self.api,os.getcwd()+"/"+self.binfile))
command = subprocess.getoutput("time curl -X POST --connect-timeout 10 '%s' --data-binary @%s"%(self.api,os.getcwd()+"/"+self.binfile))
#command = subprocess.getoutput("time curl -X POST '%s' --data-binary @%s"%(self.api,os.getcwd()+"/"+self.binfile))
if "CAES+" in command:
print("进程%s__%s..."%(str(i),str(data)))
info = command.split('\n')[-3:]
info1 = "\n".join(info)
with open(logfile,"a") as f:
f.write(info1+"\n")
f.close()
else:
print("进程%s__%s..."%(str(i),str(data)))
with open(logfile,"a") as f:
f.write("Faild\n")
f.write(command+"\n")
f.close()
def multirun(self):
ps = int(int(self.maxpool) - 1)
p = Pool(ps)
for i in range(self.maxpool):
print("开启子进程%s"%i)
p.apply_async(conumers_wrapper,args=(self,i))
#print(res.get)
print('等待所有添加的进程运行完毕。。。')
p.close()
p.join()
endtime = time.strftime("%Y%m%d_%X",time.localtime())
if self.status == "success":
logfile = os.getcwd()+"/"+"headbeat.log"+"_%s"%date_time
else:
logfile = os.getcwd() + "/" + "roomlist.log"+"_%s"%date_time
with open(logfile,"a") as f:
f.write("============[%s]============\n"%endtime)
f.close()
print('End!!,PID:%s'% os.getpid())
def conumers_wrapper(cls_instance,i):
return cls_instance.conumers(i)
if name == "main":
q = Manager().Queue()
Yc = YaCe('压测接口','二进制证认文件',开多少个进程,queue(队列),maxrequest=100(模拟测试多少次访问),status="faild"(这里因为测试的两个接口,返回不一样用status参数区分测试的接口的返回值处理))
Yc.prorequest()
print("++++++")
global date_time
datetime = time.strftime("%Y%m%d%X",time.localtime())
Yc.multirun()
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。