python多進(jìn)程及通信實(shí)現(xiàn)異步任務(wù)的方法

第python多進(jìn)程及通信實(shí)現(xiàn)異步任務(wù)的方法目錄一、python多進(jìn)程及通信基本用法1、多進(jìn)程的基本實(shí)現(xiàn)a、Process重寫run方法b、使用Process和target方法c、直接使用Process類2、多進(jìn)程的通信a、Queueb、Pipe二、python多進(jìn)程實(shí)戰(zhàn)1、使用進(jìn)程池快速抽取數(shù)據(jù)2、多進(jìn)程及通信完成數(shù)據(jù)清洗和保存3、多進(jìn)程及通信實(shí)現(xiàn)異步任務(wù)需求寫在最前面，說(shuō)實(shí)話python多進(jìn)程這塊兒知識(shí)對(duì)于很少使用python多進(jìn)程或者沒(méi)有實(shí)際使用過(guò)多python進(jìn)程解決問(wèn)題的人來(lái)說(shuō)，還是有一定難度的。本人也是很少接觸多進(jìn)程的場(chǎng)景，對(duì)于python多進(jìn)程的使用也是比較陌生的。在接觸了一些多進(jìn)程的業(yè)務(wù)場(chǎng)景下，對(duì)python多進(jìn)程的使用進(jìn)行了學(xué)習(xí)，覺(jué)得很有必要進(jìn)行一個(gè)梳理總結(jié)。

一、python多進(jìn)程及通信基本用法

python中多進(jìn)程及其通信，是比較重要的一塊兒內(nèi)容，作為python程序員，這塊兒內(nèi)容要基本掌握。

1、多進(jìn)程的基本實(shí)現(xiàn)

python多進(jìn)程的使用一般是調(diào)用multiprocessing包中的Process和Pool(進(jìn)程池)，其中Process的用法又有多種，基本函數(shù)

p.start()啟動(dòng)一個(gè)已經(jīng)初始化的進(jìn)程

p.join()讓進(jìn)程運(yùn)行完了以后，主進(jìn)程再執(zhí)行

a、Process重寫run方法

MultiOneProcess類繼承了multiprocessing的Process類，然后重寫它的run方法，實(shí)現(xiàn)具體業(yè)務(wù)邏輯功能；主程序啟動(dòng)10個(gè)進(jìn)程。

frommultiprocessingimportProcess

count=0

classMultiOneProcess(Process):

def__init__(self,name):

super().__init__()

=name

defrun(self)-None:

globalcount

count+=1

print('processname%sisrunning----count:%d'%(,count))

if__name__=='__main__':

p_list=[]

foriinrange(10):

name='process_%d'%i

p=MultiOneProcess(name=name)

p.start()

p_list.append(p)

forpinp_list:

p.join()

print('thismainprocess')

b、使用Process和target方法

定義一個(gè)進(jìn)程類繼承Process類，同時(shí)在super()初始化中傳入target函數(shù)

frommultiprocessingimportProcess

count=0

classMultiTwoProcess(Process):

def__init__(self,name):

super().__init__(target=self.do_fun)

=name

defdo_fun(self):

globalcount

count+=1

print('processname%sisrunning----count:%d'%(name,count))

if__name__=='__main__':

p_list=[]

foriinrange(10):

name='process_%d'%i

p=MultiTwoProcess(name)

p.start()

p_list.append(p)

forpinp_list:

p.join()

print('thismainprocess')

代碼中定義了一個(gè)類MultiTwoProcess類，類中定義了do_fun函數(shù)，把它作為參數(shù)傳入到target中。

c、直接使用Process類

傳入target函數(shù)，同時(shí)傳入args參數(shù)，注意args參數(shù)是一個(gè)元組，切不能省略最后一個(gè)逗號(hào)

frommultiprocessingimportProcess

count=0

defdo_fun(name):

globalcount

count+=1

print('processname%sisrunning----count:%d'%(name,count))

if__name__=='__main__':

p_list=[]

foriinrange(10):

name='process_%d'%i

p=Process(target=do_fun,args=(name,))

p.start()

p_list.append(p)

forpinp_list:

p.join()

print('thismainprocess')

以上三者運(yùn)行的結(jié)果，是一樣的，如下：

2、多進(jìn)程的通信

進(jìn)程之間的通信一般都采用Queue和pipe，區(qū)別是：pipe只能在兩個(gè)進(jìn)程之間調(diào)用，而Queue是可以多個(gè)進(jìn)程間調(diào)用的；效率上pipe效率更高，Queue是基于pipe實(shí)現(xiàn)的，效率比pipe要低一點(diǎn)。

a、Queue

常用API，

存放數(shù)據(jù)

queue.put(obj,block=True,timeout=None)

當(dāng)block=False的時(shí)候，如果Queue已經(jīng)滿了，那么就會(huì)跑出Queue.Full異常；

當(dāng)block=True且timeout有正值的時(shí)候，Queue已經(jīng)滿了，Queue會(huì)阻塞timeout時(shí)間，超出時(shí)間就會(huì)拋出同樣的異常

獲取數(shù)據(jù)

queue.get(block=True,timeout=None)

當(dāng)block=False的時(shí)候，如果Queue為空，那么就會(huì)跑出Queue.Empty異常；

當(dāng)block=True且timeout有正值的時(shí)候，Queue已經(jīng)為空，Queue會(huì)阻塞timeout時(shí)間，超出時(shí)間就會(huì)拋出同樣的異常

以上2個(gè)API是阻塞；還有兩個(gè)非堵塞的API

queue.put(obj,block=False)和queue.put_nowait(obj)等效

queue.get(block=False)和queue.get_nowait()等效

簡(jiǎn)單的實(shí)現(xiàn)，一個(gè)進(jìn)程發(fā)送數(shù)據(jù)，另外2個(gè)進(jìn)程接收數(shù)據(jù)，就可以使用queue通信

frommultiprocessingimportProcess,Queue

defsend(q):

whileTrue:

q.put('發(fā)送一個(gè)數(shù)據(jù)')

defreceive1(q):

whileTrue:

s=q.get()

print('receive1:',s)

defreceive2(q):

whileTrue:

s=q.get()

print('receive2:',s)

if__name__=='__main__':

q=Queue()

p1=Process(target=send,args=(q,))

p2=Process(target=receive1,args=(q,))

p3=Process(target=receive2,args=(q,))

p1.start()

p2.start()

p3.start()

p1進(jìn)程不斷的往q中存放數(shù)據(jù)；p2和p3不停的從q中取數(shù)據(jù)(有競(jìng)爭(zhēng)的再取)，所以打印結(jié)果是無(wú)序的

b、Pipe

Pipe(duplex=True)返回2個(gè)連通端(p1,p2)；當(dāng)duplex=True時(shí)，雙向通信，p1發(fā)送，p2接收；p2發(fā)送，p1接收。

當(dāng)duplex=True時(shí)，單向通信，p1只能發(fā)送，p2只能接收。

常用API，pipe.send()pipe.recv()

frommultiprocessingimportProcess,Pipe

deffun2(p):

whileTrue:

s=p.recv()

print('接收一個(gè)數(shù)據(jù)：',s)

deffun1(p):

whileTrue:

print('發(fā)送一個(gè)數(shù)據(jù)：pipe')

p.send('pipe')

if__name__=='__main__':

pi1,pi2=Pipe(duplex=True)

p1=Process(target=fun1,args=(pi1,))

p2=Process(target=fun2,args=(pi2,))

p1.start()

p2.start()

結(jié)果如下：

二、python多進(jìn)程實(shí)戰(zhàn)

不同的業(yè)務(wù)場(chǎng)景使用多進(jìn)程的方式和復(fù)雜度也不相同，就我遇見過(guò)的一些場(chǎng)景進(jìn)行演示和說(shuō)明。

1、使用進(jìn)程池快速抽取數(shù)據(jù)

場(chǎng)景描述：有1000個(gè)Excel文件的數(shù)據(jù)需要進(jìn)行抽取和清洗，要把不符合我們需求的數(shù)據(jù)過(guò)濾掉，保留質(zhì)量很高的數(shù)據(jù)；每個(gè)Excel都有幾十萬(wàn)或者上百萬(wàn)的數(shù)據(jù)，那么怎么快速的完成這個(gè)任務(wù)呢？

首先整體上而言，可以把單個(gè)Excel的處理并行起來(lái)；那么可以使用多進(jìn)程，其次這個(gè)需要返回結(jié)果，要保留合格的數(shù)據(jù)，比較簡(jiǎn)單的就是采用進(jìn)程池了，它能夠很方便的把進(jìn)程處理的結(jié)果進(jìn)行返回，并且返回的還是一個(gè)生成器；如果還需要更快，那么可以把單個(gè)Excel中的每條數(shù)據(jù)的處理并行起來(lái)。代碼層面上，采用pool進(jìn)程池來(lái)完成這個(gè)任務(wù)(本文沒(méi)有對(duì)進(jìn)程池的使用和API做說(shuō)明)，具體的實(shí)現(xiàn)方式采取pool.imap()

if__name__=='__main__':

#所有Excel的路徑

all_paths=glob('../data/original_data/*')

sysInfo_list=['我通過(guò)了好友請(qǐng)求，現(xiàn)在你倆可以開始聊天了','我通過(guò)了你的朋友驗(yàn)證請(qǐng)求，現(xiàn)在我們可以開始聊天了','已通過(guò)你的朋友驗(yàn)證請(qǐng)求，現(xiàn)在可以開始聊天了','不支持此消息，請(qǐng)?jiān)谑謾C(jī)上查看',

'微信紅包']

interval=25

iflen(all_paths)//interval*intervallen(all_paths):

k=len(all_paths)//interval+1

else:

k=len(all_paths)//interval

#分段處理，每段25個(gè)Excel

foriinrange(k):

paths=all_paths[i*interval:(i+1)*interval]

ifi*interval=100andi*interval200:

params=[]

forpathintqdm(paths):

params.append((path,sysInfo_list))

#多進(jìn)程處理——進(jìn)程池、以及進(jìn)度顯示

withPool(20)asp:

res=list(tqdm(p.imap(extract_data,params),total=len(params),desc='extract_data'))

all_df=[]

fordfsinres:

iflen(dfs)0:

all_df.extend(dfs)

df=pd.concat(all_df,axis=0)

save_path='../data/weikong_clean_data_'+str(i*interval)+'_'+str(i*interval+len(paths)-1)+'.xlsx'

writer=pd.ExcelWriter(save_path)

df.to_excel(writer,index=False)

writer.save()

writer.close()

2、多進(jìn)程及通信完成數(shù)據(jù)清洗和保存

場(chǎng)景描述：從Excel中讀取數(shù)據(jù)，數(shù)據(jù)格式是整通整通的對(duì)話，每通對(duì)話有一定的輪數(shù)；保存數(shù)據(jù)到2個(gè)txt中，一個(gè)是順序保留，一個(gè)是倒序保留；整體對(duì)話順序不變，每通對(duì)話內(nèi)部順序倒序。

正序：

倒序：

要想實(shí)現(xiàn)這樣的任務(wù)，粗暴的做法是，用兩個(gè)list，一個(gè)保留正序的，一個(gè)保留倒序的，然后分別對(duì)這兩個(gè)list進(jìn)行文件寫入操作。但是如果數(shù)據(jù)量很多在內(nèi)存有限的時(shí)候，只能滿足不了兩個(gè)list的情況下怎么實(shí)現(xiàn)呢？

我的實(shí)現(xiàn)方式就是開啟兩個(gè)進(jìn)程，一個(gè)進(jìn)程保留一個(gè)正序list，寫入文件的同時(shí)對(duì)每個(gè)元素(每通)對(duì)話進(jìn)行倒序，然后把倒序后的數(shù)據(jù)通過(guò)Queue或者Pipe傳入到另外一個(gè)進(jìn)程，讓另外的進(jìn)程進(jìn)行寫文件操作。

defsave_mmi_train_data(queue):

withopen('../data/finetune_mmi_data/train.txt','w',encoding='utf-8')asf:

whileTrue:

save_list=queue.get()

iflen(save_list)==0:

break

forlineinsave_list:

f.write(line)

defsave_mmi_val_data(queue):

withopen('../data/finetune_mmi_data/val.txt','w',encoding='utf-8')asf:

whileTrue:

save_list=queue.get()

iflen(save_list)==0:

break

forlineinsave_list:

f.write(line)

defget_funtine_data(paths):

all_groups=[]

forpathintqdm(paths,desc='loaddatafromexcle'):

df=pd.read_excel(path)

df.dropna(inplace=True)

df.drop_duplicates(inplace=True,keep='first')

groups=list(df.groupby(by=['坐席id','客戶微信id']))

all_groups.extend(groups)

print('len(all_groups)',len(all_groups))

train,val=train_test_split(all_groups,test_size=10000/len(all_groups),random_state=1)

print('len(train)',len(train))

print('len(val)',len(val))

train_std_path='../data/finetune_std_data/train.txt'

val_std_path='../data/finetune_std_data/val.txt'

train_mmi_queue=Queue()

save_funtine_data(train,train_std_path,train_mmi_queue,save_mmi_train_data)

val_mmi_queue=Queue()

save_funtine_data(val,val_std_path,val_mmi_queue,save_mmi_val_data)

defsave_funtine_data(groups,save_std_path,queue,fun):

p=Process(target=fun,args=(queue,))

p.start()

withopen(save_std_path,'w',encoding='utf-8')asf:

forgroupintqdm(groups,desc='findandsavefuntinedialoguedatas'):

new_df=group[1]

df_roles=new_df['是否客服'].values.tolist()

df_contents=new_df['消息內(nèi)容'].values.tolist()

roles=[]

contents=[]

forrole,contentinzip(df_roles,df_contents):

content=content.replace('\n','')

content=emoji.replace_emoji(content,'')

iflen(content)0andcontent!="":

roles.append(role)

contents.append(content)

save_list=[]

save_str=""

forindex,roleinenumerate(roles):

content=contents[index].replace('\n','')

content=emoji.replace_emoji(content,'')

ifcontent[-1]notinpunctuations:

content+='；'

ifindex==0:

ifrole=="是":

save_str+="坐席："+content

else:

save_str+="客戶："+content

else:

ifrole!=roles[index-1]:

f.write(save_str[0:-1]+'\n')

save_list.append(save_str[0:-1]+'\n')

ifrole=="是":

save_str="坐席："+content

else:

save_str="客戶："+content

else:

save_str+=content

iflen(save_str)1:

save_list.append(save_str[0:-1]+'\n')

f.write(save_str[0:-1]+'\n')

f.write('\n')

#切片反轉(zhuǎn)

save_list=save_list[::-1]

save_list.append('\n')

iflen(save_list)0:

queue.put(save_list)

#注意傳入一個(gè)空值，讓倒序進(jìn)程結(jié)束

queue.put([])

p.join()

要注意的是，倒序進(jìn)程中使用whileTrue無(wú)限循環(huán)，需要傳入一個(gè)空值，能夠讓它在正序進(jìn)程結(jié)束的同時(shí)知道數(shù)據(jù)寫完了，跳出循環(huán)。以上代碼比較簡(jiǎn)單就不一一說(shuō)明了。

3、多進(jìn)程及通信實(shí)現(xiàn)異步任務(wù)需求

場(chǎng)景描述：假定一個(gè)模型推理系統(tǒng)，網(wǎng)絡(luò)模塊負(fù)責(zé)接受請(qǐng)求傳輸?shù)臄?shù)據(jù)，把數(shù)據(jù)傳輸給數(shù)據(jù)處理模塊；數(shù)據(jù)處理模塊負(fù)責(zé)處理數(shù)據(jù)(比如說(shuō)語(yǔ)音流或者視頻流等，這些數(shù)據(jù)處理對(duì)CPU的消耗很大)，處理完后把數(shù)據(jù)傳輸給模型推理模塊；模型推理模塊負(fù)責(zé)對(duì)數(shù)據(jù)進(jìn)行推理并把結(jié)果返回給網(wǎng)絡(luò)模塊。要求就是網(wǎng)絡(luò)模塊、數(shù)據(jù)處理模塊和模型推理模塊是獨(dú)立的，可以并行的完成自己的任務(wù)，3個(gè)模塊是異步的，其實(shí)可以把這個(gè)系統(tǒng)簡(jiǎn)化的使用多進(jìn)程來(lái)實(shí)現(xiàn)。

每個(gè)模塊可以用一個(gè)進(jìn)程來(lái)表示，內(nèi)部的邏輯可以開啟子進(jìn)程來(lái)實(shí)現(xiàn)，然后模塊直接的數(shù)據(jù)傳輸就可以使用多進(jìn)程的通信來(lái)實(shí)現(xiàn)，同時(shí)也創(chuàng)建一個(gè)全局的Queue變量，讓每個(gè)模塊的進(jìn)程按需使用。

畫了一個(gè)簡(jiǎn)單的結(jié)構(gòu)和流程圖，如下：

注意的是模塊之間的數(shù)據(jù)傳輸，使用queue傳輸?shù)臅r(shí)候，數(shù)據(jù)量越小，效率越高，所以可以在網(wǎng)絡(luò)模塊這端提前把數(shù)據(jù)進(jìn)行處理。

函數(shù)入口文件

importa

importb

importc

fromwhole_queueimportWholeQueue

importos

if__name__=='__main__':

print("mainprocess:",os.getpid())

whole_queue=WholeQueue()

b_pool_size=2

c_pool_size=6

Module_list=[

a.A(whole_queue,b_pool_size),

b.B(whole_queue,b_pool_size,c_pool_size),

c.C(whole_queue,c_pool_size)

forpinModule_list:

p.start()

公共隊(duì)列類

classWholeQueue():

def__init__(self):

self.queues=dict()

defregister(self,queuename,queue):

self.queues[queuename]=queue

各個(gè)子模塊類

a

frommultiprocessingimportProcess,Queue

importtime

importrandom

importos

classA(Process):

def__init__(self,whole_queue,b_pool_size):

super().__init__(target=self.do_run)

self.whole_queue=whole_queue

self.b_pool_size=b_pool_size

self.queue_list=[]

queue=Queue()

self.whole_queue.register('A',queue)

self.queue_list.append(queue)

self.count=0

defdo_run(self):

print("A.do_runprocess:",os.getpid())

a_send_pro=Process(target=self.send)

a_send_pro.start()

a_receive_pro=Process(target=self.receive)

a_receive_pro.start()

defsend(self):

print("A.sendprocess:",os.getpid())

whileTrue:

time.sleep(0.001)

self.whole_queue.queues['B_%d'%(self.count%self.b_pool_size)].put_nowait(self.count)

self.count+=1

defreceive(self):

print("A.receiveprocess:",os.getpid())

whileTrue:

rece=self.whole_queue.queues['A'].get()

print(rece)

b

frommultiprocessingimportProcess,Queue

importtime

importrandom

importos

classB(Process):

def__init__(self,whole_queue,b_pool_size,c_pool_size):

super().__init__(target=self.do_run)

self.whole_queue=whole_queue

self.b_pool_size=b_pool_size

self.c_pool_size=c_pool_size

self.queue_list=[]

foriinrange(self.b_pool_size):

queue=Queue()

self.whole_queue.register('B_%d'%i,queue)

self.queue_list.append(queue)

self.count=0

defdo_run(self):

print("B.do_runprocess:",os.getpid())

foriinrange(self.b_pool_size):

p=Process(target=ponent,args=(self.queue_list[i],))

p.start()

defcomponent(self,queue):

print("B.componentprocess:",os.getpid())

whileTrue:

time.sleep(0.01)

info=queue.get()

componext_info='component_'+str(info)

self.whole_queue.queues['C_%d'%(info%self.c_pool_size)].put(componext_info)

c

frommultiprocessingimportProcess,Queue

frommodelimportModel

importtime

importrandom

importos

classC(Process):

def__init__(self,whole_queue,c_pool_size):

super().__init__(target=self.do_run)

self.whole_queue=whole_queue

self.c_pool_size=c_pool_size

self.queue_list=[]

foriinrange(self.c_pool_size):

queue=Queue()

self.whole_queue.register('C_%d'%i,queue)

self.queue_list.append(queue)

#self.cache_queue=None

#self.result_queue=None

#self.infer_queue=None

defdo_run(self):

cache_queue=Queue()

result_queue=Queue()

infer_queue=Queue()

print("C.do_runprocess:",os.getpid())

foriinra