在批評Python的討論中，常常說起Python多線程是多么的難用。還有人對 global interpreter lock(也被親切的稱為“GIL”)指指點點，說它阻礙了Python的多線程程序同時運行。因此，如果你是從其他語言（比如C++或Java）轉(zhuǎn)過來的話，Python線程模塊并不會像你想象的那樣去運行。必須要說明的是，我們還是可以用Python寫出能并發(fā)或并行的代碼，并且能帶來性能的顯著提升，只要你能顧及到一些事情。如果你還沒看過的話，我建議你看看Eqbal Quran的文章《Ruby中的并發(fā)和并行》。

在本文中，我們將會寫一個小的Python腳本，用于下載Imgur上最熱門的圖片。我們將會從一個按順序下載圖片的版本開始做起，即一個一個地下載。在那之前，你得注冊一個Imgur上的應(yīng)用。如果你還沒有Imgur賬戶，請先注冊一個。

本文中的腳本在Python3.4.2中測試通過。稍微改一下，應(yīng)該也能在Python2中運行——urllib是兩個版本中區(qū)別最大的部分。
開始動手

讓我們從創(chuàng)建一個叫“download.py”的Python模塊開始。這個文件包含了獲取圖片列表以及下載這些圖片所需的所有函數(shù)。我們將這些功能分成三個單獨的函數(shù)：

  get_links
  download_link
  setup_download_dir

第三個函數(shù)，“setup_download_dir”，用于創(chuàng)建下載的目標(biāo)目錄（如果不存在的話）。

Imgur的API要求HTTP請求能支持帶有client ID的“Authorization”頭部。你可以從你注冊的Imgur應(yīng)用的面板上找到這個client ID，而響應(yīng)會以JSON進(jìn)行編碼。我們可以使用Python的標(biāo)準(zhǔn)JSON庫去解碼。下載圖片更簡單，你只需要根據(jù)它們的URL獲取圖片，然后寫入到一個文件即可。

代碼如下：
 

import json
import logging
import os
from pathlib import Path
from urllib.request import urlopen, Request
 
logger = logging.getLogger(__name__)
 
def get_links(client_id):
  headers = {'Authorization': 'Client-ID {}'.format(client_id)}
  req = Request('https://api.imgur.com/3/gallery/', headers=headers, method='GET')
  with urlopen(req) as resp:
    data = json.loads(resp.readall().decode('utf-8'))
  return map(lambda item: item['link'], data['data'])
 
def download_link(directory, link):
  logger.info('Downloading %s', link)
  download_path = directory / os.path.basename(link)
  with urlopen(link) as image, download_path.open('wb') as f:
    f.write(image.readall())
 
def setup_download_dir():
  download_dir = Path('images')
  if not download_dir.exists():
    download_dir.mkdir()
  return download_dir

接下來，你需要寫一個模塊，利用這些函數(shù)去逐個下載圖片。我們給它命名為“single.py”。它包含了我們最原始版本的Imgur圖片下載器的主要函數(shù)。這個模塊將會通過環(huán)境變量“IMGUR_CLIENT_ID”去獲取Imgur的client ID。它將會調(diào)用“setup_download_dir”去創(chuàng)建下載目錄。最后，使用get_links函數(shù)去獲取圖片的列表，過濾掉所有的GIF和專輯URL，然后用“download_link”去將圖片下載并保存在磁盤中。下面是“single.py”的代碼：

import logging
import os
from time import time
 
from download import setup_download_dir, get_links, download_link
 
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logging.getLogger('requests').setLevel(logging.CRITICAL)
logger = logging.getLogger(__name__)
 
def main():
  ts = time()
  client_id = os.getenv('IMGUR_CLIENT_ID')
  if not client_id:
    raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!")
  download_dir = setup_download_dir()
  links = [l for l in get_links(client_id) if l.endswith('.jpg')]
  for link in links:
    download_link(download_dir, link)
  print('Took {}s'.format(time() - ts))
 
if __name__ == '__main__':
  main()

在我的筆記本上，這個腳本花了19.4秒去下載91張圖片。請注意這些數(shù)字在不同的網(wǎng)絡(luò)上也會有所不同。19.4秒并不是非常的長，但是如果我們要下載更多的圖片怎么辦呢？或許是900張而不是90張。平均下載一張圖片要0.2秒，900張的話大概需要3分鐘。那么9000張圖片將會花掉30分鐘。好消息是使用了并發(fā)或者并行后，我們可以將這個速度顯著地提高。

接下來的代碼示例將只會顯示導(dǎo)入特有模塊和新模塊的import語句。所有相關(guān)的Python腳本都可以在這方便地找到this GitHub repository。
使用線程

線程是最出名的實現(xiàn)并發(fā)和并行的方式之一。操作系統(tǒng)一般提供了線程的特性。線程比進(jìn)程要小，而且共享同一塊內(nèi)存空間。

在這里，我們將寫一個替代“single.py”的新模塊。它將創(chuàng)建一個有八個線程的池，加上主線程的話總共就是九個線程。之所以是八個線程，是因為我的電腦有8個CPU內(nèi)核，而一個工作線程對應(yīng)一個內(nèi)核看起來還不錯。在實踐中，線程的數(shù)量是仔細(xì)考究的，需要考慮到其他的因素，比如在同一臺機(jī)器上跑的的其他應(yīng)用和服務(wù)。

下面的腳本幾乎跟之前的一樣，除了我們現(xiàn)在有個新的類，DownloadWorker，一個Thread類的子類。運行無限循環(huán)的run方法已經(jīng)被重寫。在每次迭代時，它調(diào)用“self.queue.get()”試圖從一個線程安全的隊列里獲取一個URL。它將會一直堵塞，直到隊列中出現(xiàn)一個要處理元素。一旦工作線程從隊列中得到一個元素，它將會調(diào)用之前腳本中用來下載圖片到目錄中所用到的“download_link”方法。下載完成之后，工作線程向隊列發(fā)送任務(wù)完成的信號。這非常重要，因為隊列一直在跟蹤隊列中的任務(wù)數(shù)。如果工作線程沒有發(fā)出任務(wù)完成的信號，“queue.join()”的調(diào)用將會令整個主線程都在阻塞狀態(tài)。
 

from queue import Queue
from threading import Thread
 
class DownloadWorker(Thread):
  def __init__(self, queue):
    Thread.__init__(self)
    self.queue = queue
 
  def run(self):
    while True:
      # Get the work from the queue and expand the tuple
      # 從隊列中獲取任務(wù)并擴(kuò)展tuple
      directory, link = self.queue.get()
      download_link(directory, link)
      self.queue.task_done()
 
def main():
  ts = time()
  client_id = os.getenv('IMGUR_CLIENT_ID')
  if not client_id:
    raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!")
  download_dir = setup_download_dir()
  links = [l for l in get_links(client_id) if l.endswith('.jpg')]
  # Create a queue to communicate with the worker threads
  queue = Queue()
  # Create 8 worker threads
  # 創(chuàng)建八個工作線程
  for x in range(8):
    worker = DownloadWorker(queue)
    # Setting daemon to True will let the main thread exit even though the workers are blocking
    # 將daemon設(shè)置為True將會使主線程退出，即使worker都阻塞了
    worker.daemon = True
    worker.start()
  # Put the tasks into the queue as a tuple
  # 將任務(wù)以tuple的形式放入隊列中
  for link in links:
    logger.info('Queueing {}'.format(link))
    queue.put((download_dir, link))
  # Causes the main thread to wait for the queue to finish processing all the tasks
  # 讓主線程等待隊列完成所有的任務(wù)
  queue.join()
  print('Took {}'.format(time() - ts))

在同一個機(jī)器上運行這個腳本，下載時間變成了4.1秒！即比之前的例子快4.7倍。雖然這快了很多，但還是要提一下，由于GIL的緣故，在這個進(jìn)程中同一時間只有一個線程在運行。因此，這段代碼是并發(fā)的但不是并行的。而它仍然變快的原因是這是一個IO密集型的任務(wù)。進(jìn)程下載圖片時根本毫不費力，而主要的時間都花在了等待網(wǎng)絡(luò)上。這就是為什么線程可以提供很大的速度提升。每當(dāng)線程中的一個準(zhǔn)備工作時，進(jìn)程可以不斷轉(zhuǎn)換線程。使用Python或其他有GIL的解釋型語言中的線程模塊實際上會降低性能。如果你的代碼執(zhí)行的是CPU密集型的任務(wù)，例如解壓gzip文件，使用線程模塊將會導(dǎo)致執(zhí)行時間變長。對于CPU密集型任務(wù)和真正的并行執(zhí)行，我們可以使用多進(jìn)程（multiprocessing）模塊。

官方的Python實現(xiàn)——CPython——帶有GIL，但不是所有的Python實現(xiàn)都是這樣的。比如，IronPython，使用.NET框架實現(xiàn)的Python就沒有GIL，基于Java實現(xiàn)的Jython也同樣沒有。你可以點這查看現(xiàn)有的Python實現(xiàn)。
生成多進(jìn)程

多進(jìn)程模塊比線程模塊更易使用，因為我們不需要像線程示例那樣新增一個類。我們唯一需要做的改變在主函數(shù)中。

為了使用多進(jìn)程，我們得建立一個多進(jìn)程池。通過它提供的map方法，我們把URL列表傳給池，然后8個新進(jìn)程就會生成，它們將并行地去下載圖片。這就是真正的并行，不過這是有代價的。整個腳本的內(nèi)存將會被拷貝到各個子進(jìn)程中。在我們的例子中這不算什么，但是在大型程序中它很容易導(dǎo)致嚴(yán)重的問題。
 

from functools import partial
from multiprocessing.pool import Pool
 
def main():
  ts = time()
  client_id = os.getenv('IMGUR_CLIENT_ID')
  if not client_id:
    raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!")
  download_dir = setup_download_dir()
  links = [l for l in get_links(client_id) if l.endswith('.jpg')]
  download = partial(download_link, download_dir)
  with Pool(8) as p:
    p.map(download, links)
  print('Took {}s'.format(time() - ts))

分布式任務(wù)

你已經(jīng)知道了線程和多進(jìn)程模塊可以給你自己的電腦跑腳本時提供很大的幫助，那么在你想要在不同的機(jī)器上執(zhí)行任務(wù)，或者在你需要擴(kuò)大規(guī)模而超過一臺機(jī)器的的能力范圍時，你該怎么辦呢？一個很好的使用案例是網(wǎng)絡(luò)應(yīng)用的長時間后臺任務(wù)。如果你有一些很耗時的任務(wù)，你不會希望在同一臺機(jī)器上占用一些其他的應(yīng)用代碼所需要的子進(jìn)程或線程。這將會使你的應(yīng)用的性能下降，影響到你的用戶們。如果能在另外一臺甚至很多臺其他的機(jī)器上跑這些任務(wù)就好了。

Python庫RQ非常適用于這類任務(wù)。它是一個簡單卻很強(qiáng)大的庫。首先將一個函數(shù)和它的參數(shù)放入隊列中。它將函數(shù)調(diào)用的表示序列化（pickle） ，然后將這些表示添加到一個Redis列表中。任務(wù)進(jìn)入隊列只是第一步，什么都還沒有做。我們至少還需要一個能去監(jiān)聽任務(wù)隊列的worker（工作線程）。

第一步是在你的電腦上安裝和使用Redis服務(wù)器，或是擁有一臺能正常的使用的Redis服務(wù)器的使用權(quán)。接著，對于現(xiàn)有的代碼只需要一些小小的改動。先創(chuàng)建一個RQ隊列的實例并通過redis-py 庫傳給一臺Redis服務(wù)器。然后，我們執(zhí)行“q.enqueue(download_link, download_dir, link)”，而不只是調(diào)用“download_link” 。enqueue方法的第一個參數(shù)是一個函數(shù)，當(dāng)任務(wù)真正執(zhí)行時，其他的參數(shù)或關(guān)鍵字參數(shù)將會傳給該函數(shù)。

最后一步是啟動一些worker。RQ提供了方便的腳本，可以在默認(rèn)隊列上運行起worker。只要在終端窗口中執(zhí)行“rqworker”，就可以開始監(jiān)聽默認(rèn)隊列了。請確認(rèn)你當(dāng)前的工作目錄與腳本所在的是同一個。如果你想監(jiān)聽別的隊列，你可以執(zhí)行“rqworker queue_name”，然后將會開始執(zhí)行名為queue_name的隊列。RQ的一個很好的點就是，只要你可以連接到Redis，你就可以在任意數(shù)量上的機(jī)器上跑起任意數(shù)量的worker；因此，它可以讓你的應(yīng)用擴(kuò)展性得到提升。下面是RQ版本的代碼：
 

from redis import Redis
from rq import Queue
 
def main():
  client_id = os.getenv('IMGUR_CLIENT_ID')
  if not client_id:
    raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!")
  download_dir = setup_download_dir()
  links = [l for l in get_links(client_id) if l.endswith('.jpg')]
  q = Queue(connection=Redis(host='localhost', port=6379))
  for link in links:
    q.enqueue(download_link, download_dir, link)

然而RQ并不是Python任務(wù)隊列的唯一解決方案。RQ確實易用并且能在簡單的案例中起到很大的作用，但是如果有更高級的需求，我們可以使用其他的解決方案（例如 Celery）。
總結(jié)

如果你的代碼是IO密集型的，線程和多進(jìn)程可以幫到你。多進(jìn)程比線程更易用，但是消耗更多的內(nèi)存。如果你的代碼是CPU密集型的，多進(jìn)程就明顯是更好的選擇——特別是所使用的機(jī)器是多核或多CPU的。對于網(wǎng)絡(luò)應(yīng)用，在你需要擴(kuò)展到多臺機(jī)器上執(zhí)行任務(wù)，RQ是更好的選擇。

最新評論