使用map函數(shù)實(shí)現(xiàn)Python程序并行化

Python 在程序并行化方面多少有些聲名狼藉。撇開(kāi)技術(shù)上的問(wèn)題，例如線(xiàn)程的實(shí)現(xiàn)和 GIL，我覺(jué)得錯(cuò)誤的教學(xué)指導(dǎo)才是主要問(wèn)題。常見(jiàn)的經(jīng)典 Python 多線(xiàn)程、多進(jìn)程教程多顯得偏"重"。而且往往隔靴搔癢，沒(méi)有深入探討日常工作中最有用的內(nèi)容。

傳統(tǒng)的例子

簡(jiǎn)單搜索下"Python 多線(xiàn)程教程"，不難發(fā)現(xiàn)幾乎所有的教程都給出涉及類(lèi)和隊(duì)列的例子：

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(creat_thumbnail, images)
  pool.close()
  pool.join()

哈，看起來(lái)有些像 Java 不是嗎？

我并不是說(shuō)使用生產(chǎn)者/消費(fèi)者模型處理多線(xiàn)程/多進(jìn)程任務(wù)是錯(cuò)誤的（事實(shí)上，這一模型自有其用武之地）。只是，處理日常腳本任務(wù)時(shí)我們可以使用更有效率的模型。

問(wèn)題在于…

首先，你需要一個(gè)樣板類(lèi)；
其次，你需要一個(gè)隊(duì)列來(lái)傳遞對(duì)象；
而且，你還需要在通道兩端都構(gòu)建相應(yīng)的方法來(lái)協(xié)助其工作（如果需想要進(jìn)行雙向通信或是保存結(jié)果還需要再引入一個(gè)隊(duì)列）。

worker 越多，問(wèn)題越多

按照這一思路，你現(xiàn)在需要一個(gè) worker 線(xiàn)程的線(xiàn)程池。下面是一篇 IBM 經(jīng)典教程中的例子——在進(jìn)行網(wǎng)頁(yè)檢索時(shí)通過(guò)多線(xiàn)程進(jìn)行加速。

#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)行，但仔細(xì)看看我們需要做些什么：構(gòu)造不同的方法、追蹤一系列的線(xiàn)程，還有為了解決惱人的死鎖問(wèn)題，我們需要進(jìn)行一系列的 join 操作。這還只是開(kāi)始……

至此我們回顧了經(jīng)典的多線(xiàn)程教程，多少有些空洞不是嗎？樣板化而且易出錯(cuò)，這樣事倍功半的風(fēng)格顯然不那么適合日常使用，好在我們還有更好的方法。

何不試試 map

map 這一小巧精致的函數(shù)是簡(jiǎn)捷實(shí)現(xiàn) Python 程序并行化的關(guān)鍵。map 源于 Lisp 這類(lèi)函數(shù)式編程語(yǔ)言。它可以通過(guò)一個(gè)序列實(shí)現(xiàn)兩個(gè)函數(shù)之間的映射。

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

上面的這兩行代碼將 urls 這一序列中的每個(gè)元素作為參數(shù)傳遞到 urlopen 方法中，并將所有結(jié)果保存到 results 這一列表中。其結(jié)果大致相當(dāng)于：

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

map 函數(shù)一手包辦了序列操作、參數(shù)傳遞和結(jié)果保存等一系列的操作。

為什么這很重要呢？這是因?yàn)榻柚_的庫(kù)，map 可以輕松實(shí)現(xiàn)并行化操作。

在 Python 中有個(gè)兩個(gè)庫(kù)包含了 map 函數(shù)：multiprocessing 和它鮮為人知的子庫(kù) multiprocessing.dummy.

這里多扯兩句：multiprocessing.dummy？mltiprocessing 庫(kù)的線(xiàn)程版克隆？這是蝦米？即便在 multiprocessing 庫(kù)的官方文檔里關(guān)于這一子庫(kù)也只有一句相關(guān)描述。而這句描述譯成人話(huà)基本就是說(shuō):"嘛，有這么個(gè)東西，你知道就成."相信我，這個(gè)庫(kù)被嚴(yán)重低估了！

dummy 是 multiprocessing 模塊的完整克隆，唯一的不同在于 multiprocessing 作用于進(jìn)程，而 dummy 模塊作用于線(xiàn)程（因此也包括了 Python 所有常見(jiàn)的多線(xiàn)程限制）。
所以替換使用這兩個(gè)庫(kù)異常容易。你可以針對(duì) IO 密集型任務(wù)和 CPU 密集型任務(wù)來(lái)選擇不同的庫(kù)。

動(dòng)手嘗試

使用下面的兩行代碼來(lái)引用包含并行化 map 函數(shù)的庫(kù)：

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

實(shí)例化 Pool 對(duì)象：

pool = ThreadPool()

這條簡(jiǎn)單的語(yǔ)句替代了 example2.py 中 buildworkerpool 函數(shù) 7 行代碼的工作。它生成了一系列的 worker 線(xiàn)程并完成初始化工作、將它們儲(chǔ)存在變量中以方便訪(fǎng)問(wèn)。

Pool 對(duì)象有一些參數(shù)，這里我所需要關(guān)注的只是它的第一個(gè)參數(shù)：processes. 這一參數(shù)用于設(shè)定線(xiàn)程池中的線(xiàn)程數(shù)。其默認(rèn)值為當(dāng)前機(jī)器 CPU 的核數(shù)。

一般來(lái)說(shuō)，執(zhí)行 CPU 密集型任務(wù)時(shí)，調(diào)用越多的核速度就越快。但是當(dāng)處理網(wǎng)絡(luò)密集型任務(wù)時(shí)，事情有有些難以預(yù)計(jì)了，通過(guò)實(shí)驗(yàn)來(lái)確定線(xiàn)程池的大小才是明智的。

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

線(xiàn)程數(shù)過(guò)多時(shí)，切換線(xiàn)程所消耗的時(shí)間甚至?xí)^(guò)實(shí)際工作時(shí)間。對(duì)于不同的工作，通過(guò)嘗試來(lái)找到線(xiàn)程池大小的最優(yōu)值是個(gè)不錯(cuò)的主意。

創(chuàng)建好 Pool 對(duì)象后，并行化的程序便呼之欲出了。我們來(lái)看看改寫(xiě)后的 example2.py

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

實(shí)際起作用的代碼只有 4 行，其中只有一行是關(guān)鍵的。map 函數(shù)輕而易舉的取代了前文中超過(guò) 40 行的例子。為了更有趣一些，我統(tǒng)計(jì)了不同方法、不同線(xiàn)程池大小的耗時(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)

結(jié)果：

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

很棒的結(jié)果不是嗎？這一結(jié)果也說(shuō)明了為什么要通過(guò)實(shí)驗(yàn)來(lái)確定線(xiàn)程池的大小。在我的機(jī)器上當(dāng)線(xiàn)程池大小大于 9 帶來(lái)的收益就十分有限了。

另一個(gè)真實(shí)的例子

生成上千張圖片的縮略圖
這是一個(gè) CPU 密集型的任務(wù)，并且十分適合進(jìn)行并行化。

基礎(chǔ)單進(jì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)

  for image in images:
    create_thumbnail(Image)

上邊這段代碼的主要工作就是將遍歷傳入的文件夾中的圖片文件，一一生成縮略圖，并將這些縮略圖保存到特定文件夾中。

這我的機(jī)器上，用這一程序處理 6000 張圖片需要花費(fèi) 27.9 秒。

如果我們使用 map 函數(shù)來(lái)代替 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(creat_thumbnail, images)
  pool.close()
  pool.join()

5.6 秒！

雖然只改動(dòng)了幾行代碼，我們卻明顯提高了程序的執(zhí)行速度。在生產(chǎn)環(huán)境中，我們可以為 CPU 密集型任務(wù)和 IO 密集型任務(wù)分別選擇多進(jìn)程和多線(xiàn)程庫(kù)來(lái)進(jìn)一步提高執(zhí)行速度——這也是解決死鎖問(wèn)題的良方。此外，由于 map 函數(shù)并不支持手動(dòng)線(xiàn)程管理，反而使得相關(guān)的 debug 工作也變得異常簡(jiǎn)單。

到這里，我們就實(shí)現(xiàn)了（基本）通過(guò)一行 Python 實(shí)現(xiàn)并行化。