關於Python並行任務技巧的幾點補完

早上逛微博發現了segmentfault上的這篇文章：關於python並行任務技巧（。看過之後大有裨益。順手試了試後遇到幾個小坑，記錄下來作為補完（作者也有點語焉不詳哦^_^）。

第一點是傳入的function，只能接收乙個傳入引數，一開始以為在傳入的序列裡用tuple可以自動解包成多個引數傳進去，可惜實踐後是不行的：

#coding=utf8
from multiprocessing import pool
def do_add(n1, n2):
return n1+n2
pool = pool(5)
print pool.map(do_add, [(1,2),(3,4),(5,6)])
pool.close()
pool.join()

執行後結果就報錯了：

traceback (most recent call last):
file "mt.py", line 8, in print pool.map(do_add, [(1,2),(3,4),(5,6)])
file "/system/library/frameworks/python.framework/versions/2.7/lib/python2.7/multiprocessing/pool.py", line 250, in map
return self.map_async(func, iterable, chunksize).get()
file "/system/library/frameworks/python.framework/versions/2.7/lib/python2.7/multiprocessing/pool.py", line 554, in get
raise self._value
typeerror: do_add() takes exactly 2 arguments (1 given)

第二是傳入的function如果要做長期執行，比如放乙個死迴圈在裡面長期執行的話，必須處理必要的異常，不然ctrl+c殺不掉程序，比如：

#coding=utf8
from multiprocessing import pool
import time
def do_add(n1):
while true:
time.sleep(1)
print n1
pool = pool(5)
print pool.map(do_add, [1,2,3,4,5,6])
pool.close()
pool.join()

這段**一跑起來是ctrl+c殺不掉的，最後只能把console整個關掉才行。不過這麼寫就ok了：

#coding=utf8
from multiprocessing import pool
import time
def do_add(n1):
try:
while true:
time.sleep(1)
print n1
except:
return n1
pool = pool(5)
print pool.map(do_add, [1,2,3,4,5,6])
pool.close()
pool.join()

補完的補完，有網友提供了解決辦法，使用functools的partial可以解決，詳見爆棧（

第三點是為什麼要在子程序裡用死迴圈讓其長期執行。竊以為作者的直接把上千個任務暴力丟給程序池的做法並不是最高效的方式，即便是正在執行的程序數和cpu數能匹配得切到好處，但是一大堆的程序切換的開銷也會有相當的負擔。但是建立幾個長期執行的工作程序，每個工作程序處理多個任務，省略掉了大量開啟關閉程序的開銷，原理上來說會效率高一些。不過這個問題我沒有實測過。再不過其實從原理上來說這個開銷雖然有但是並不是有多麼大，很多時候完全可以忽略，比如作者用的例子。所以其實更確切一點的需求反而是用於實現生產者消費者模式。因為在作者的例子裡，任務數是固定的，不可控的，更多的時候我們反而是需要用生產者建立任務，由worker程序去執行任務。舉個例子，生產者監聽乙個redis的佇列，有新url放進去的時候就通知worker程序去取。

**如下：

#coding=utf8
from multiprocessing import pool, queue
import redis
import requests
queue = queue(20)
def consumer():
r = redis.redis(host='127.0.0.1',port=6379,db=1)
while true:
k, url = r.blpop(['pool',])
queue.put(url)
def worker():
while true:
url = queue.get()
print requests.get(url).text
def process(ptype):
try:
if ptype:
consumer()
else:
worker()
except:
pass
pool = pool(5)
print pool.map(process, [1,0,0,0,0])
pool.close()
pool.join()

比起經典的方式來說簡單很多，效率高，易懂，而且沒什麼死鎖的陷阱。

關於Python並行任務技巧的幾點補完

Python多工並行併發

python並行任務之生產消費模式

python 模擬器多開任務並行

關於Python並行任務技巧的幾點補完

Python多工 並行 併發

python並行任務之生產消費模式

python 模擬器多開任務並行

相關推薦

Python多工並行併發