快捷導(dǎo)航

一文帶你掌握Python多線程同步的核心機(jī)制與最佳實(shí)踐

更新時間：2026年01月29日 09:06:43 作者：xingzhemengyou1

Python 的 threading 模塊提供了多種同步原語,每種都有其特定的適用場景,本文將詳細(xì)介紹這些核心機(jī)制、使用方法和最佳實(shí)踐,感興趣的小伙伴可以了解下

Python 多線程同步是確保多個線程安全、有序地訪問共享資源，避免數(shù)據(jù)競爭（Race Condition）和保證數(shù)據(jù)一致性的關(guān)鍵技術(shù)。Python 的 threading 模塊提供了多種同步原語，每種都有其特定的適用場景。以下將詳細(xì)介紹這些核心機(jī)制、使用方法和最佳實(shí)踐。

一、線程鎖（Lock）

線程鎖是最基礎(chǔ)、最常用的同步工具，用于確保在任意時刻只有一個線程能進(jìn)入被保護(hù)的代碼段（臨界區(qū)）。

基本用法：通過 threading.Lock() 創(chuàng)建鎖對象，使用 acquire() 獲取鎖，release() 釋放鎖。推薦使用 with 語句自動管理鎖的獲取和釋放，避免因異常導(dǎo)致鎖無法釋放。

import threading

# 創(chuàng)建鎖
lock = threading.Lock()
shared_data = 0

def increment():
    global shared_data
    for _ in range(100000):
        # 獲取鎖
        lock.acquire()
        try:
            shared_data += 1
        finally:
            # 釋放鎖
            lock.release()

# 使用 with 語句更安全
def increment_safe():
    global shared_data
    for _ in range(100000):
        with lock:  # 自動獲取和釋放鎖
            shared_data += 1

# 創(chuàng)建線程
t1 = threading.Thread(target=increment_safe)
t2 = threading.Thread(target=increment_safe)

t1.start()
t2.start()
t1.join()
t2.join()

print(f"最終結(jié)果: {shared_data}")  # 應(yīng)該是 200000

可重入鎖（RLock）：普通鎖（Lock）在同一線程內(nèi)不可重入，連續(xù)調(diào)用 acquire() 會導(dǎo)致死鎖。threading.RLock()（可重入鎖）允許同一線程多次獲取鎖，通常用于遞歸函數(shù)或多次調(diào)用需要同步的同一方法。

import threading

rlock = threading.RLock()

def recursive_func(n):
    with rlock:
        print(f"Entering Level {n}")
        if n > 0:
            recursive_func(n - 1)
        print(f"Exiting Level {n}")

print("Starting thread with recursive RLock example")
thread = threading.Thread(target=recursive_func, args=(3,))
thread.start()
thread.join()
print("Thread completed successfully")

二、信號量（Semaphore）

信號量用于控制同時訪問某個共享資源的線程數(shù)量，其內(nèi)部維護(hù)一個計數(shù)器。threading.Semaphore(value=N) 初始化時指定最大并發(fā)數(shù)，acquire() 減少計數(shù)，release() 增加計數(shù)。

import threading
import time

# 最多允許3個線程同時訪問
sem = threading.Semaphore(3)

def access_resource(thread_id):
    print(f"Thread {thread_id} is waiting to access")
    with sem:
        print(f"Thread {thread_id} has acquired the semaphore and is working")
        time.sleep(1)
        print(f"Thread {thread_id} is releasing the semaphore")

print("Starting threads with semaphore limit of 3...")
threads = []
for i in range(10):
    t = threading.Thread(target=access_resource, args=(i,))
    threads.append(t)
    t.start()
    time.sleep(0.1)  # 稍微延遲創(chuàng)建線程，便于觀察

for t in threads:
    t.join()

print("All threads completed")

threading.BoundedSemaphore 是信號量的變體，可防止計數(shù)超過初始值。

import threading
import time

# 創(chuàng)建一個有界信號量，最多允許3個線程同時訪問
bounded_sem = threading.BoundedSemaphore(3)

def normal_access(thread_id):
    """正常使用有界信號量的函數(shù)"""
    print(f"Thread {thread_id} is waiting to access")
    with bounded_sem:
        print(f"Thread {thread_id} has acquired the bounded semaphore and is working")
        time.sleep(1)
        print(f"Thread {thread_id} is releasing the bounded semaphore")

def demonstrate_bounded_semaphore_error():
    """演示有界信號量防止過度釋放的特性"""
    print("\n演示BoundedSemaphore防止過度釋放的特性:")
    # 先獲取信號量
    print("嘗試獲取信號量...")
    bounded_sem.acquire()
    print("信號量獲取成功")
    
    # 正常釋放一次
    print("正常釋放一次信號量...")
    bounded_sem.release()
    print("信號量正常釋放")
    
    # 嘗試過度釋放
    try:
        print("嘗試過度釋放信號量(這將引發(fā)ValueError異常)...")
        bounded_sem.release()  # 這里會拋出ValueError
        print("如果看到這條消息，說明有界信號量沒有正常工作！")
    except ValueError as e:
        print(f"成功捕獲異常: {e}")
        print("這證明BoundedSemaphore可以防止過度釋放的情況發(fā)生")

print("=== 有界信號量(BoundedSemaphore)示例程序 ===")
print("1. 首先運(yùn)行多線程正常訪問的場景")

# 創(chuàng)建并啟動多個線程
threads = []
for i in range(5):
    t = threading.Thread(target=normal_access, args=(i,))
    threads.append(t)
    t.start()
    time.sleep(0.2)  # 稍微延遲創(chuàng)建線程，便于觀察

# 等待所有線程完成
for t in threads:
    t.join()

print("\n所有線程完成正常訪問")

# 演示有界信號量的錯誤檢測功能
demonstrate_bounded_semaphore_error()

print("\n=== 程序結(jié)束 ===")

三、條件變量（Condition）

條件變量用于復(fù)雜的線程間協(xié)調(diào)，允許一個或多個線程等待特定條件成立后再繼續(xù)執(zhí)行。它通常與一個鎖關(guān)聯(lián)，提供了 wait()、notify() 和 notify_all() 方法。

典型應(yīng)用：生產(chǎn)者-消費(fèi)者模型

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
生產(chǎn)者-消費(fèi)者模型實(shí)現(xiàn)

本程序演示使用threading.Condition（條件變量）實(shí)現(xiàn)經(jīng)典的生產(chǎn)者-消費(fèi)者模型。
生產(chǎn)者線程生成隨機(jī)數(shù)并放入固定大小的緩沖區(qū)，消費(fèi)者線程從緩沖區(qū)取出數(shù)據(jù)。
當(dāng)緩沖區(qū)滿時，生產(chǎn)者等待；當(dāng)緩沖區(qū)空時，消費(fèi)者等待。
"""

import threading
import time
import random

# 共享緩沖區(qū)，用于存儲生產(chǎn)者生產(chǎn)的物品
buffer = []

# 緩沖區(qū)最大容量
MAX_SIZE = 10

# 創(chuàng)建條件變量，用于線程間通信和同步
# Condition內(nèi)部維護(hù)一個鎖，提供wait()、notify()等方法進(jìn)行線程協(xié)作
condition = threading.Condition()

class Producer(threading.Thread):
    """
    生產(chǎn)者類，繼承自threading.Thread
    負(fù)責(zé)生成隨機(jī)數(shù)并添加到緩沖區(qū)
    """
    def run(self):
        """
        線程運(yùn)行方法
        使用條件變量確保線程安全的緩沖區(qū)訪問
        """
        global buffer
        
        # 無限循環(huán)生產(chǎn)物品
        while True:
            # 獲取條件變量的鎖
            with condition:
                # 檢查緩沖區(qū)是否已滿
                # 注意：此處應(yīng)使用while循環(huán)而非if語句，防止虛假喚醒
                while len(buffer) == MAX_SIZE:
                    print("Buffer full, producer waiting")
                    # 釋放鎖并等待，直到被其他線程通知
                    condition.wait()
                
                # 生產(chǎn)一個隨機(jī)物品
                item = random.randint(1, 100)
                # 將物品添加到緩沖區(qū)
                buffer.append(item)
                print(f"Produced {item}, buffer: {buffer}")
                
                # 通知一個等待的消費(fèi)者線程
                condition.notify()
            
            # 生產(chǎn)間隔隨機(jī)時間，模擬生產(chǎn)過程
            time.sleep(random.random())

class Consumer(threading.Thread):
    """
    消費(fèi)者類，繼承自threading.Thread
    負(fù)責(zé)從緩沖區(qū)取出物品進(jìn)行消費(fèi)
    """
    def run(self):
        """
        線程運(yùn)行方法
        使用條件變量確保線程安全的緩沖區(qū)訪問
        """
        global buffer
        
        # 無限循環(huán)消費(fèi)物品
        while True:
            # 獲取條件變量的鎖
            with condition:
                # 檢查緩沖區(qū)是否為空
                # 同樣使用while循環(huán)防止虛假喚醒
                while not buffer:
                    print("Buffer empty, consumer waiting")
                    # 釋放鎖并等待，直到被其他線程通知
                    condition.wait()
                
                # 從緩沖區(qū)取出第一個物品（FIFO原則）
                item = buffer.pop(0)
                print(f"Consumed {item}, buffer: {buffer}")
                
                # 通知一個等待的生產(chǎn)者線程
                condition.notify()
            
            # 消費(fèi)間隔隨機(jī)時間，模擬消費(fèi)過程
            time.sleep(random.random())

# 程序入口點(diǎn)
if __name__ == "__main__":
    print("生產(chǎn)者-消費(fèi)者模型演示程序啟動...")
    print(f"緩沖區(qū)最大容量: {MAX_SIZE}")
    
    # 創(chuàng)建并啟動生產(chǎn)者線程
    Producer().start()
    
    # 創(chuàng)建并啟動消費(fèi)者線程
    Consumer().start()
    
    print("線程已啟動，按Ctrl+C停止程序")

wait_for(predicate, timeout) 方法可等待某個條件表達(dá)式為真，使代碼更簡潔。

四、事件（Event）

事件是一種簡單的線程間通信機(jī)制，一個線程發(fā)出“事件已發(fā)生”的信號，其他線程等待該信號。通過 threading.Event() 創(chuàng)建事件對象，主要方法有 set()、clear() 和 wait()。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
Threading.Event示例程序

本程序演示了Python中threading.Event的基本用法。
Event對象是線程間同步的一種機(jī)制，類似于一個信號標(biāo)志，
可以控制線程的執(zhí)行和等待狀態(tài)。
"""

import threading
import time

# 創(chuàng)建Event對象，初始狀態(tài)為False
event = threading.Event()

def waiter():
    """
    等待者線程函數(shù)
    
    該函數(shù)模擬一個需要等待某個事件發(fā)生才能繼續(xù)執(zhí)行的線程
    當(dāng)event被設(shè)置（set）時，wait()方法才會解除阻塞
    """
    print("Waiter: waiting for event")
    # 阻塞當(dāng)前線程，直到event被設(shè)置
    event.wait()  # 阻塞直到事件被設(shè)置
    print("Waiter: event received, proceeding")

def setter():
    """
    設(shè)置者線程函數(shù)
    
    該函數(shù)模擬一個在某個時刻觸發(fā)事件的線程
    通過調(diào)用event.set()方法，將事件標(biāo)志設(shè)置為True，
    喚醒所有等待該事件的線程
    """
    # 模擬耗時操作
    time.sleep(2)
    print("Setter: setting event")
    # 設(shè)置事件，喚醒所有等待的線程
    event.set()  # 設(shè)置事件，喚醒所有等待的線程

# 創(chuàng)建并啟動等待者線程
threading.Thread(target=waiter).start()
# 創(chuàng)建并啟動設(shè)置者線程
threading.Thread(target=setter).start()

五、其他同步工具與線程安全數(shù)據(jù)結(jié)構(gòu)

屏障（Barrier）：使一組線程必須全部到達(dá)某個點(diǎn)后才能繼續(xù)執(zhí)行，適用于分階段任務(wù)。
線程本地數(shù)據(jù)（threading.local）：為每個線程創(chuàng)建獨(dú)立的變量副本，避免共享，從根本上解決同步問題。
隊(duì)列（Queue）：queue 模塊提供的 Queue、LifoQueue、PriorityQueue 是線程安全的，內(nèi)部已實(shí)現(xiàn)鎖機(jī)制，非常適合生產(chǎn)者-消費(fèi)者模式，無需顯式同步。

六、全局解釋器鎖（GIL）的影響與最佳實(shí)踐

GIL 的影響： Python 的全局解釋器鎖（GIL）確保同一時刻只有一個線程執(zhí)行 Python 字節(jié)碼。這意味著對于 CPU 密集型任務(wù)，多線程無法利用多核優(yōu)勢，性能提升有限甚至可能下降。但對于 I/O 密集型任務(wù)（如網(wǎng)絡(luò)請求、文件讀寫），線程在等待 I/O 時會釋放 GIL，因此多線程仍能有效提升并發(fā)性能。

多線程同步的最佳實(shí)踐：

選擇合適的工具：簡單互斥用 Lock，控制并發(fā)數(shù)量用 Semaphore，復(fù)雜協(xié)調(diào)用 Condition 或 Event，數(shù)據(jù)傳遞用 Queue。
避免死鎖：確保多個鎖的獲取順序一致；考慮使用帶超時的 acquire(timeout)；優(yōu)先使用可重入鎖（RLock）或高層抽象（如 Queue）。
最小化鎖的持有時間：只將必須同步的代碼放入臨界區(qū)，盡快釋放鎖以提高并發(fā)性。
考慮使用線程安全的數(shù)據(jù)結(jié)構(gòu)：如 queue.Queue，可減少手動同步的復(fù)雜度。
對于CPU密集型任務(wù)考慮多進(jìn)程：使用 multiprocessing 模塊繞過 GIL 限制，充分利用多核CPU。

總結(jié)

Python 提供了豐富的線程同步機(jī)制，從基礎(chǔ)的 Lock、Semaphore 到更高級的 Condition、Event 和 Barrier，以及線程安全的 Queue。選擇何種機(jī)制取決于具體場景：保護(hù)簡單共享變量可用 Lock；控制資源并發(fā)數(shù)用 Semaphore；實(shí)現(xiàn)線程間依賴和通知用 Condition 或 Event；在線程間傳遞數(shù)據(jù)則優(yōu)先使用 Queue。同時，必須注意 GIL 對多線程并行的限制，并遵循避免死鎖、最小化鎖范圍等最佳實(shí)踐，才能編寫出高效、健壯的多線程程序。