一行代碼完成並行任務

時間 2019-11-09

標籤一行代碼完成並行任務简体版

原文原文鏈接

衆所周知，Python的並行處理能力很不理想。我認爲若是不考慮線程和GIL的標準參數（它們大可能是合法的），其緣由不是由於技術不到位，而是咱們的使用方法不恰當。大多數關於Python線程和多進程的教材雖然都很出色，可是內容繁瑣冗長。它們的確在開篇鋪陳了許多有用信息，但每每都不會涉及真正能提升平常工做的部分。html

經典例子

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

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

#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()

唔…….感受有點像Java。web

我如今並不想說明使用Producer / Consume來解決線程/多進程的方法是錯誤的——由於它確定正確，並且在不少狀況下它是最佳方法。但我不認爲這是平時寫代碼的最佳選擇。segmentfault

它的問題所在（我的觀點）

首先，你須要建立一個樣板式的鋪墊類。而後，你再建立一個隊列，經過其傳遞對象和監管隊列的兩端來完成任務。（若是你想實現數據的交換或存儲，一般還涉及另外一個隊列的參與）。網絡

Worker越多，問題越多。

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

#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()

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

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

介紹：Map

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

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

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

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

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

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

有兩個可以支持經過map函數來完成並行的庫：一個是multiprocessing，另外一個是不爲人知但功能強大的子文件：multiprocessing.dummy。

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

Dummy就是多進程模塊的克隆文件。惟一不一樣的是，多進程模塊使用的是進程，而dummy則使用線程（固然，它有全部Python常見的限制）。也就是說，數據由一個傳遞給另外一個。這可以使得數據輕鬆的在這兩個之間進行前進和回躍，特別是對於探索性程序來講十分有用，由於你不用肯定框架調用究竟是IO 仍是CPU模式。

準備開始

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

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

再初始化:

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吧！

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句仍是簡單的固定寫法。調用map就能完成咱們前面例子中40行的內容！爲了更形象地代表兩種方法的差別，我還分別給它們運行的時間計時。

# 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)

結果：

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

至關出色！而且也代表了爲何要細心調試pool的大小。在這裏，只要大於9，就能使其運行速度加快。

實例2：

生成成千上萬的縮略圖

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

基礎的單線程建立

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)

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

個人電腦處理大約6000張圖片用了27.9秒。

若是咱們用並行調用map來代替for循環的話：

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秒！

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

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

原文：Parallelism in one line
轉載自：伯樂在線 - colleen__chen