python多進(jìn)程及通信實(shí)現(xiàn)異步任務(wù)的方法
寫在最前面，說實(shí)話python多進(jìn)程這塊兒知識對于很少使用python多進(jìn)程或者沒有實(shí)際使用過多python進(jìn)程解決問題的人來說，還是有一定難度的。本人也是很少接觸多進(jìn)程的場景，對于python多進(jìn)程的使用也是比較陌生的。在接觸了一些多進(jìn)程的業(yè)務(wù)場景下，對python多進(jìn)程的使用進(jìn)行了學(xué)習(xí)，覺得很有必要進(jìn)行一個梳理總結(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()啟動一個已經(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ù)邏輯功能；主程序啟動10個進(jìn)程。
from multiprocessing import Process
count = 0
class MultiOneProcess(Process):
def __init__(self,name):
super().__init__()
self.name = name
def run(self) -> None:
global count
count += 1
print('process name %s is running----count:%d'%(self.name, count))
if __name__ == '__main__':
p_list = []
for i in range(10):
name = 'process_%d'%i
p = MultiOneProcess(name = name)
p.start()
p_list.append(p)
for p in p_list:
p.join()
print('this main process')b、使用Process和target方法
定義一個進(jìn)程類繼承Process類，同時在super()初始化中傳入target函數(shù)
from multiprocessing import Process
count = 0
class MultiTwoProcess(Process):
def __init__(self,name):
super().__init__(target=self.do_fun)
self.name = name
def do_fun(self):
global count
count += 1
print('process name %s is running----count:%d' % (name, count))
if __name__ == '__main__':
p_list = []
for i in range(10):
name = 'process_%d'%i
p = MultiTwoProcess(name)
p.start()
p_list.append(p)
for p in p_list:
p.join()
print('this main process')代碼中定義了一個類MultiTwoProcess類，類中定義了do_fun函數(shù)，把它作為參數(shù)傳入到target中。
c、直接使用Process類
傳入target函數(shù)，同時傳入args參數(shù)，注意args參數(shù)是一個元組，切不能省略最后一個逗號
from multiprocessing import Process
count = 0
def do_fun(name):
global count
count += 1
print('process name %s is running----count:%d' % (name, count))
if __name__ == '__main__':
p_list = []
for i in range(10):
name = 'process_%d'%i
p = Process(target=do_fun,args=(name,))
p.start()
p_list.append(p)
for p in p_list:
p.join()
print('this main process')以上三者運(yùn)行的結(jié)果，是一樣的，如下：
2、多進(jìn)程的通信
進(jìn)程之間的通信一般都采用Queue和pipe，區(qū)別是：pipe只能在兩個進(jìn)程之間調(diào)用，而Queue是可以多個進(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的時候，如果Queue已經(jīng)滿了，那么就會跑出Queue.Full異常；
當(dāng)block=True且timeout有正值的時候，Queue已經(jīng)滿了，Queue會阻塞timeout時間，超出時間就會拋出同樣的異常
獲取數(shù)據(jù)
queue.get(block=True, timeout=None)
當(dāng)block=False的時候，如果Queue為空，那么就會跑出Queue.Empty異常；
當(dāng)block=True且timeout有正值的時候，Queue已經(jīng)為空，Queue會阻塞timeout時間，超出時間就會拋出同樣的異常
以上2個API是阻塞；還有兩個非堵塞的API
queue.put(obj, block=False) 和 queue.put_nowait(obj)等效
queue.get(block=False) 和 queue.get_nowait()等效
簡單的實(shí)現(xiàn)，一個進(jìn)程發(fā)送數(shù)據(jù)，另外2個進(jìn)程接收數(shù)據(jù)，就可以使用queue通信
from multiprocessing import Process, Queue
def send(q):
while True:
q.put('發(fā)送一個數(shù)據(jù)')
def receive1(q):
while True:
s = q.get()
print('receive1:', s)
def receive2(q):
while True:
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ù)(有競爭的再取)，所以打印結(jié)果是無序的
b、Pipe
Pipe(duplex=True)返回2個連通端(p1,p2)；當(dāng)duplex=True時，雙向通信，p1發(fā)送，p2接收；p2發(fā)送，p1接收。
當(dāng)duplex=True時，單向通信，p1只能發(fā)送，p2只能接收。
常用API， pipe.send() pipe.recv()
from multiprocessing import Process, Pipe
def fun2(p):
while True:
s = p.recv()
print('接收一個數(shù)據(jù)：',s)
def fun1(p):
while True:
print('發(fā)送一個數(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ù)場景使用多進(jìn)程的方式和復(fù)雜度也不相同，就我遇見過的一些場景進(jìn)行演示和說明。
1、使用進(jìn)程池快速抽取數(shù)據(jù)
場景描述：有1000個Excel文件的數(shù)據(jù)需要進(jìn)行抽取和清洗，要把不符合我們需求的數(shù)據(jù)過濾掉，保留質(zhì)量很高的數(shù)據(jù)；每個Excel都有幾十萬或者上百萬的數(shù)據(jù)，那么怎么快速的完成這個任務(wù)呢？
首先整體上而言，可以把單個Excel的處理并行起來；那么可以使用多進(jìn)程，其次這個需要返回結(jié)果，要保留合格的數(shù)據(jù)，比較簡單的就是采用進(jìn)程池了，它能夠很方便的把進(jìn)程處理的結(jié)果進(jìn)行返回，并且返回的還是一個生成器；如果還需要更快，那么可以把單個Excel中的每條數(shù)據(jù)的處理并行起來。代碼層面上，采用pool進(jìn)程池來完成這個任務(wù)(本文沒有對進(jìn)程池的使用和API做說明)，具體的實(shí)現(xiàn)方式采取pool.imap()
if __name__ == '__main__':
#所有Excel的路徑
all_paths = glob('../data/original_data/*')
sysInfo_list = ['我通過了好友請求，現(xiàn)在你倆可以開始聊天了', '我通過了你的朋友驗(yàn)證請求，現(xiàn)在我們可以開始聊天了', '已通過你的朋友驗(yàn)證請求，現(xiàn)在可以開始聊天了', '不支持此消息，請?jiān)谑謾C(jī)上查看',
'微信紅包']
interval = 25
if len(all_paths)//interval * interval < len(all_paths):
k = len(all_paths)//interval + 1
else:
k = len(all_paths) // interval
#分段處理，每段25個Excel
for i in range(k):
paths = all_paths[i*interval:(i+1)*interval]
if i*interval >= 100 and i*interval < 200:
params = []
for path in tqdm(paths):
params.append((path, sysInfo_list))
#多進(jìn)程處理——進(jìn)程池、以及進(jìn)度顯示
with Pool(20) as p:
res = list(tqdm(p.imap(extract_data, params), total=len(params), desc='extract_data'))
all_df = []
for dfs in res:
if len(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ù)清洗和保存
場景描述：從Excel中讀取數(shù)據(jù)，數(shù)據(jù)格式是整通整通的對話，每通對話有一定的輪數(shù)；保存數(shù)據(jù)到2個txt中，一個是順序保留，一個是倒序保留；整體對話順序不變，每通對話內(nèi)部順序倒序。
正序：
倒序：
要想實(shí)現(xiàn)這樣的任務(wù)，粗暴的做法是，用兩個list，一個保留正序的，一個保留倒序的，然后分別對這兩個list進(jìn)行文件寫入操作。但是如果數(shù)據(jù)量很多在內(nèi)存有限的時候，只能滿足不了兩個list的情況下怎么實(shí)現(xiàn)呢？
我的實(shí)現(xiàn)方式就是開啟兩個進(jìn)程，一個進(jìn)程保留一個正序list，寫入文件的同時對每個元素(每通)對話進(jìn)行倒序，然后把倒序后的數(shù)據(jù)通過Queue或者Pipe傳入到另外一個進(jìn)程，讓另外的進(jìn)程進(jìn)行寫文件操作。
def save_mmi_train_data(queue):
with open('../data/finetune_mmi_data/train.txt','w',encoding='utf-8') as f:
while True:
save_list = queue.get()
if len(save_list) == 0:
break
for line in save_list:
f.write(line)
def save_mmi_val_data(queue):
with open('../data/finetune_mmi_data/val.txt','w',encoding='utf-8') as f:
while True:
save_list = queue.get()
if len(save_list) == 0:
break
for line in save_list:
f.write(line)
def get_funtine_data(paths):
all_groups = []
for path in tqdm(paths,desc='load data from excle'):
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)
def save_funtine_data(groups,save_std_path,queue,fun):
p = Process(target=fun,args=(queue,))
p.start()
with open(save_std_path,'w', encoding='utf-8') as f:
for group in tqdm(groups, desc='find and save funtine dialogue datas'):
new_df = group[1]
df_roles = new_df['是否客服'].values.tolist()
df_contents = new_df['消息內(nèi)容'].values.tolist()
roles = []
contents = []
for role,content in zip(df_roles,df_contents):
content = content.replace('\n', '')
content = emoji.replace_emoji(content, '')
if len(content) > 0 and content != "":
roles.append(role)
contents.append(content)
save_list = []
save_str = ""
for index, role in enumerate(roles):
content = contents[index].replace('\n','')
content = emoji.replace_emoji(content, '')
if content[-1] not in punctuations:
content += '；'
if index == 0:
if role == "是":
save_str += "坐席："+content
else:
save_str += "客戶："+content
else:
if role != roles[index-1]:
f.write(save_str[0:-1]+'\n')
save_list.append(save_str[0:-1]+'\n')
if role == "是":
save_str = "坐席：" + content
else:
save_str = "客戶：" + content
else:
save_str += content
if len(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')
if len(save_list) > 0:
queue.put(save_list)
#注意傳入一個空值，讓倒序進(jìn)程結(jié)束
queue.put([])
p.join()要注意的是，倒序進(jìn)程中使用while True 無限循環(huán)，需要傳入一個空值，能夠讓它在正序進(jìn)程結(jié)束的同時知道數(shù)據(jù)寫完了，跳出循環(huán)。以上代碼比較簡單就不一一說明了。
3、多進(jìn)程及通信實(shí)現(xiàn)異步任務(wù)需求
場景描述：假定一個模型推理系統(tǒng)，網(wǎng)絡(luò)模塊負(fù)責(zé)接受請求傳輸?shù)臄?shù)據(jù)，把數(shù)據(jù)傳輸給數(shù)據(jù)處理模塊；數(shù)據(jù)處理模塊負(fù)責(zé)處理數(shù)據(jù)(比如說語音流或者視頻流等，這些數(shù)據(jù)處理對CPU的消耗很大)，處理完后把數(shù)據(jù)傳輸給模型推理模塊；模型推理模塊負(fù)責(zé)對數(shù)據(jù)進(jìn)行推理并把結(jié)果返回給網(wǎng)絡(luò)模塊。要求就是網(wǎng)絡(luò)模塊、數(shù)據(jù)處理模塊和模型推理模塊是獨(dú)立的，可以并行的完成自己的任務(wù)，3個模塊是異步的，其實(shí)可以把這個系統(tǒng)簡化的使用多進(jìn)程來實(shí)現(xiàn)。
每個模塊可以用一個進(jìn)程來表示，內(nèi)部的邏輯可以開啟子進(jìn)程來實(shí)現(xiàn)，然后模塊直接的數(shù)據(jù)傳輸就可以使用多進(jìn)程的通信來實(shí)現(xiàn)，同時也創(chuàng)建一個全局的Queue變量，讓每個模塊的進(jìn)程按需使用。
畫了一個簡單的結(jié)構(gòu)和流程圖，如下：
注意的是模塊之間的數(shù)據(jù)傳輸，使用queue傳輸?shù)臅r候，數(shù)據(jù)量越小，效率越高，所以可以在網(wǎng)絡(luò)模塊這端提前把數(shù)據(jù)進(jìn)行處理。
函數(shù)入口文件
import a
import b
import c
from whole_queue import WholeQueue
import os
if __name__ == '__main__':
print("main process:",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)
]
for p in Module_list:
p.start()公共隊(duì)列類
class WholeQueue():
def __init__(self):
self.queues = dict()
def register(self,queuename,queue):
self.queues[queuename] = queue各個子模塊類
a
from multiprocessing import Process,Queue
import time
import random
import os
class A(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
def do_run(self):
print("A.do_run process:", os.getpid())
a_send_pro = Process(target = self.send)
a_send_pro.start()
a_receive_pro = Process(target = self.receive)
a_receive_pro.start()
def send(self):
print("A.send process:", os.getpid())
while True:
time.sleep(0.001)
self.whole_queue.queues['B_%d'%(self.count%self.b_pool_size)].put_nowait(self.count)
self.count += 1
def receive(self):
print("A.receive process:", os.getpid())
while True:
rece = self.whole_queue.queues['A'].get()
print(rece)
b
from multiprocessing import Process,Queue
import time
import random
import os
class B(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 = []
for i in range(self.b_pool_size):
queue = Queue()
self.whole_queue.register('B_%d'% i , queue)
self.queue_list.append(queue)
self.count = 0
def do_run(self):
print("B.do_run process:", os.getpid())
for i in range(self.b_pool_size):
p = Process(target=self.component,args=(self.queue_list[i],))
p.start()
def component(self, queue):
print("B.component process:", os.getpid())
while True:
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
from multiprocessing import Process,Queue
from model import Model
import time
import random
import os
class C(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 = []
for i in range(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
def do_run(self):
cache_queue = Queue()
result_queue = Queue()
infer_queue = Queue()
print("C.do_run process:", os.getpid())
for i in range(self.c_pool_size):
p = Process(target=self.receive,args=(self.queue_list[i], cache_queue,))
p.start()
cache_p = Process(target=self.cache,args=(cache_queue, infer_queue,))
cache_p.start()
predict_p = Process(target=self.predict,args=(infer_queue, result_queue))
predict_p.start()
while True:
res = result_queue.get()
for ele in res:
self.whole_queue.queues['A'].put(ele)
def receive(self, queue,cache_queue):
print("C.receive process:", os.getpid())
while True:
info = queue.get()
receive_info = 'receive_' + info
cache_queue.put(receive_info)
def cache(self,cache_queue, infer_queue):
timeLast = time.time()
print("C.cache process:", os.getpid())
caches = []
while True:
data = cache_queue.get()
caches.append(data)
if len(caches) > 128 or time.time() - timeLast > 1:
timeLast = time.time()
infer_queue.put(caches)
caches = []
def predict(self,infer_queue, result_queue):
print("C.predict process:", os.getpid())
# 模型必須在這里初始化
model = Model()
while True:
data = infer_queue.get()
result = model(data)
result = [ 'modelpredict_' + ele for ele in result]
time.sleep(random.uniform(0.1,0.5))
result_queue.put(result)代碼比較好理解，需要注意的是子進(jìn)程在使用變量的時候，例如初始后的模型，應(yīng)該要每一個子進(jìn)程獨(dú)立的進(jìn)行初始化，不然會報(bào)錯，就是C類中模型初始化不能在init中初始后，然后傳入到每個子進(jìn)程中去——而應(yīng)該在每個子進(jìn)程中初始化。
到此這篇關(guān)于python多進(jìn)程及通信實(shí)現(xiàn)異步任務(wù)需求的文章就介紹到這了,更多相關(guān)python異步任務(wù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Python使用captcha制作驗(yàn)證碼的實(shí)現(xiàn)示例
本文主要介紹了Python使用captcha制作驗(yàn)證碼的實(shí)現(xiàn)示例，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2022-08-08
這篇文章主要給大家介紹了關(guān)于Python WSGI的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2018-08-08
批量替換是我們在日常工作中經(jīng)常會遇到的一個問題，下面這篇文章主要給大家介紹了關(guān)于python批量替換多文件字符串問題的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧。2018-04-04
這篇文章主要介紹了Python探索之ModelForm代碼詳解，具有一定參考價(jià)值，需要的朋友可以了解下。2017-10-10
這篇文章主要介紹了python元組的可變與不可變問題，具有很好的參考價(jià)值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-12-12
pytorch 中autograd.grad()函數(shù)的用法說明
這篇文章主要介紹了pytorch 中autograd.grad()函數(shù)的用法說明，具有很好的參考價(jià)值，希望對大家有所幫助。一起跟隨小編過來看看吧2021-05-05
這篇文章主要介紹了怎么樣用Python寫了20行代碼來破解找回已經(jīng)忘記的密碼，密碼的排列組合是有限的，遍歷嘗試就能夠試出正確的密碼，但這個工作要怎么用python實(shí)現(xiàn)，請看下文2021-08-08
python?selenium中Excel數(shù)據(jù)維護(hù)指南
這篇文章主要給大家介紹了關(guān)于python?selenium中Excel數(shù)據(jù)維護(hù)的相關(guān)資料,文中通過實(shí)例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2022-03-03
這篇文章主要介紹了python批量修改文件名的實(shí)現(xiàn)代碼,需要的朋友可以參考下2014-09-09
這篇文章主要介紹了如何利用python處理原始音頻數(shù)據(jù)，文章以audioop的相關(guān)資料展開內(nèi)容，audioop是python標(biāo)準(zhǔn)庫中用于處理原始音頻數(shù)據(jù)的模塊，封裝了一些便捷的編碼轉(zhuǎn)換函數(shù)。下文詳細(xì)內(nèi)容,需要的朋友可以參考以下2022-01-01