在工作中遇到要對(duì)開發(fā)的接口做壓力測(cè)試，以前沒(méi)有做過(guò)開清楚什么壓測(cè)工具好用，正好接口不是什么復(fù)雜的接口，curl -X post "接口地址" --data-binary @二進(jìn)制認(rèn)證文件 OK！（@表示驗(yàn)證數(shù)據(jù)是文件類型）

創(chuàng)新互聯(lián)公司10多年成都定制網(wǎng)站服務(wù);為您提供網(wǎng)站建設(shè),網(wǎng)站制作,網(wǎng)頁(yè)設(shè)計(jì)及高端網(wǎng)站定制服務(wù),成都定制網(wǎng)站及推廣,對(duì)成都玻璃鋼雕塑等多個(gè)領(lǐng)域擁有多年的網(wǎng)站推廣經(jīng)驗(yàn)的網(wǎng)站建設(shè)公司。

既然這樣那我就寫個(gè)腳本好了,腳本內(nèi)容如下:

======================================================================
#!/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("進(jìn)程%s任務(wù)完成..."%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("進(jìn)程%s__%s..."%(str(i),str(data)))
                    with open(logfile,"a") as f:
                        f.write(command+"\n")
                    f.close()
                else:
                                        print("進(jìn)程%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("進(jìn)程%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("進(jìn)程%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("開啟子進(jìn)程%s"%i)
        p.apply_async(self.conumers,args=(self,i))
    print('等待所有添加的進(jìn)程運(yùn)行完畢。。。')
    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('壓測(cè)接口','二進(jìn)制證認(rèn)文件',開多少個(gè)進(jìn)程,queue(隊(duì)列),maxrequest=100（模擬測(cè)試多少次訪問(wèn)）,status="faild"（這里因?yàn)闇y(cè)試的兩個(gè)接口，返回不一樣用status參數(shù)區(qū)分測(cè)試的接口的返回值處理）)
Yc.prorequest()
print("++++++")
global date_time
datetime = time.strftime("%Y%m%d%X",time.localtime())
Yc.multirun()

====================================================================================
問(wèn)題
到這里寫完了，測(cè)試的問(wèn)題來(lái)了，從腳本來(lái)看如果運(yùn)行成功，會(huì)有多進(jìn)程在處理隊(duì)列的輸出，可是結(jié)果的輸出確是如下：

01
2
3
4
5
6
7
8
9
None
None
++++++
開啟子進(jìn)程0
開啟子進(jìn)程1
等待所有添加的進(jìn)程運(yùn)行完畢。。。
End!!,PID:4819

原因
子進(jìn)程conumers方法完全沒(méi)有運(yùn)行，也沒(méi)有報(bào)錯(cuò)這就尷尬了；查了大量的文檔資料；發(fā)現(xiàn)這個(gè)pool方法都使用了queue.Queue將task傳遞給工作進(jìn)程。multiprocessing必須將數(shù)據(jù)序列化以在進(jìn)程間傳遞。方法只有在模塊的頂層時(shí)才能被序列化，跟類綁定的方法不能被序列化，就會(huì)出現(xiàn)上面的異常 ; 那腫么辦，我不是一個(gè)輕易放棄的人，終于被我找到了方法;

注意
解決方作者是在python3下測(cè)試了，python2下用腳本的subprocess要換成value,command = commands.getstatusoutput

解決方法1（親測(cè)）
1.首先要看報(bào)錯(cuò)，需要對(duì)腳本修改如下:
YaCe類下的multirun方法下修改
for i in range(self.maxpool):
print("開啟子進(jìn)程%s"%i)
p.apply_async(self.conumers,args=(self,i))
為
for i in range(self.maxpool):
print("開啟子進(jìn)程%s"%i)
res = p.apply_async(self.conumers,args=(self,i))
print(res.get)
這就可以看到報(bào)錯(cuò):
cPickle.PicklingError: Can't pickle <type 'instancemethod'>: attribute lookup builtin.instancemethod failed

2.解決方法如下在腳本中加一個(gè)新的函數(shù)
(1).def conumers_wrapper(cls_instance,i):
return cls_instance.conumers(i)

(2).修改YaCe下multirun方法
for i in range(self.maxpool):
print("開啟子進(jìn)程%s"%i)
res = p.apply_async(self.conumers,args=(self,i))
print(res.get())
為
for i in range(self.maxpool):
print("開啟子進(jìn)程%s"%i)
res = p.apply_async(conumers_wrapper,args=(self,i))
print(res.get)

問(wèn)題解決了，運(yùn)行一下腳本結(jié)果還有報(bào)錯(cuò)：
RuntimeError: Queue objects should only be shared between processes through inheritance

原因
這里不可以用Queue,要改用Manager.Queue；因?yàn)檫M(jìn)程之前的同共離用Queue會(huì)用問(wèn)題；

完結(jié)
最終代碼如下:

==================================================================================
#!/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("進(jìn)程%s任務(wù)完成..."%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("進(jìn)程%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("進(jìn)程%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("進(jìn)程%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("開啟子進(jìn)程%s"%i)
        p.apply_async(conumers_wrapper,args=(self,i))
    #print(res.get)
    print('等待所有添加的進(jìn)程運(yùn)行完畢。。。')
    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('壓測(cè)接口','二進(jìn)制證認(rèn)文件',開多少個(gè)進(jìn)程,queue(隊(duì)列),maxrequest=100（模擬測(cè)試多少次訪問(wèn)）,status="faild"（這里因?yàn)闇y(cè)試的兩個(gè)接口，返回不一樣用status參數(shù)區(qū)分測(cè)試的接口的返回值處理）)
Yc.prorequest()
print("++++++")
global date_time
datetime = time.strftime("%Y%m%d%X",time.localtime())
Yc.multirun()

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)cdcxhl.cn，海內(nèi)外云服務(wù)器15元起步，三天無(wú)理由+7*72小時(shí)售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國(guó)服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡(jiǎn)單易用、服務(wù)可用性高、性價(jià)比高”等特點(diǎn)與優(yōu)勢(shì)，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場(chǎng)景需求。

分享名稱：python下Pool與target方法寫在同一個(gè)類里要注意的坑-創(chuàng)新互聯(lián)
鏈接分享：http://newbst.com/article48/dgipep.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供企業(yè)網(wǎng)站制作、企業(yè)建站、網(wǎng)站維護(hù)、網(wǎng)站導(dǎo)航、品牌網(wǎng)站建設(shè)、App設(shè)計(jì)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)