眾所周知，Python的并行處理能力很不理想。我認(rèn)為如果不考慮線程和GIL的標(biāo)準(zhǔn)參數(shù)（它們大多是合法的），其原因 不是因?yàn)榧夹g(shù)不到位而是我們的使用方法不恰當(dāng)。大多數(shù)關(guān)于Python線程和多進(jìn)程的教材雖然都很出色，但是內(nèi)容繁瑣冗長(zhǎng)。它們的確在開篇鋪陳了許多有用 的信息，但往往都不會(huì)涉及真正能提高日常工作的部分。

經(jīng)典例子

DDG上以“Python threading tutorial （Python線程教程）”為關(guān)鍵字的熱門搜索結(jié)果表明：幾乎每篇文章中給出的例子都是相同的類+隊(duì)列。

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

#Example.py 
''''' 
Standard Producer/Consumer Threading Pattern 
''' 
  
import time 
import threading 
import Queue 
  
class Consumer(threading.Thread): 
    def __init__(self, queue): 
        threading.Thread.__init__(self) 
        self._queue = queue 
  
    def run(self): 
        while True: 
            # queue.get() blocks the current thread until 
            # an item is retrieved. 
            msg = self._queue.get() 
            # Checks if the current message is 
            # the "Poison Pill" 
            if isinstance(msg, str) and msg == '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!' 
  
def Producer(): 
    # 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.. 
    while time.time() - start_time < 5: 
        # "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() 

唔…….感覺有點(diǎn)像Java。

我現(xiàn)在并不想說明使用Producer / Consume來解決線程/多進(jìn)程的方法是錯(cuò)誤的——因?yàn)樗隙ㄕ_，而且在很多情況下它是最佳方法。但我不認(rèn)為這是平時(shí)寫代碼的最佳選擇。

它的問題所在（個(gè)人觀點(diǎn)）

首先，你需要?jiǎng)?chuàng)建一個(gè)樣板式的鋪墊類。然后，你再創(chuàng)建一個(gè)隊(duì)列，通過其傳遞對(duì)象和監(jiān)管隊(duì)列的兩端來完成任務(wù)。（如果你想實(shí)現(xiàn)數(shù)據(jù)的交換或存儲(chǔ)，通常還涉及另一個(gè)隊(duì)列的參與）。

Worker越多，問題越多。

接下來，你應(yīng)該會(huì)創(chuàng)建一個(gè)worker類的pool來提高Python的速度。下面是IBM tutorial給出的較好的方法。這也是程序員們?cè)诶枚嗑€程檢索web頁面時(shí)的常用方法。

#Example2.py 
''''' 
A more realistic thread pool example 
''' 
  
import time 
import threading 
import Queue 
import urllib2 
  
class Consumer(threading.Thread): 
    def __init__(self, queue): 
        threading.Thread.__init__(self) 
        self._queue = queue 
  
    def run(self): 
        while True: 
            content = self._queue.get() 
            if isinstance(content, str) and content == 'quit': 
                break 
            response = urllib2.urlopen(content) 
        print 'Bye byes!' 
  
def Producer(): 
    urls = [ 
        'http://www.python.org', 'http://www.yahoo.com' 
        'http://www.scala.org', 'http://www.google.com' 
        # etc.. 
    ] 
    queue = Queue.Queue() 
    worker_threads = build_worker_pool(queue, 4) 
    start_time = time.time() 
  
    # Add the urls to process 
    for url in urls: 
        queue.put(url)  
    # Add the poison pillv 
    for worker in worker_threads: 
        queue.put('quit') 
    for worker in worker_threads: 
        worker.join() 
  
    print 'Done! Time taken: {}'.format(time.time() - start_time) 
  
def build_worker_pool(queue, size): 
    workers = [] 
    for _ in range(size): 
        worker = Consumer(queue) 
        worker.start() 
        workers.append(worker) 
    return workers 
  
if __name__ == '__main__': 
    Producer() 

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

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

#p#

介紹：Map

Map是一個(gè)很棒的小功能，同時(shí)它也是Python并行代碼快速運(yùn)行的關(guān)鍵。給不熟悉的人講解一下吧，map是從函數(shù)語言Lisp來的。map函數(shù)能夠按序映射出另一個(gè)函數(shù)。例如

urls = ['http://www.yahoo.com', 'http://www.reddit.com'] 
results = map(urllib2.urlopen, urls) 

這里調(diào)用urlopen方法來把調(diào)用結(jié)果全部按序返回并存儲(chǔ)到一個(gè)列表里。就像：

results = [] 
for url in urls: 
    results.append(urllib2.urlopen(url)) 

Map按序處理這些迭代。調(diào)用這個(gè)函數(shù)，它就會(huì)返回給我們一個(gè)按序存儲(chǔ)著結(jié)果的簡(jiǎn)易列表。

為什么它這么厲害呢？因?yàn)橹灰辛撕线m的庫，map能使并行運(yùn)行得十分流暢！

有兩個(gè)能夠支持通過map函數(shù)來完成并行的庫：一個(gè)是multiprocessing，另一個(gè)是鮮為人知但功能強(qiáng)大的子文件：multiprocessing.dummy。

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

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

準(zhǔn)備開始

要做到通過map函數(shù)來完成并行，你應(yīng)該先導(dǎo)入裝有它們的模塊：

from multiprocessing import Pool 
from multiprocessing.dummy import Pool as ThreadPool 

再初始化：

pool = ThreadPool() 

這簡(jiǎn)單的一句就能代替我們的build_worker_pool 函數(shù)在example2.py中的所有工作。換句話說，它創(chuàng)建了許多有效的worker，啟動(dòng)它們來為接下來的工作做準(zhǔn)備，以及把它們存儲(chǔ)在不同的位置，方便使用。

Pool對(duì)象需要一些參數(shù)，但最重要的是：進(jìn)程。它決定pool中的worker數(shù)量。如果你不填的話，它就會(huì)默認(rèn)為你電腦的內(nèi)核數(shù)值。

如果你在CPU模式下使用多進(jìn)程pool，通常內(nèi)核數(shù)越大速度就越快（還有很多其它因素）。但是，當(dāng)進(jìn)行線程或者處理網(wǎng)絡(luò)綁定之類的工作時(shí)，情況會(huì)比較復(fù)雜所以應(yīng)該使用pool的準(zhǔn)確大小。

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

如果你運(yùn)行過多線程，多線程間的切換將會(huì)浪費(fèi)許多時(shí)間，所以你最好耐心調(diào)試出最適合的任務(wù)數(shù)。

我們現(xiàn)在已經(jīng)創(chuàng)建了pool對(duì)象，馬上就能有簡(jiǎn)單的并行程序了，所以讓我們重新寫example2.py中的url opener吧！

import urllib2 
from multiprocessing.dummy import Pool as ThreadPool 
  
urls = [ 
    'http://www.python.org', 
    '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/', 
    'http://planet.python.org/', 
    '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句還是簡(jiǎn)單的固定寫法。調(diào)用map就能完成我們前面例子中40行的內(nèi)容！為了更形象地表明兩種方法的差異，我還分別給它們運(yùn)行的時(shí)間計(jì)時(shí)。

# results = [] 
# for url in urls: 
#   result = urllib2.urlopen(url) 
#   results.append(result) 
  
# # ------- VERSUS ------- # 
  
# # ------- 4 Pool ------- # 
# pool = ThreadPool(4) 
# results = pool.map(urllib2.urlopen, urls) 
  
# # ------- 8 Pool ------- # 
  
# pool = ThreadPool(8) 
# results = pool.map(urllib2.urlopen, urls) 
  
# # ------- 13 Pool ------- # 
  
# pool = ThreadPool(13) 
# results = pool.map(urllib2.urlopen, urls) 

#p#

結(jié)果：

#                       Single thread:  14.4 Seconds 
#                              4 Pool:   3.1 Seconds 
#                              8 Pool:   1.4 Seconds 
#                             13 Pool:   1.3 Seconds 

相當(dāng)出色！并且也表明了為什么要細(xì)心調(diào)試pool的大小。在這里，只要大于9，就能使其運(yùn)行速度加快。

實(shí)例2：

生成成千上萬的縮略圖

我們?cè)贑PU模式下來完成吧！我工作中就經(jīng)常需要處理大量的圖像文件夾。其任務(wù)之一就是創(chuàng)建縮略圖。這在并行任務(wù)中已經(jīng)有很成熟的方法了。

基礎(chǔ)的單線程創(chuàng)建

import os 
import PIL 
  
from multiprocessing import Pool 
from PIL import Image 
  
SIZE = (75,75) 
SAVE_DIRECTORY = 'thumbs' 
  
def get_image_paths(folder): 
    return (os.path.join(folder, f) 
            for f in os.listdir(folder) 
            if 'jpeg' in f) 
  
def create_thumbnail(filename): 
    im = Image.open(filename) 
    im.thumbnail(SIZE, Image.ANTIALIAS) 
    base, fname = os.path.split(filename) 
    save_path = os.path.join(base, SAVE_DIRECTORY, fname) 
    im.save(save_path) 
  
if __name__ == '__main__': 
    folder = os.path.abspath( 
        '11_18_2013_R000_IQM_Big_Sur_Mon__e10d1958e7b766c3e840') 
    os.mkdir(os.path.join(folder, SAVE_DIRECTORY)) 
  
    images = get_image_paths(folder) 
  
    for image in images: 
             create_thumbnail(Image) 

對(duì)于一個(gè)例子來說，這是有點(diǎn)難，但本質(zhì)上，這就是向程序傳遞一個(gè)文件夾，然后將其中的所有圖片抓取出來，并最終在它們各自的目錄下創(chuàng)建和儲(chǔ)存縮略圖。

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

如果我們用并行調(diào)用map來代替for循環(huán)的話：

import os 
import PIL 
  
from multiprocessing import Pool 
from PIL import Image 
  
SIZE = (75,75) 
SAVE_DIRECTORY = 'thumbs' 
  
def get_image_paths(folder): 
    return (os.path.join(folder, f) 
            for f in os.listdir(folder) 
            if 'jpeg' in f) 
  
def create_thumbnail(filename): 
    im = Image.open(filename) 
    im.thumbnail(SIZE, Image.ANTIALIAS) 
    base, fname = os.path.split(filename) 
    save_path = os.path.join(base, SAVE_DIRECTORY, fname) 
    im.save(save_path) 
  
if __name__ == '__main__': 
    folder = os.path.abspath( 
        '11_18_2013_R000_IQM_Big_Sur_Mon__e10d1958e7b766c3e840') 
    os.mkdir(os.path.join(folder, SAVE_DIRECTORY)) 
  
    images = get_image_paths(folder) 
  
    pool = Pool() 
        pool.map(create_thumbnail,images) 
        pool.close() 
        pool.join() 

5.6秒！

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

好了，文章結(jié)束了。一行完成并行任務(wù)。

原文鏈接：https://medium.com/building-things-on-the-internet/40e9b2b36148

譯文鏈接：http://blog.jobbole.com/58700/