Python中怎樣實(shí)現多線(xiàn)程

今天就跟大家聊聊有關(guān)Python中怎樣實(shí)現多線(xiàn)程，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結了以下內容，希望大家根據這篇文章可以有所收獲。

線(xiàn)程簡(jiǎn)介

多線(xiàn)程能讓你像運行一個(gè)獨立的程序一樣運行一段長(cháng)代碼。這有點(diǎn)像調用子進(jìn)程(subprocess)，不過(guò)區別是你調用的是一個(gè)函數或者一個(gè)類(lèi)，而不是獨立的程序。在我看來(lái)，舉例說(shuō)明更有助于解釋。下面來(lái)看一個(gè)簡(jiǎn)單的例子：

import threading  def doubler(number): """ 可以被線(xiàn)程使用的一個(gè)函數 """ print(threading.currentThread.getName + '\n') print(number * 2) print  if __name__ == '__main__': for i in range(5): my_thread = threading.Thread(target=doubler, args=(i,)) my_thread.start

這里，我們導入 threading 模塊并且創(chuàng )建一個(gè)叫  doubler的常規函數。這個(gè)函數接受一個(gè)值，然后把這個(gè)值翻一番。它還會(huì )打印出調用這個(gè)函數的線(xiàn)程的名稱(chēng)，并在最后打印一行空行。然后在代碼的最后一塊，我們創(chuàng )建五個(gè)線(xiàn)程并且依次啟動(dòng)它們。在我們實(shí)例化一個(gè)線(xiàn)程時(shí)，你會(huì )注意到，我們把  doubler 函數傳給target參數，同時(shí)也給 doubler 函數傳遞了參數。Args參數看起來(lái)有些奇怪，那是因為我們需要傳遞一個(gè)序列給 doubler  函數，但它只接受一個(gè)變量，所以我們把逗號放在尾部來(lái)創(chuàng )建只有一個(gè)參數的序列。

需要注意的是，如果你想等待一個(gè)線(xiàn)程結束，那么需要調用 join方法。

當你運行以上這段代碼，會(huì )得到以下輸出內容：

Thread-1  0  Thread-2  2  Thread-3  4  Thread-4  6  Thread-5  8

當然，通常情況下你不會(huì )希望輸出打印到標準輸出。如果不幸真的這么做了，那么最終的顯示效果將會(huì )非?；靵y。你應該使用 Python 的 logging  模塊。它是線(xiàn)程安全的，并且表現出色。讓我們用 logging模塊修改上面的例子并且給我們的線(xiàn)程命名。代碼如下：

import logging import threading  def get_logger: logger = logging.getLogger("threading_example") logger.setLevel(logging.DEBUG)  fh = logging.FileHandler("threading.log") fmt = '%(asctime)s - %(threadName)s - %(levelname)s - %(message)s' formatter = logging.Formatter(fmt) fh.setFormatter(formatter)  logger.addHandler(fh) return logger  def doubler(number, logger): """ 可以被線(xiàn)程使用的一個(gè)函數 """ logger.debug('doubler function executing') result = number * 2 logger.debug('doubler function ended with: {}'.format( result))  if __name__ == '__main__': logger = get_logger thread_names = ['Mike', 'George', 'Wanda', 'Dingbat', 'Nina'] for i in range(5): my_thread = threading.Thread( target=doubler, name=thread_names[i], args=(i,logger)) my_thread.start

代碼中最大的改變就是加入了  get_logger函數。這段代碼將創(chuàng )建一個(gè)被設置為調試級別的日志記錄器。它將日志保存在當前目錄(即腳本運行所在的目錄)下，然后設置每行日志的格式。格式包括時(shí)間戳、線(xiàn)程名、日志記錄級別以及日志信息。

在 doubler 函數中，我們把 print語(yǔ)句換成 logging 語(yǔ)句。你會(huì )注發(fā)現，在創(chuàng )建線(xiàn)程時(shí)，我們給 doubler 函數傳入了 logger  對象。這樣做的原因是，如果在每個(gè)線(xiàn)程中實(shí)例化 logging 對象，那么將會(huì )產(chǎn)生多個(gè) logging  單例(singleton)，并且日志中將會(huì )有很多重復的內容。

最后，創(chuàng )建一個(gè)名稱(chēng)列表，然后使用 name關(guān)鍵字參數為每一個(gè)線(xiàn)程設置具體名稱(chēng)，這樣就可以為線(xiàn)程命名。運行以上代碼，將會(huì )得到包含以下內容的日志文件：

2016-07-24 20:39:50,055 - Mike - DEBUG - doubler function executing 2016-07-24 20:39:50,055 - Mike - DEBUG - doubler function ended with: 0 2016-07-24 20:39:50,055 - George - DEBUG - doubler function executing 2016-07-24 20:39:50,056 - George - DEBUG - doubler function ended with: 2 2016-07-24 20:39:50,056 - Wanda - DEBUG - doubler function executing 2016-07-24 20:39:50,056 - Wanda - DEBUG - doubler function ended with: 4 2016-07-24 20:39:50,056 - Dingbat - DEBUG - doubler function executing 2016-07-24 20:39:50,057 - Dingbat - DEBUG - doubler function ended with: 6 2016-07-24 20:39:50,057 - Nina - DEBUG - doubler function executing 2016-07-24 20:39:50,057 - Nina - DEBUG - doubler function ended with: 8

輸出結果不言自明，所以繼續介紹其他內容。在本節中再多說(shuō)一點(diǎn)，即通過(guò)繼承 threading.Thread實(shí)現多線(xiàn)程。舉最后一個(gè)例子，通過(guò)繼承  threading.Thread 創(chuàng )建子類(lèi)，而不是直接調用 Thread 函數。

更新后的代碼如下：

import logging import threading  class MyThread(threading.Thread): def __init__(self, number, logger): threading.Thread.__init__(self) self.number = number self.logger = logger  def run(self): """ 運行線(xiàn)程 """ logger.debug('Calling doubler') doubler(self.number, self.logger)  def get_logger: logger = logging.getLogger("threading_example") logger.setLevel(logging.DEBUG)  fh = logging.FileHandler("threading_class.log") fmt = '%(asctime)s - %(threadName)s - %(levelname)s - %(message)s' formatter = logging.Formatter(fmt) fh.setFormatter(formatter)  logger.addHandler(fh) return logger  def doubler(number, logger): """ 可以被線(xiàn)程使用的一個(gè)函數 """ logger.debug('doubler function executing') result = number * 2 logger.debug('doubler function ended with: {}'.format( result))  if __name__ == '__main__': logger = get_logger thread_names = ['Mike', 'George', 'Wanda', 'Dingbat', 'Nina'] for i in range(5): thread = MyThread(i, logger) thread.setName(thread_names[i]) thread.start

這個(gè)例子中，我們只是創(chuàng )建一個(gè)繼承于 threading.Thread的子類(lèi)。像之前一樣，傳入一個(gè)需要翻一番的數字，以及 logging  對象。但是這次，設置線(xiàn)程名稱(chēng)的方式有點(diǎn)不太一樣，變成了通過(guò)調用 thread  對象的setName方法來(lái)設置。不過(guò)仍然需要調用start來(lái)啟動(dòng)線(xiàn)程，不過(guò)你可能注意到我們并不需要在子類(lèi)中定義該方法。當調用start時(shí)，它會(huì )通過(guò)調用run方法來(lái)啟動(dòng)線(xiàn)程。在我們的類(lèi)中，我們調用  doubler 函數來(lái)做處理。輸出結果中除了一些添加的額外信息內容幾乎差不多。運行下這個(gè)腳本，看看你會(huì )得到什么。

線(xiàn)程鎖與線(xiàn)程同步

當你有多個(gè)線(xiàn)程，就需要考慮怎樣避免線(xiàn)程沖突。我的意思是說(shuō)，你可能遇到多個(gè)線(xiàn)程同時(shí)訪(fǎng)問(wèn)同一資源的情況。如果不考慮這些問(wèn)題并且制定相應的解決方案，那么在開(kāi)發(fā)產(chǎn)品過(guò)程中，你總會(huì )在最糟糕的時(shí)候遇到這些棘手的問(wèn)題。

解決辦法就是使用線(xiàn)程鎖。鎖由 Python 的 threading  模塊提供，并且它最多被一個(gè)線(xiàn)程所持有。當一個(gè)線(xiàn)程試圖獲取一個(gè)已經(jīng)鎖在資源上的鎖時(shí)，該線(xiàn)程通常會(huì )暫停運行，直到這個(gè)鎖被釋放。來(lái)讓我們看一個(gè)非常典型沒(méi)有卻應具備鎖功能的例子：

import threading  total = 0  def update_total(amount): """ Updates the total by the given amount """ global total total += amount print (total) if __name__ == '__main__': for i in range(10): my_thread = threading.Thread( target=update_total, args=(5,)) my_thread.start

如果往以上代碼添加  time.sleep函數并給出不同長(cháng)度的時(shí)間，可能會(huì )讓這個(gè)例子更有意思。無(wú)論如何，這里的問(wèn)題是，一個(gè)線(xiàn)程可能已經(jīng)調用update_total函數并且還沒(méi)有更新完成，此時(shí)另一個(gè)線(xiàn)程也有可能調用它并且嘗試更新內容。根據操作執行順序的不同，該值可能只被增加一次。

讓我們給這個(gè)函數添加鎖。有兩種方法可以實(shí)現。第一種方式是使用 try/finally，從而確保鎖肯定會(huì )被釋放。下面是示例：

import threading  total = 0  lock = threading.Lock def update_total(amount): """ Updates the total by the given amount """ global total lock.acquire try: total += amount finally: lock.release print (total)  if __name__ == '__main__': for i in range(10): my_thread = threading.Thread( target=update_total, args=(5,)) my_thread.start

如上，在我們做任何處理之前就獲取鎖。然后嘗試更新 total 的值，最后釋放鎖并打印出 total 的當前值。事實(shí)上，我們可以使用 Python 的  with語(yǔ)句避免使用 try/finally 這種較為繁瑣的語(yǔ)句：

import threading  total = 0  lock = threading.Lock  def update_total(amount): """ Updates the total by the given amount """ global total with lock: total += amount print (total)  if __name__ == '__main__': for i in range(10): my_thread = threading.Thread( target=update_total, args=(5,)) my_thread.start

正如你看到的那樣，我們不再需要 try/finally作為上下文管理器，而是由with語(yǔ)句作為替代。

當然你也會(huì )遇到要在代碼中通過(guò)多個(gè)線(xiàn)程訪(fǎng)問(wèn)多個(gè)函數的情況。當你第一次編寫(xiě)并發(fā)代碼時(shí)，代碼可能是這樣的：

import threading  total = 0  lock = threading.Lock def do_something: lock.acquire try: print('Lock acquired in the do_something function') finally: lock.release print('Lock released in the do_something function') return "Done doing something"  def do_something_else: lock.acquire try: print('Lock acquired in the do_something_else function') finally: lock.release print('Lock released in the do_something_else function') return "Finished something else"  if __name__ == '__main__': result_one = do_something result_two = do_something_else

這樣的代碼在上面的情況下能夠正常工作，但假設你有多個(gè)線(xiàn)程都調用這兩個(gè)函數呢。當一個(gè)線(xiàn)程正在運行這兩個(gè)函數，然后另外一個(gè)線(xiàn)程也可能會(huì )修改這些數據，最后得到的就是不正確的結果。問(wèn)題是，你甚至可能沒(méi)有馬上意識到結果錯了。有什么解決辦法呢?讓我們試著(zhù)找出答案。

通常首先想到的就是在調用這兩個(gè)函數的地方上鎖。讓我們試著(zhù)修改上面的例子，修改成如下所示：

import threading  total = 0  lock = threading.RLock def do_something:  with lock: print('Lock acquired in the do_something function') print('Lock released in the do_something function') return "Done doing something"   def do_something_else: with lock: print('Lock acquired in the do_something_else function') print('Lock released in the do_something_else function') return "Finished something else"  def main: with lock: result_one = do_something result_two = do_something_else print (result_one) print (result_two)  if __name__ == '__main__': main

當你真正運行這段代碼時(shí)，你會(huì )發(fā)現它只是掛起了。究其原因，是因為我們只告訴 threading  模塊獲取鎖。所以當我們調用第一個(gè)函數時(shí)，它發(fā)現鎖已經(jīng)被獲取，隨后便把自己掛起了，直到鎖被釋放，然而這將永遠不會(huì )發(fā)生。

真正的解決辦法是使用重入鎖(Re-Entrant Lock)。threading 模塊提供的解決辦法是使用RLock函數。即把lock =  threading.lock替換為lock = threading.RLock，然后重新運行代碼，現在代碼就可以正常運行了。

如果你想在線(xiàn)程中運行以上代碼，那么你可以用以下代碼取代直接調用 main函數：

if __name__ == '__main__': for i in range(10): my_thread = threading.Thread( target=main) my_thread.start

每個(gè)線(xiàn)程都會(huì )運行 main 函數，main 函數則會(huì )依次調用另外兩個(gè)函數。最終也會(huì )產(chǎn)生 10 組結果集。

定時(shí)器

Threading 模塊有一個(gè)優(yōu)雅的  Timer類(lèi)，你可以用它來(lái)實(shí)現在指定時(shí)間后要發(fā)生的動(dòng)作。它們實(shí)際上會(huì )啟動(dòng)自己的自定義線(xiàn)程，通過(guò)調用常規線(xiàn)程上的start方法即可運行。你也可以調用它的cancel方法停止定時(shí)器。值得注意的是，你甚至可以在開(kāi)始定時(shí)器之前取消它。

有一天，我遇到一個(gè)特殊的情況：我需要與已經(jīng)啟動(dòng)的子進(jìn)程通信，但是我需要它有超時(shí)處理。雖然處理這種特殊問(wèn)題有很多不同的方法，不過(guò)我最喜歡的解決方案是使用  threading 模塊的 Timer 類(lèi)。

在下面這個(gè)例子中，我們將使用 ping指令作為演示。在 Linux 系統中，ping 命令會(huì )一直運行下去直到你手動(dòng)殺死它。所以在 Linux  世界里，Timer 類(lèi)就顯得非常方便。示例如下：

import subprocess from threading import Timer  kill = lambda process: process.kill cmd = ['ping', 'www.google.com'] ping = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)  my_timer = Timer(5, kill, [ping]) try: my_timer.start stdout, stderr = ping.communicate finally: my_timer.cancel print (str(stdout))

這里我們在 lambda 表達式中調用 kill 殺死進(jìn)程。接下來(lái)啟動(dòng) ping 命令，然后創(chuàng )建 Timer  對象。你會(huì )注意到，第一個(gè)參數就是需要等待的秒數，第二個(gè)參數是需要調用的函數，緊跟其后的參數是要調用函數的入參。在本例中，我們的函數是一個(gè) lambda  表達式，傳入的是一個(gè)只有一個(gè)元素的列表。如果你運行這段代碼，它應該會(huì )運行 5 秒鐘，然后打印出 ping 的結果。

其他線(xiàn)程組件

Threading  模塊包含對其他功能的支持。例如，你可以創(chuàng )建信號量(Semaphore)，這是計算機科學(xué)中最古老的同步原語(yǔ)之一?；旧?，一個(gè)信號量管理一個(gè)內置的計數器。當你調用acquire時(shí)計數器就會(huì )遞減，相反當你調用release時(shí)就會(huì )遞增。根據其設計，計數器的值無(wú)法小于零，所以如果正好在計數器為零時(shí)調用  acquire 方法，該方法將阻塞線(xiàn)程。

譯者注：通常使用信號量時(shí)都會(huì )初始化一個(gè)大于零的值，如 semaphore = threading.Semaphore(2)

另一個(gè)非常有用的同步工具就是事件(Event)。它允許你使用信號(signal)實(shí)現線(xiàn)程通信。在下一節中我們將舉一個(gè)使用事件的實(shí)例。

最后，在 Python 3.2 中加入了 Barrier對象。Barrier 是管理線(xiàn)程池中的同步原語(yǔ)，在線(xiàn)程池中多條線(xiàn)程需要相互等待對方。如果要傳遞  barrier，每一條線(xiàn)程都要調用wait方法，在其他線(xiàn)程調用該方法之前線(xiàn)程將會(huì )阻塞。全部調用之后將會(huì )同時(shí)釋放所有線(xiàn)程。

線(xiàn)程通信

某些情況下，你會(huì )希望線(xiàn)程之間互相通信。就像先前提到的，你可以通過(guò)創(chuàng )建  Event對象達到這個(gè)目的。但更常用的方法是使用隊列(Queue)。在我們的例子中，這兩種方式都會(huì )有所涉及。下面讓我們看看到底是什么樣子的：

import threading from queue import Queue  def creator(data, q): """ 生成用于消費的數據，等待消費者完成處理 """ print('Creating data and putting it on the queue') for item in data: evt = threading.Event q.put((item, evt))  print('Waiting for data to be doubled') evt.wait  def my_consumer(q): """ 消費部分數據，并做處理  這里所做的只是將輸入翻一倍  """ while True: data, evt = q.get print('data found to be processed: {}'.format(data)) processed = data * 2 print(processed) evt.set q.task_done  if __name__ == '__main__': q = Queue data = [5, 10, 13, -1] thread_one = threading.Thread(target=creator, args=(data, q)) thread_two = threading.Thread(target=my_consumer, args=(q,)) thread_one.start thread_two.start  q.join

讓我們掰開(kāi)揉碎分析一下。首先，我們有一個(gè)創(chuàng )建者(creator)函數(亦稱(chēng)作生產(chǎn)者(producer))，我們用它來(lái)創(chuàng )建想要操作(或者消費)的數據。然后用另外一個(gè)函數  my_consumer來(lái)處理剛才創(chuàng )建出來(lái)的數據。Creator 函數使用 Queue  的put方法向隊列中插入數據，消費者將會(huì )持續不斷的檢測有沒(méi)有更多的數據，當發(fā)現有數據時(shí)就會(huì )處理數據。Queue  對象處理所有的獲取鎖和釋放鎖的過(guò)程，這些不用我們太關(guān)心。

在這個(gè)例子中，先創(chuàng )建一個(gè)列表，然后創(chuàng )建兩個(gè)線(xiàn)程，一個(gè)用作生產(chǎn)者，一個(gè)作為消費者。你會(huì )發(fā)現，我們給兩個(gè)線(xiàn)程都傳遞了 Queue  對象，這兩個(gè)線(xiàn)程隱藏了關(guān)于鎖處理的細節。隊列實(shí)現了數據從第一個(gè)線(xiàn)程到第二個(gè)線(xiàn)程的傳遞。當第一個(gè)線(xiàn)程把數據放入隊列時(shí)，同時(shí)也傳遞一個(gè) Event  事件，緊接著(zhù)掛起自己，等待該事件結束。在消費者側，也就是第二個(gè)線(xiàn)程，則做數據處理工作。當完成數據處理后就會(huì )調用 Event 事件的  set方法，通知第一個(gè)線(xiàn)程已經(jīng)把數據處理完畢了，可以繼續生產(chǎn)了。