問題
你的程序中有多個線程，你需要在這些線程之間安全地交換信息或數據
解決方案
從一個線程向另一個線程發送數據最安全的方式可能就是使用 queue 庫中的隊列了。創建一個被多個線程共享的 Queue 對象，這些線程通過使用 put() 和 get() 操作來向隊列中添加或者刪除元素。 例如：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
from queue import Queuefrom threading import Thread# A thread that produces datadef producer(out_q): while True: # Produce some data ... out_q.put(data)# A thread that consumes datadef consumer(in_q): while True:# Get some data data = in_q.get() # Process the data ...# Create the shared queue and launch both threadsq = Queue()t1 = Thread(target=consumer, args=(q,))t2 = Thread(target=producer, args=(q,))t1.start()t2.start() |
Queue 對象已經包含了必要的鎖，所以你可以通過它在多個線程間多安全地共享數據。 當使用隊列時，協調生產者和消費者的關閉問題可能會有一些麻煩。一個通用的解決方法是在隊列中放置一個特殊的值，當消費者讀到這個值的時候，終止執行。例如：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
from queue import Queuefrom threading import Thread# Object that signals shutdown_sentinel = object()# A thread that produces datadef producer(out_q): while running: # Produce some data ... out_q.put(data) # Put the sentinel on the queue to indicate completion out_q.put(_sentinel)# A thread that consumes datadef consumer(in_q): while True: # Get some data data = in_q.get() # Check for termination if data is _sentinel: in_q.put(_sentinel) break # Process the data ... |
本例中有一個特殊的地方：消費者在讀到這個特殊值之后立即又把它放回到隊列中，將之傳遞下去。這樣，所有監聽這個隊列的消費者線程就可以全部關閉了。 盡管隊列是最常見的線程間通信機制，但是仍然可以自己通過創建自己的數據結構并添加所需的鎖和同步機制來實現線程間通信。最常見的方法是使用 Condition 變量來包裝你的數據結構。下邊這個例子演示了如何創建一個線程安全的優先級隊列
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
import heapqimport threadingclass PriorityQueue: def __init__(self): self._queue = [] self._count = 0 self._cv = threading.Condition() def put(self, item, priority): with self._cv: heapq.heappush(self._queue, (-priority, self._count, item)) self._count += 1 self._cv.notify() def get(self): with self._cv: while len(self._queue) == 0: self._cv.wait() return heapq.heappop(self._queue)[-1] |
使用隊列來進行線程間通信是一個單向、不確定的過程。通常情況下，你沒有辦法知道接收數據的線程是什么時候接收到的數據并開始工作的。不過隊列對象提供一些基本完成的特性，比如下邊這個例子中的 task_done() 和 join() ：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
from queue import Queuefrom threading import Thread# A thread that produces datadef producer(out_q): while running: # Produce some data ... out_q.put(data)# A thread that consumes datadef consumer(in_q): while True: # Get some data data = in_q.get() # Process the data ... # Indicate completion in_q.task_done()# Create the shared queue and launch both threadsq = Queue()t1 = Thread(target=consumer, args=(q,))t2 = Thread(target=producer, args=(q,))t1.start()t2.start()# Wait for all produced items to be consumedq.join() |
如果一個線程需要在一個“消費者”線程處理完特定的數據項時立即得到通知，你可以把要發送的數據和一個 Event 放到一起使用，這樣“生產者”就可以通過這個Event對象來監測處理的過程了。示例如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
from queue import Queuefrom threading import Thread, Event# A thread that produces datadef producer(out_q): while running: # Produce some data ... # Make an (data, event) pair and hand it to the consumer evt = Event() out_q.put((data, evt)) ... # Wait for the consumer to process the item evt.wait()# A thread that consumes datadef consumer(in_q): while True: # Get some data data, evt = in_q.get() # Process the data ... # Indicate completion evt.set() |
討論
基于簡單隊列編寫多線程程序在多數情況下是一個比較明智的選擇。從線程安全隊列的底層實現來看，你無需在你的代碼中使用鎖和其他底層的同步機制，這些只會把你的程序弄得亂七八糟。此外，使用隊列這種基于消息的通信機制可以被擴展到更大的應用范疇，比如，你可以把你的程序放入多個進程甚至是分布式系統而無需改變底層的隊列結構。 使用線程隊列有一個要注意的問題是，向隊列中添加數據項時并不會復制此數據項，線程間通信實際上是在線程間傳遞對象引用。如果你擔心對象的共享狀態，那你最好只傳遞不可修改的數據結構（如：整型、字符串或者元組）或者一個對象的深拷貝。例如：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
from queue import Queuefrom threading import Threadimport copy# A thread that produces datadef producer(out_q): while True: # Produce some data ... out_q.put(copy.deepcopy(data))# A thread that consumes datadef consumer(in_q): while True: # Get some data data = in_q.get() # Process the data ... |
Queue 對象提供一些在當前上下文很有用的附加特性。比如在創建 Queue 對象時提供可選的 size 參數來限制可以添加到隊列中的元素數量。對于“生產者”與“消費者”速度有差異的情況，為隊列中的元素數量添加上限是有意義的。比如，一個“生產者”產生項目的速度比“消費者” “消費”的速度快，那么使用固定大小的隊列就可以在隊列已滿的時候阻塞隊列，以免未預期的連鎖效應擴散整個程序造成死鎖或者程序運行失常。在通信的線程之間進行“流量控制”是一個看起來容易實現起來困難的問題。如果你發現自己曾經試圖通過擺弄隊列大小來解決一個問題，這也許就標志著你的程序可能存在脆弱設計或者固有的可伸縮問題。 get() 和 put() 方法都支持非阻塞方式和設定超時，例如：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
import queueq = queue.Queue()try: data = q.get(block=False)except queue.Empty: ...try: q.put(item, block=False)except queue.Full: ...try: data = q.get(timeout=5.0)except queue.Empty: ... |
這些操作都可以用來避免當執行某些特定隊列操作時發生無限阻塞的情況，比如，一個非阻塞的 put() 方法和一個固定大小的隊列一起使用，這樣當隊列已滿時就可以執行不同的代碼。比如輸出一條日志信息并丟棄。
|
1
2
3
4
5
6
|
def producer(q): ... try: q.put(item, block=False) except queue.Full: log.warning('queued item %r discarded!', item) |
如果你試圖讓消費者線程在執行像 q.get() 這樣的操作時，超時自動終止以便檢查終止標志，你應該使用 q.get() 的可選參數 timeout ，如下：
|
1
2
3
4
5
6
7
8
9
10
|
_running = Truedef consumer(q): while _running: try: item = q.get(timeout=5.0) # Process item ... except queue.Empty: pass |
最后，有 q.qsize() ， q.full() ， q.empty() 等實用方法可以獲取一個隊列的當前大小和狀態。但要注意，這些方法都不是線程安全的。可能你對一個隊列使用 empty() 判斷出這個隊列為空，但同時另外一個線程可能已經向這個隊列中插入一個數據項。所以，你最好不要在你的代碼中使用這些方法。
以上就是Python如何實現線程間通信的詳細內容，更多關于Python 線程間通信的資料請關注服務器之家其它相關文章！
原文鏈接：https://python3-cookbook.readthedocs.io/zh_CN/latest/c12/p03_communicating_between_threads.html
