Python 多線程通信的常用方法匯總

更新時間：2026年01月14日 09:10:01 作者：Looooking

Python中常用的線程通信方式主要有共享變量、Queue、Event、Condition、Semaphore和Barrier,這些方法都通過鎖機制保證線程安全,用于實現(xiàn)多線程之間的通信和數(shù)據(jù)交換,本文介紹Python 之多線程通信的幾種常用方法,感興趣的朋友跟隨小編一起看看吧

共享變量

共享變量是最簡單的線程通信方式，比如進(jìn)行計數(shù)更新等操作，但需要配合鎖來保證線程安全。

import threading
# 共享變量
shared_data = 0
lock = threading.Lock()
def worker():
    global shared_data
    with lock:  # 使用鎖保證線程安全
        shared_data += 1
threads = []
for i in range(5):
    t = threading.Thread(target=worker)
    threads.append(t)
    t.start()
for t in threads:
    t.join()
print(f"最終結(jié)果: {shared_data}")  # 應(yīng)該是5

Queue隊列

最常用的線程安全通信方式，是生產(chǎn)者-消費者模式的理想選擇。比如使用優(yōu)先級隊列優(yōu)先消費高優(yōu)先級的數(shù)據(jù)（序號越低，優(yōu)先級越高，越優(yōu)先消費）。

from time import sleep
from random import random, randint
from threading import Thread
from queue import PriorityQueue
queue = PriorityQueue()
def producer(queue):
    print('Producer: Running')
    for i in range(5):
        # create item with priority
        value = random()
        priority = randint(0, 5)
        item = (priority, value)
        queue.put(item)
    # wait for all items to be processed
    queue.join()
    queue.put(None)
    print('Producer: Done')
def consumer(queue):
    print('Consumer: Running')
    while True:
        # get a unit of work
        item = queue.get()
        if item is None:
            break
        sleep(item[1])
        print(item)
        queue.task_done()
    print('Consumer: Done')
producer = Thread(target=producer, args=(queue,))
producer.start()
consumer = Thread(target=consumer, args=(queue,))
consumer.start()
producer.join()
consumer.join()

Producer: Running
Consumer: Running
(0, 0.9945246262101098)
(2, 0.35853829355476663)
(2, 0.4794139132317813)
(3, 0.8460111545035349)
(5, 0.6047655828611674)
Producer: Done
Consumer: Done

Event事件

線程模提供了 Event 用于線程間的簡單信號傳遞。Event 對象管理內(nèi)部標(biāo)志的狀態(tài)。標(biāo)志初始為False，并通過 set() 方法變?yōu)?True，通過 clear() 方法重新設(shè)置為 False。wait() 方法會阻塞，直到標(biāo)志變?yōu)?True。

比如下面使用 Event 通知，模擬交通信號燈周期性變化及車輛通行之間的協(xié)同運行。車輛根據(jù)信號燈的狀態(tài)判斷是否通行還是等待；車輛通行完畢以后，只剩下信號燈周期性變化。

from threading import *
import time
def signal_state():
    while True:
        time.sleep(5)
        print("Traffic Police Giving GREEN Signal")
        event.set()
        time.sleep(10)
        print("Traffic Police Giving RED Signal")
        event.clear()
def traffic_flow():
    num = 0
    while num < 10:
        print("Waiting for GREEN Signal")
        event.wait()
        print("GREEN Signal ... Traffic can move")
        while event.is_set():
            num = num + 1
            print("Vehicle No:", num, " Crossing the Signal")
            time.sleep(2)
        print("RED Signal ... Traffic has to wait")
event = Event()
t1 = Thread(target=signal_state)
t2 = Thread(target=traffic_flow)
t1.start()
t2.start()

Waiting for GREEN Signal
Traffic Police Giving GREEN Signal
GREEN Signal ... Traffic can move
Vehicle No: 1  Crossing the Signal
Vehicle No: 2  Crossing the Signal
Vehicle No: 3  Crossing the Signal
Vehicle No: 4  Crossing the Signal
Vehicle No: 5  Crossing the Signal
Traffic Police Giving RED Signal
RED Signal ... Traffic has to wait
Waiting for GREEN Signal
Traffic Police Giving GREEN Signal
GREEN Signal ... Traffic can move
Vehicle No: 6  Crossing the Signal
Vehicle No: 7  Crossing the Signal
Vehicle No: 8  Crossing the Signal
Vehicle No: 9  Crossing the Signal
Vehicle No: 10  Crossing the Signal
Traffic Police Giving RED Signal
RED Signal ... Traffic has to wait
Traffic Police Giving GREEN Signal
Traffic Police Giving RED Signal
Traffic Police Giving GREEN Signal
Traffic Police Giving RED Signal
...

Condition條件對象

線程模塊中的 Condition 類實現(xiàn)了條件變量對象。條件對象會強制一個或多個線程等待，直到被另一個線程通知。Condition 用于更復(fù)雜的線程同步，允許線程等待特定條件。比如上面的 Event 的實現(xiàn)，其內(nèi)部也是在使用 Condition。

import threading
import time
# 共享資源
buffer = []
MAX_ITEMS = 5
condition = threading.Condition()
def producer():
    """生產(chǎn)者"""
    for i in range(10):
        time.sleep(0.2)
        with condition:
            while len(buffer) >= MAX_ITEMS:
                print("Buffer full，wait...")
                condition.wait()  # 等待緩沖區(qū)有空位
            item = f"item-{i}"
            buffer.append(item)
            print(f"Producer: {item}, Buffer: {len(buffer)}")
            condition.notify_all()  # 通知消費者
def consumer():
    """消費者"""
    for i in range(10):
        time.sleep(0.8)
        with condition:
            while len(buffer) == 0:
                print("Buffer empty，wait...")
                condition.wait()  # 等待緩沖區(qū)有數(shù)據(jù)
            item = buffer.pop(0)
            print(f"Consumer: {item}, Buffer: {len(buffer)}")
            condition.notify_all()  # 通知生產(chǎn)者
# 創(chuàng)建線程
prod = threading.Thread(target=producer)
cons = threading.Thread(target=consumer)
prod.start()
cons.start()
prod.join()
cons.join()

Producer: item-0, Buffer: 1
Producer: item-1, Buffer: 2
Producer: item-2, Buffer: 3
Consumer: item-0, Buffer: 2
Producer: item-3, Buffer: 3
Producer: item-4, Buffer: 4
Producer: item-5, Buffer: 5
Buffer full，wait...
Consumer: item-1, Buffer: 4
Producer: item-6, Buffer: 5
Buffer full，wait...
Consumer: item-2, Buffer: 4
Producer: item-7, Buffer: 5
Buffer full，wait...
Consumer: item-3, Buffer: 4
Producer: item-8, Buffer: 5
Buffer full，wait...
Consumer: item-4, Buffer: 4
Producer: item-9, Buffer: 5
Consumer: item-5, Buffer: 4
Consumer: item-6, Buffer: 3
Consumer: item-7, Buffer: 2
Consumer: item-8, Buffer: 1
Consumer: item-9, Buffer: 0

Semaphore信號量

Semaphore 信號量控制對共享資源的訪問數(shù)量。信號量的基本概念是使用一個內(nèi)部計數(shù)器，每個 acquire() 調(diào)用將其遞減，每個 release() 調(diào)用將其遞增。計數(shù)器永遠(yuǎn)不能低于零；當(dāng) acquire() 發(fā)現(xiàn)計數(shù)器為零時，它會阻塞，直到某個其他線程調(diào)用 release()。當(dāng)然，從源碼看，信號量也是通過 Condition 條件對象來進(jìn)行實現(xiàn)的。

import threading
import time
# 信號量，限制最多3個線程同時訪問
semaphore = threading.Semaphore(3)
def access_resource(thread_id):
    """訪問共享資源"""
    with semaphore:
        print(f"Thread {thread_id} acquire\n", end="")
        time.sleep(2)  # 模擬資源訪問
        print(f"Thread {thread_id} release\n", end="")
# 創(chuàng)建10個線程
threads = []
for i in range(10):
    t = threading.Thread(target=access_resource, args=(i,))
    threads.append(t)
    t.start()
for t in threads:
    t.join()

Thread 0 acquire
Thread 1 acquire
Thread 2 acquire
Thread 0 release
Thread 3 acquire
Thread 1 release
Thread 2 release
Thread 4 acquire
Thread 5 acquire
Thread 3 release
Thread 6 acquire
Thread 4 release
Thread 7 acquire
Thread 5 release
Thread 8 acquire
Thread 6 release
Thread 9 acquire
Thread 7 release
Thread 8 release
Thread 9 release

Barrier屏障

Barrier 使多個線程等待，直到指定數(shù)目的線程都到達(dá)某個點，這些線程才會被同時喚醒，然后繼續(xù)往后執(zhí)行（需要注意的是：如果沒有設(shè)置 timeout，且總的線程數(shù)無法整除給定的線程數(shù) parties 時，會導(dǎo)致線程阻塞，形成死鎖）。

import threading
import time
# 創(chuàng)建屏障，等待3個線程（注意：如果總的線程數(shù)無法整除3，則會導(dǎo)致線程阻塞）
barrier = threading.Barrier(3)
def worker(worker_id):
    """工作線程"""
    print(f"Worker {worker_id} start")
    time.sleep(worker_id)  # 模擬不同工作速度
    print(f"Worker {worker_id} arrive")
    barrier.wait()  # 等待所有線程到達(dá)
    print(f"Worker {worker_id} continue")
# 創(chuàng)建3個線程
threads = []
for i in range(6):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()
for t in threads:
    t.join()

Worker 0 start
Worker 0 arrive
Worker 1 start
Worker 2 start
Worker 3 start
Worker 4 start
Worker 5 start
Worker 1 arrive
Worker 2 arrive
Worker 2 continue
Worker 0 continue
Worker 1 continue
Worker 3 arrive
Worker 4 arrive
Worker 5 arrive
Worker 5 continue
Worker 3 continue
Worker 4 continue

不管以什么樣的方式進(jìn)行線程通信，最重要的當(dāng)屬線程安全，線程通信的各種設(shè)計，也是建立在通過鎖的機制保證線程安全的情況下來實現(xiàn)各種功能的。

到此這篇關(guān)于Python 之多線程通信的幾種常用方法的文章就介紹到這了,更多相關(guān)Python多線程通信內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: