最近接觸一個項目，要在多個虛擬機中運行任務，參考別人之前項目的代碼，采用了多進程來處理，于是上網查了查python中的多進程
一、先說說Queue（隊列對象）
Queue是python中的標準庫，可以直接import 引用，之前學習的時候有聽過著名的“先吃先拉”與“后吃先吐”，其實就是這里說的隊列，隊列的構造的時候可以定義它的容量，別吃撐了，吃多了，就會報錯,構造的時候不寫或者寫個小于1的數則表示無限多
import Queue
q = Queue.Queue(10)
向隊列中放值(put)
q.put(‘yang')
q.put(4)
q.put([‘yan','xing'])
在隊列中取值get()
默認的隊列是先進先出的
>>> q.get()
‘yang'
>>> q.get()
4
>>> q.get()
[‘yan', ‘xing']
當一個隊列為空的時候如果再用get取則會堵塞，所以取隊列的時候一般是用到
get_nowait()方法，這種方法在向一個空隊列取值的時候會拋一個Empty異常
所以更常用的方法是先判斷一個隊列是否為空，如果不為空則取值
隊列中常用的方法
Queue.qsize() 返回隊列的大小
Queue.empty() 如果隊列為空，返回True,反之False
Queue.full() 如果隊列滿了，返回True,反之False
Queue.get([block[, timeout]]) 獲取隊列，timeout等待時間
Queue.get_nowait() 相當Queue.get(False)
非阻塞 Queue.put(item) 寫入隊列，timeout等待時間
Queue.put_nowait(item) 相當Queue.put(item, False)
二、multiprocessing中使用子進程概念
from multiprocessing import Process
可以通過Process來構造一個子進程
p = Process(target=fun,args=(args))
再通過p.start()來啟動子進程
再通過p.join()方法來使得子進程運行結束后再執行父進程
from multiprocessing import Processimport os # 子進程要執行的代碼def run_proc(name): print 'Run child process %s (%s)...' % (name, os.getpid()) if __name__=='__main__': print 'Parent process %s.' % os.getpid() p = Process(target=run_proc, args=('test',)) print 'Process will start.' p.start() p.join() print 'Process end.' 
三、在multiprocessing中使用pool
如果需要多個子進程時可以考慮使用進程池(pool)來管理
from multiprocessing import Pool
from multiprocessing import Poolimport os, time def long_time_task(name): print 'Run task %s (%s)...' % (name, os.getpid()) start = time.time() time.sleep(3) end = time.time() print 'Task %s runs %0.2f seconds.' % (name, (end - start)) if __name__=='__main__': print 'Parent process %s.' % os.getpid() p = Pool() for i in range(5): p.apply_async(long_time_task, args=(i,)) print 'Waiting for all subprocesses done...' p.close() p.join() print 'All subprocesses done.'
pool創建子進程的方法與Process不同，是通過
p.apply_async(func,args=(args))實現，一個池子里能同時運行的任務是取決你電腦的cpu數量，如我的電腦現在是有4個cpu，那會子進程task0,task1,task2,task3可以同時啟動，task4則在之前的一個某個進程結束后才開始
上面的程序運行后的結果其實是按照上圖中1，2，3分開進行的，先打印1，3秒后打印2，再3秒后打印3
代碼中的p.close()是關掉進程池子，是不再向里面添加進程了，對Pool對象調用join()方法會等待所有子進程執行完畢，調用join()之前必須先調用close()，調用close()之后就不能繼續添加新的Process了。
當時也可以是實例pool的時候給它定義一個進程的多少
如果上面的代碼中p=Pool(5)那么所有的子進程就可以同時進行
三、多個子進程間的通信
多個子進程間的通信就要采用第一步中說到的Queue，比如有以下的需求，一個子進程向隊列中寫數據，另外一個進程從隊列中取數據，
#coding:gbkfrom multiprocessing import Process, Queueimport os, time, random# 寫數據進程執行的代碼:def write(q): for value in ['A', 'B', 'C']: print 'Put %s to queue...' % value q.put(value) time.sleep(random.random())# 讀數據進程執行的代碼:def read(q): while True: if not q.empty(): value = q.get(True) print 'Get %s from queue.' % value time.sleep(random.random()) else: breakif __name__=='__main__': # 父進程創建Queue，并傳給各個子進程： q = Queue() pw = Process(target=write, args=(q,)) pr = Process(target=read, args=(q,)) # 啟動子進程pw，寫入: pw.start() # 等待pw結束: pw.join() # 啟動子進程pr，讀取: pr.start() pr.join() # pr進程里是死循環，無法等待其結束，只能強行終止: print print '所有數據都寫入并且讀完'
四、關于上面代碼的幾個有趣的問題
if __name__=='__main__': # 父進程創建Queue，并傳給各個子進程： q = Queue() p = Pool() pw = p.apply_async(write,args=(q,)) pr = p.apply_async(read,args=(q,)) p.close() p.join() print print '所有數據都寫入并且讀完'
如果main函數寫成上面的樣本，本來我想要的是將會得到一個隊列，將其作為參數傳入進程池子里的每個子進程，但是卻得到
RuntimeError: Queue objects should only be shared between processes through inheritance
的錯誤，查了下，大意是隊列對象不能在父進程與子進程間通信，這個如果想要使用進程池中使用隊列則要使用multiprocess的Manager類
if __name__=='__main__': manager = multiprocessing.Manager() # 父進程創建Queue，并傳給各個子進程： q = manager.Queue() p = Pool() pw = p.apply_async(write,args=(q,)) time.sleep(0.5) pr = p.apply_async(read,args=(q,)) p.close() p.join() print print '所有數據都寫入并且讀完'
這樣這個隊列對象就可以在父進程與子進程間通信，不用池則不需要Manager，以后再擴展multiprocess中的Manager類吧
關于鎖的應用，在不同程序間如果有同時對同一個隊列操作的時候，為了避免錯誤，可以在某個函數操作隊列的時候給它加把鎖，這樣在同一個時間內則只能有一個子進程對隊列進行操作，鎖也要在manager對象中的鎖
#coding:gbk from multiprocessing import Process,Queue,Poolimport multiprocessingimport os, time, random # 寫數據進程執行的代碼:def write(q,lock): lock.acquire() #加上鎖 for value in ['A', 'B', 'C']: print 'Put %s to queue...' % value q.put(value) lock.release() #釋放鎖 # 讀數據進程執行的代碼:def read(q): while True: if not q.empty(): value = q.get(False) print 'Get %s from queue.' % value time.sleep(random.random()) else: break if __name__=='__main__': manager = multiprocessing.Manager() # 父進程創建Queue，并傳給各個子進程： q = manager.Queue() lock = manager.Lock() #初始化一把鎖 p = Pool() pw = p.apply_async(write,args=(q,lock)) pr = p.apply_async(read,args=(q,)) p.close() p.join() print print '所有數據都寫入并且讀完'
新聞熱點
疑難解答
