引言

對(duì)于 Python 來(lái)說(shuō)，并不缺少并發(fā)選項(xiàng)，其標(biāo)準(zhǔn)庫(kù)中包括了對(duì)線程、進(jìn)程和異步 I/O 的支持。在許多情況下，通過(guò)創(chuàng)建諸如異步、線程和子進(jìn)程之類的高層模塊，Python 簡(jiǎn)化了各種并發(fā)方法的使用。除了標(biāo)準(zhǔn)庫(kù)之外，還有一些第三方的解決方案，例如 Twisted、Stackless 和進(jìn)程模塊。本文重點(diǎn)關(guān)注于使用 Python 的線程，并使用了一些實(shí)際的示例進(jìn)行說(shuō)明。雖然有許多很好的聯(lián)機(jī)資源詳細(xì)說(shuō)明了線程 API，但本文嘗試提供一些實(shí)際的示例，以說(shuō)明一些常見(jiàn)的線程使用模式。

全局解釋器鎖 (Global Interpretor Lock) 說(shuō)明 Python 解釋器并不是線程安全的。當(dāng)前線程必須持有全局鎖，以便對(duì) Python 對(duì)象進(jìn)行安全地訪問(wèn)。因?yàn)橹挥幸粋€(gè)線程可以獲得 Python 對(duì)象/C API，所以解釋器每經(jīng)過(guò) 100 個(gè)字節(jié)碼的指令，就有規(guī)律地釋放和重新獲得鎖。解釋器對(duì)線程切換進(jìn)行檢查的頻率可以通過(guò) sys.setcheckinterval() 函數(shù)來(lái)進(jìn)行控制。

此外，還將根據(jù)潛在的阻塞 I/O 操作，釋放和重新獲得鎖。有關(guān)更詳細(xì)的信息，請(qǐng)參見(jiàn)參考資料部分中的 Gil and Threading State 和 Threading the Global Interpreter Lock。

需要說(shuō)明的是，因?yàn)?GIL，CPU 受限的應(yīng)用程序?qū)o(wú)法從線程的使用中受益。使用 Python 時(shí)，建議使用進(jìn)程，或者混合創(chuàng)建進(jìn)程和線程。

首先弄清進(jìn)程和線程之間的區(qū)別，這一點(diǎn)是非常重要的。線程與進(jìn)程的不同之處在于，它們共享狀態(tài)、內(nèi)存和資源。對(duì)于線程來(lái)說(shuō)，這個(gè)簡(jiǎn)單的區(qū)別既是它的優(yōu)勢(shì)，又是它的缺點(diǎn)。一方面，線程是輕量級(jí)的，并且相互之間易于通信，但另一方面，它們也帶來(lái)了包括死鎖、爭(zhēng)用條件和高復(fù)雜性在內(nèi)的各種問(wèn)題。幸運(yùn)的是，由于 GIL 和隊(duì)列模塊，與采用其他的語(yǔ)言相比，采用 Python 語(yǔ)言在線程實(shí)現(xiàn)的復(fù)雜性上要低得多。
使用 Python 線程

要繼續(xù)學(xué)習(xí)本文中的內(nèi)容，我假定您已經(jīng)安裝了 Python 2.5 或者更高版本，因?yàn)楸疚闹械脑S多示例都將使用 Python 語(yǔ)言的新特性，而這些特性僅出現(xiàn)于 Python2.5 之后。要開(kāi)始使用 Python 語(yǔ)言的線程，我們將從簡(jiǎn)單的 "Hello World" 示例開(kāi)始：
hello_threads_example

    import threading
    import datetime
    
    class ThreadClass(threading.Thread):
     def run(self):
      now = datetime.datetime.now()
      print "%s says Hello World at time: %s" % 
      (self.getName(), now)
    
    for i in range(2):
     t = ThreadClass()
     t.start()

如果運(yùn)行這個(gè)示例，您將得到下面的輸出：

   # python hello_threads.py 
   Thread-1 says Hello World at time: 2008-05-13 13:22:50.252069
   Thread-2 says Hello World at time: 2008-05-13 13:22:50.252576

仔細(xì)觀察輸出結(jié)果，您可以看到從兩個(gè)線程都輸出了 Hello World 語(yǔ)句，并都帶有日期戳。如果分析實(shí)際的代碼，那么將發(fā)現(xiàn)其中包含兩個(gè)導(dǎo)入語(yǔ)句；一個(gè)語(yǔ)句導(dǎo)入了日期時(shí)間模塊，另一個(gè)語(yǔ)句導(dǎo)入線程模塊。類 ThreadClass 繼承自 threading.Thread，也正因?yàn)槿绱耍枰x一個(gè) run 方法，以此執(zhí)行您在該線程中要運(yùn)行的代碼。在這個(gè) run 方法中唯一要注意的是，self.getName() 是一個(gè)用于確定該線程名稱的方法。

最后三行代碼實(shí)際地調(diào)用該類，并啟動(dòng)線程。如果注意的話，那么會(huì)發(fā)現(xiàn)實(shí)際啟動(dòng)線程的是 t.start()。在設(shè)計(jì)線程模塊時(shí)考慮到了繼承，并且線程模塊實(shí)際上是建立在底層線程模塊的基礎(chǔ)之上的。對(duì)于大多數(shù)情況來(lái)說(shuō)，從 threading.Thread 進(jìn)行繼承是一種最佳實(shí)踐，因?yàn)樗鼊?chuàng)建了用于線程編程的常規(guī) API。
使用線程隊(duì)列

如前所述，當(dāng)多個(gè)線程需要共享數(shù)據(jù)或者資源的時(shí)候，可能會(huì)使得線程的使用變得復(fù)雜。線程模塊提供了許多同步原語(yǔ)，包括信號(hào)量、條件變量、事件和鎖。當(dāng)這些選項(xiàng)存在時(shí)，最佳實(shí)踐是轉(zhuǎn)而關(guān)注于使用隊(duì)列。相比較而言，隊(duì)列更容易處理，并且可以使得線程編程更加安全，因?yàn)樗鼈兡軌蛴行У貍魉蛦蝹€(gè)線程對(duì)資源的所有訪問(wèn)，并支持更加清晰的、可讀性更強(qiáng)的設(shè)計(jì)模式。

在下一個(gè)示例中，您將首先創(chuàng)建一個(gè)以串行方式或者依次執(zhí)行的程序，獲取網(wǎng)站的 URL，并顯示頁(yè)面的前 1024 個(gè)字節(jié)。有時(shí)使用線程可以更快地完成任務(wù)，下面就是一個(gè)典型的示例。首先，讓我們使用 urllib2 模塊以獲取這些頁(yè)面（一次獲取一個(gè)頁(yè)面），并且對(duì)代碼的運(yùn)行時(shí)間進(jìn)行計(jì)時(shí)：
URL 獲取序列

    import urllib2
    import time
    
    hosts = ["http://yahoo.com", "http://google.com", "http://amazon.com",
    "http://ibm.com", "http://apple.com"]
    
    start = time.time()
    #grabs urls of hosts and prints first 1024 bytes of page
    for host in hosts:
     url = urllib2.urlopen(host)
     print url.read(1024)
    
    print "Elapsed Time: %s" % (time.time() - start)

在運(yùn)行以上示例時(shí)，您將在標(biāo)準(zhǔn)輸出中獲得大量的輸出結(jié)果。但最后您將得到以下內(nèi)容：

    Elapsed Time: 2.40353488922

讓我們仔細(xì)分析這段代碼。您僅導(dǎo)入了兩個(gè)模塊。首先，urllib2 模塊減少了工作的復(fù)雜程度，并且獲取了 Web 頁(yè)面。然后，通過(guò)調(diào)用 time.time()，您創(chuàng)建了一個(gè)開(kāi)始時(shí)間值，然后再次調(diào)用該函數(shù)，并且減去開(kāi)始值以確定執(zhí)行該程序花費(fèi)了多長(zhǎng)時(shí)間。最后分析一下該程序的執(zhí)行速度，雖然“2.5 秒”這個(gè)結(jié)果并不算太糟，但如果您需要檢索數(shù)百個(gè) Web 頁(yè)面，那么按照這個(gè)平均值，就需要花費(fèi)大約 50 秒的時(shí)間。研究如何創(chuàng)建一種可以提高執(zhí)行速度的線程化版本：
URL 獲取線程化

     #!/usr/bin/env python
     import Queue
     import threading
     import urllib2
     import time
     
     hosts = ["http://yahoo.com", "http://google.com", "http://amazon.com",
     "http://ibm.com", "http://apple.com"]
     
     queue = Queue.Queue()
     
     class ThreadUrl(threading.Thread):
     """Threaded Url Grab"""
      def __init__(self, queue):
       threading.Thread.__init__(self)
       self.queue = queue
     
      def run(self):
       while True:
        #grabs host from queue
        host = self.queue.get()
      
        #grabs urls of hosts and prints first 1024 bytes of page
        url = urllib2.urlopen(host)
        print url.read(1024)
      
        #signals to queue job is done
        self.queue.task_done()
     
     start = time.time()
     def main():
     
      #spawn a pool of threads, and pass them queue instance 
      for i in range(5):
       t = ThreadUrl(queue)
       t.setDaemon(True)
       t.start()
       
      #populate queue with data  
       for host in hosts:
        queue.put(host)
      
      #wait on the queue until everything has been processed   
      queue.join()
     
     main()
     print "Elapsed Time: %s" % (time.time() - start)

對(duì)于這個(gè)示例，有更多的代碼需要說(shuō)明，但與第一個(gè)線程示例相比，它并沒(méi)有復(fù)雜多少，這正是因?yàn)槭褂昧岁?duì)列模塊。在 Python 中使用線程時(shí)，這個(gè)模式是一種很常見(jiàn)的并且推薦使用的方式。具體工作步驟描述如下：

•     創(chuàng)建一個(gè) Queue.Queue() 的實(shí)例，然后使用數(shù)據(jù)對(duì)它進(jìn)行填充。
•     將經(jīng)過(guò)填充數(shù)據(jù)的實(shí)例傳遞給線程類，后者是通過(guò)繼承 threading.Thread 的方式創(chuàng)建的。
•     生成守護(hù)線程池。
•     每次從隊(duì)列中取出一個(gè)項(xiàng)目，并使用該線程中的數(shù)據(jù)和 run 方法以執(zhí)行相應(yīng)的工作。
•     在完成這項(xiàng)工作之后，使用 queue.task_done() 函數(shù)向任務(wù)已經(jīng)完成的隊(duì)列發(fā)送一個(gè)信號(hào)。
•     對(duì)隊(duì)列執(zhí)行 join 操作，實(shí)際上意味著等到隊(duì)列為空，再退出主程序。

在使用這個(gè)模式時(shí)需要注意一點(diǎn)：通過(guò)將守護(hù)線程設(shè)置為 true，將允許主線程或者程序僅在守護(hù)線程處于活動(dòng)狀態(tài)時(shí)才能夠退出。這種方式創(chuàng)建了一種簡(jiǎn)單的方式以控制程序流程，因?yàn)樵谕顺鲋埃梢詫?duì)隊(duì)列執(zhí)行 join 操作、或者等到隊(duì)列為空。隊(duì)列模塊文檔詳細(xì)說(shuō)明了實(shí)際的處理過(guò)程，請(qǐng)參見(jiàn)參考資料：

    join()
    保持阻塞狀態(tài)，直到處理了隊(duì)列中的所有項(xiàng)目為止。在將一個(gè)項(xiàng)目添加到該隊(duì)列時(shí)，未完成的任務(wù)的總數(shù)就會(huì)增加。當(dāng)使用者線程調(diào)用 task_done() 以表示檢索了該項(xiàng)目、并完成了所有的工作時(shí)，那么未完成的任務(wù)的總數(shù)就會(huì)減少。當(dāng)未完成的任務(wù)的總數(shù)減少到零時(shí)，join() 就會(huì)結(jié)束阻塞狀態(tài)。

使用多個(gè)隊(duì)列

因?yàn)樯厦娼榻B的模式非常有效，所以可以通過(guò)連接附加線程池和隊(duì)列來(lái)進(jìn)行擴(kuò)展，這是相當(dāng)簡(jiǎn)單的。在上面的示例中，您僅僅輸出了 Web 頁(yè)面的開(kāi)始部分。而下一個(gè)示例則將返回各線程獲取的完整 Web 頁(yè)面，然后將結(jié)果放置到另一個(gè)隊(duì)列中。然后，對(duì)加入到第二個(gè)隊(duì)列中的另一個(gè)線程池進(jìn)行設(shè)置，然后對(duì) Web 頁(yè)面執(zhí)行相應(yīng)的處理。這個(gè)示例中所進(jìn)行的工作包括使用一個(gè)名為 Beautiful Soup 的第三方 Python 模塊來(lái)解析 Web 頁(yè)面。使用這個(gè)模塊，您只需要兩行代碼就可以提取所訪問(wèn)的每個(gè)頁(yè)面的 title 標(biāo)記，并將其打印輸出。
多隊(duì)列數(shù)據(jù)挖掘網(wǎng)站

import Queue
import threading
import urllib2
import time
from BeautifulSoup import BeautifulSoup

hosts = ["http://yahoo.com", "http://google.com", "http://amazon.com",
    "http://ibm.com", "http://apple.com"]

queue = Queue.Queue()
out_queue = Queue.Queue()

class ThreadUrl(threading.Thread):
  """Threaded Url Grab"""
  def __init__(self, queue, out_queue):
    threading.Thread.__init__(self)
    self.queue = queue
    self.out_queue = out_queue

  def run(self):
    while True:
      #grabs host from queue
      host = self.queue.get()

      #grabs urls of hosts and then grabs chunk of webpage
      url = urllib2.urlopen(host)
      chunk = url.read()

      #place chunk into out queue
      self.out_queue.put(chunk)

      #signals to queue job is done
      self.queue.task_done()

class DatamineThread(threading.Thread):
  """Threaded Url Grab"""
  def __init__(self, out_queue):
    threading.Thread.__init__(self)
    self.out_queue = out_queue

  def run(self):
    while True:
      #grabs host from queue
      chunk = self.out_queue.get()

      #parse the chunk
      soup = BeautifulSoup(chunk)
      print soup.findAll(['title'])

      #signals to queue job is done
      self.out_queue.task_done()

start = time.time()
def main():

  #spawn a pool of threads, and pass them queue instance
  for i in range(5):
    t = ThreadUrl(queue, out_queue)
    t.setDaemon(True)
    t.start()

  #populate queue with data
  for host in hosts:
    queue.put(host)

  for i in range(5):
    dt = DatamineThread(out_queue)
    dt.setDaemon(True)
    dt.start()


  #wait on the queue until everything has been processed
  queue.join()
  out_queue.join()

main()
print "Elapsed Time: %s" % (time.time() - start)

如果運(yùn)行腳本的這個(gè)版本，您將得到下面的輸出：

 # python url_fetch_threaded_part2.py 

 [<title>Google</title>]
 [<title>Yahoo!</title>]
 [<title>Apple</title>]
 [<title>IBM United States</title>]
 [<title>Amazon.com: Online Shopping for Electronics, Apparel,
 Computers, Books, DVDs & more</title>]
 Elapsed Time: 3.75387597084

分析這段代碼時(shí)您可以看到，我們添加了另一個(gè)隊(duì)列實(shí)例，然后將該隊(duì)列傳遞給第一個(gè)線程池類 ThreadURL。接下來(lái)，對(duì)于另一個(gè)線程池類 DatamineThread，幾乎復(fù)制了完全相同的結(jié)構(gòu)。在這個(gè)類的 run 方法中，從隊(duì)列中的各個(gè)線程獲取 Web 頁(yè)面、文本塊，然后使用 Beautiful Soup 處理這個(gè)文本塊。在這個(gè)示例中，使用 Beautiful Soup 提取每個(gè)頁(yè)面的 title 標(biāo)記、并將其打印輸出。可以很容易地將這個(gè)示例推廣到一些更有價(jià)值的應(yīng)用場(chǎng)景，因?yàn)槟莆樟嘶舅阉饕婊蛘邤?shù)據(jù)挖掘工具的核心內(nèi)容。一種思想是使用 Beautiful Soup 從每個(gè)頁(yè)面中提取鏈接，然后按照它們進(jìn)行導(dǎo)航。

總結(jié)

本文研究了 Python 的線程，并且說(shuō)明了如何使用隊(duì)列來(lái)降低復(fù)雜性和減少細(xì)微的錯(cuò)誤、并提高代碼可讀性的最佳實(shí)踐。盡管這個(gè)基本模式比較簡(jiǎn)單，但可以通過(guò)將隊(duì)列和線程池連接在一起，以便將這個(gè)模式用于解決各種各樣的問(wèn)題。在最后的部分中，您開(kāi)始研究如何創(chuàng)建更復(fù)雜的處理管道，它可以用作未來(lái)項(xiàng)目的模型。參考資料部分提供了很多有關(guān)常規(guī)并發(fā)性和線程的極好的參考資料。

最后，還有很重要的一點(diǎn)需要指出，線程并不能解決所有的問(wèn)題，對(duì)于許多情況，使用進(jìn)程可能更為合適。特別是，當(dāng)您僅需要?jiǎng)?chuàng)建許多子進(jìn)程并對(duì)響應(yīng)進(jìn)行偵聽(tīng)時(shí)，那么標(biāo)準(zhǔn)庫(kù)子進(jìn)程模塊可能使用起來(lái)更加容易。有關(guān)更多的官方說(shuō)明文檔，請(qǐng)參考參考資料部分。

最新評(píng)論