一行代碼完成並行任務

編譯：伯樂在線專欄作者 - colleen__chen

http://blog.jobbole.com/58700/

眾所周知，Python的並行處理能力很不理想。我認為如果不考慮線程和GIL的標準參數（它們大多是合法的），其原因不是因為技術不到位，而是我們的使用方法不恰當。大多數關於Python線程和多進程的教材雖然都很出色，但是內容繁瑣冗長。它們的確在開篇鋪陳了許多有用信息，但往往都不會涉及真正能提高日常工作的部分。

經典例子

DDG上以「Python threading tutorial （Python線程教程）」為關鍵字的熱門搜索結果表明：幾乎每篇文章中給出的例子都是相同的類+隊列。

事實上，它們就是以下這段使用producer/Consumer來處理線程/多進程的代碼示例：

#Example.py

Standard Producer/Consumer Threading Pattern

importtime

importthreading

importQueue

classConsumer(threading.Thread):

def__init__(self,queue):

threading.Thread.__init__(self)

self._queue=queue

defrun(self):

whileTrue:

# queue.get() blocks the current thread until

# an item is retrieved.

msg=self._queue.get()

# Checks if the current message is

# the "Poison Pill"

ifisinstance(msg,str)andmsg== quit :

# if so, exists the loop

break

# "Processes" (or in our case, prints) the queue item

print"I m a thread, and I received %s!!"%msg

# Always be friendly!

print Bye byes!

defProducer():

# Queue is used to share items between

# the threads.

queue=Queue.Queue()

# Create an instance of the worker

worker=Consumer(queue)

# start calls the internal run() method to

# kick off the thread

worker.start()

# variable to keep track of when we started

start_time=time.time()

# While under 5 seconds..

whiletime.time()-start_time

# "Produce" a piece of work and stick it in

# the queue for the Consumer to process

queue.put( something at %s %time.time())

# Sleep a bit just to avoid an absurd number of messages

time.sleep(1)

# This the "poison pill" method of killing a thread.

queue.put( quit )

# wait for the thread to close down

worker.join()

if__name__== __main__ :

Producer()

唔…….感覺有點像Java。

我現在並不想說明使用Producer / Consume來解決線程/多進程的方法是錯誤的——因為它肯定正確，而且在很多情況下它是最佳方法。但我不認為這是平時寫代碼的最佳選擇。

它的問題所在（個人觀點）

首先，你需要創建一個樣板式的鋪墊類。然後，你再創建一個隊列，通過其傳遞對象和監管隊列的兩端來完成任務。（如果你想實現數據的交換或存儲，通常還涉及另一個隊列的參與）。

Worker越多，問題越多。

接下來，你應該會創建一個worker類的pool來提高Python的速度。下面是IBM tutorial給出的較好的方法。這也是程序員們在利用多線程檢索web頁面時的常用方法。

#Example2.py

"""

Amore realistic thread poolexample

"""

import time

import threading

import Queue

import urllib2

classConsumer(threading.Thread):

def __init__(self,queue):

threading.Thread.__init__(self)

self._queue=queue

def run(self):

whileTrue:

content=self._queue.get()

ifisinstance(content,str)andcontent== "quit":

break

response=urllib2.urlopen(content)

print"Bye byes!"

def Producer():

urls=[

# etc..

]

queue=Queue.Queue()

worker_threads=build_worker_pool(queue,4)

start_time=time.time()

# Add the urls to process

forurlinurls:

queue.put(url)

# Add the poison pillv

forworkerinworker_threads:

queue.put("quit")

forworkerinworker_threads:

worker.join()

print"Done! Time taken: {}".format(time.time() - start_time)

def build_worker_pool(queue,size):

workers=[]

for_inrange(size):

worker=Consumer(queue)

worker.start()

workers.append(worker)

returnworkers

if__name__== "__main__":

Producer()

它的確能運行，但是這些代碼多麼複雜阿！它包括了初始化方法、線程跟蹤列表以及和我一樣容易在死鎖問題上出錯的人的噩夢——大量的join語句。而這些還僅僅只是繁瑣的開始！

我們目前為止都完成了什麼？基本上什麼都沒有。上面的代碼幾乎一直都只是在進行傳遞。這是很基礎的方法，很容易出錯（該死，我剛才忘了在隊列對象上還需要調用task_done()方法（但是我懶得修改了）），性價比很低。還好，我們還有更好的方法。

介紹：Map

Map是一個很棒的小功能，同時它也是Python並行代碼快速運行的關鍵。給不熟悉的人講解一下吧，map是從函數語言Lisp來的。map函數能夠按序映射出另一個函數。例如

results=map(urllib2.urlopen,urls)

這裡調用urlopen方法來把調用結果全部按序返回並存儲到一個列表裡。就像：

results=[]

forurlinurls:

results.append(urllib2.urlopen(url))

Map按序處理這些迭代。調用這個函數，它就會返回給我們一個按序存儲著結果的簡易列表。

為什麼它這麼厲害呢？因為只要有了合適的庫，map能使並行運行得十分流暢！

有兩個能夠支持通過map函數來完成並行的庫：一個是multiprocessing，另一個是鮮為人知但功能強大的子文件：multiprocessing.dummy。

題外話：這個是什麼？你從來沒聽說過dummy多進程庫？我也是最近才知道的。它在多進程的說明文檔裡面僅僅只被提到了一句。而且那一句就是大概讓你知道有這麼個東西的存在。我敢說，這樣幾近拋售的做法造成的後果是不堪設想的！

Dummy就是多進程模塊的克隆文件。唯一不同的是，多進程模塊使用的是進程，而dummy則使用線程（當然，它有所有Python常見的限制）。也就是說，數據由一個傳遞給另一個。這能夠使得數據輕鬆的在這兩個之間進行前進和回躍，特別是對於探索性程序來說十分有用，因為你不用確定框架調用到底是IO 還是CPU模式。

準備開始

要做到通過map函數來完成並行，你應該先導入裝有它們的模塊：

frommultiprocessingimportPool

frommultiprocessing.dummyimportPoolasThreadPool

再初始化:

pool = ThreadPool()

這簡單的一句就能代替我們的build_worker_pool 函數在example2.py中的所有工作。換句話說，它創建了許多有效的worker，啟動它們來為接下來的工作做準備，以及把它們存儲在不同的位置，方便使用。

Pool對象需要一些參數，但最重要的是：進程。它決定pool中的worker數量。如果你不填的話，它就會默認為你電腦的內核數值。

如果你在CPU模式下使用多進程pool，通常內核數越大速度就越快（還有很多其它因素）。但是，當進行線程或者處理網路綁定之類的工作時，情況會比較複雜所以應該使用pool的準確大小。

pool = ThreadPool(4) # Sets the pool size to 4

如果你運行過多線程，多線程間的切換將會浪費許多時間，所以你最好耐心調試出最適合的任務數。

我們現在已經創建了pool對象，馬上就能有簡單的並行程序了，所以讓我們重新寫example2.py中的url opener吧！

importurllib2

frommultiprocessing.dummyimportPoolasThreadPool

urls=[

http://www.python.org/about/ ,

http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html ,

http://www.python.org/doc/ ,

http://www.python.org/download/ ,

http://www.python.org/getit/ ,

http://www.python.org/community/ ,

https://wiki.python.org/moin/ ,

https://wiki.python.org/moin/LocalUserGroups ,

http://www.python.org/psf/ ,

http://docs.python.org/devguide/ ,

http://www.python.org/community/awards/

# etc..

]

# Make the Pool of workers

pool=ThreadPool(4)

# Open the urls in their own threads

# and return the results

results=pool.map(urllib2.urlopen,urls)

#close the pool and wait for the work to finish

pool.close()

pool.join()

看吧！這次的代碼僅用了4行就完成了所有的工作。其中3句還是簡單的固定寫法。調用map就能完成我們前面例子中40行的內容！為了更形象地表明兩種方法的差異，我還分別給它們運行的時間計時。

結果：

相當出色！並且也表明了為什麼要細心調試pool的大小。在這裡，只要大於9，就能使其運行速度加快。

實例2：

生成成千上萬的縮略圖

我們在CPU模式下來完成吧！我工作中就經常需要處理大量的圖像文件夾。其任務之一就是創建縮略圖。這在並行任務中已經有很成熟的方法了。

基礎的單線程創建

importos

importPIL

frommultiprocessingimportPool

fromPILimportImage

SIZE=(75,75)

SAVE_DIRECTORY= thumbs

defget_image_paths(folder):

forfinos.listdir(folder)

if jpeg inf)

defcreate_thumbnail(filename):

im=Image.open(filename)

im.thumbnail(SIZE,Image.ANTIALIAS)

im.save(save_path)

if__name__== __main__ :

images=get_image_paths(folder)

forimageinimages:

create_thumbnail(Image)

對於一個例子來說，這是有點難，但本質上，這就是向程序傳遞一個文件夾，然後將其中的所有圖片抓取出來，並最終在它們各自的目錄下創建和儲存縮略圖。

我的電腦處理大約6000張圖片用了27.9秒。

如果我們用並行調用map來代替for循環的話：

importos

importPIL

frommultiprocessingimportPool

fromPILimportImage

SIZE=(75,75)

SAVE_DIRECTORY= thumbs

defget_image_paths(folder):

forfinos.listdir(folder)

if jpeg inf)

defcreate_thumbnail(filename):

im=Image.open(filename)

im.thumbnail(SIZE,Image.ANTIALIAS)

im.save(save_path)

if__name__== __main__ :

images=get_image_paths(folder)

pool=Pool()

pool.map(create_thumbnail,images)

pool.close()

pool.join()

5.6秒！

對於只改變了幾行代碼而言，這是大大地提升了運行速度。這個方法還能更快，只要你將cpu 和 io的任務分別用它們的進程和線程來運行——但也常造成死鎖。總之，綜合考慮到 map這個實用的功能，以及人為線程管理的缺失，我覺得這是一個美觀，可靠還容易debug的方法。

好了，文章結束了。一行完成並行任務。

最近熱文：

1、邀你參加11期 程序員專場相親活動

2、PHP高工/架構師，18-50K，深圳南山

3、JAVA/Web前端，深圳前海，15-30K

4、如何判斷是否到了該辭職的時候？

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！

本站內容充實豐富，博大精深，小編精選每日熱門資訊，隨時更新，點擊「搶先收到最新資訊」瀏覽吧！

請您繼續閱讀更多來自 程序源 的精彩文章: