Python如何保证线程安全？锁机制到底怎么用才不出错

Python多线程编程中，线程安全的核心在于控制对共享资源的并发访问。由于操作系统调度线程的顺序不可预测，当多个线程同时读写同一块内存区域时，中间状态可能被相互覆盖，从而产生数据错误。理解锁机制，是写出正确并发程序的基础。

为什么Python需要额外的锁

很多初学者误以为Python有全局解释器锁（GIL），多线程就天然安全。实际上GIL只是保证同一时刻只有一个线程执行Python字节码，避免了解释器内部崩溃，但它不会替你管理业务数据。例如一个线程刚读出变量值还没写回，就被切换走，另一个线程也基于旧值修改，最终保存的结果就会丢失更新。

我们用一个简单例子说明问题。下面这段代码启动十个线程，每个线程对全局变量累加一万次，理论上结果应是十万，但实际运行往往小于这个数。

import threading

count = 0

def worker():
    global count
    for _ in range(10000):
        count = count + 1

threads = []
for i in range(10):
    t = threading.Thread(target=worker)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(count)  # 经常输出小于100000的值

出现偏差的原因在于count = count + 1不是原子操作，它包含读取、计算、写入三步。线程切换可能发生在任意两步之间，导致部分递增被覆盖。要消除这种竞态，就必须用锁把这段代码保护起来。

threading.Lock的基本用法

threading.Lock是最基础的同步原语。它只有两种状态：锁定与未锁定。某个线程调用acquire()后，其他线程再调用acquire()就会阻塞，直到该线程调用release()释放锁。这样就能保证被保护的代码同一时间只被一个线程执行。

改写前面的例子，引入锁对象。注意要把锁作为全局或闭包变量传递，并在修改共享数据时加锁。使用try/finally可以确保即使代码异常也能释放锁，避免死锁。

import threading

count = 0
lock = threading.Lock()

def worker():
    global count
    for _ in range(10000):
        lock.acquire()
        try:
            count = count + 1
        finally:
            lock.release()

threads = []
for i in range(10):
    t = threading.Thread(target=worker)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(count)  # 稳定输出100000

上面的写法虽然正确，但容易忘记释放。Python提供了更优雅的上下文管理器写法，用with语句自动管理锁的生命周期，代码更简洁也更安全。

import threading

count = 0
lock = threading.Lock()

def worker():
    global count
    for _ in range(10000):
        with lock:
            count = count + 1

threads = []
for i in range(10):
    t = threading.Thread(target=worker)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(count)

RLock与死锁预防

Lock不允许同一线程多次acquire，否则会死锁。如果在复杂函数中，一个加锁函数调用了另一个也需要同把锁的函数，普通Lock就会卡住。threading.RLock（可重入锁）解决了这个问题，它记录持有线程和加锁次数，同一线程可重复获取，释放次数与获取次数匹配时才真正解锁。

下面演示RLock的典型使用场景。外层函数获取锁后调用内层函数，内层再次获取同一把RLock不会阻塞。

import threading

rlock = threading.RLock()

def inner():
    with rlock:
        print('内层执行')

def outer():
    with rlock:
        print('外层执行')
        inner()

t = threading.Thread(target=outer)
t.start()
t.join()

需要注意的是，RLock虽然方便，但也可能掩盖设计问题。如果逻辑上不需要重入，优先用Lock，因为它的开销更小，且能更早暴露错误的锁使用方式。另外，锁的粒度要适中：太粗会削弱并发性能，太细则容易遗漏保护点。

其他常见锁与选择建议

除了Lock和RLock，threading模块还提供Semaphore（信号量，限制同时访问的线程数）、Condition（条件变量，用于线程间通知）、Event（事件，简化标志位等待）。它们底层都依赖锁机制，适用于不同协调需求。

实际开发中，若只是保护简单变量或字典，Lock加with足够；若涉及递归调用或同一线程多次进入临界区，选RLock；若需控制资源池大小，用BoundedSemaphore。无论哪种，都要保证加锁区间尽量短，且不要在锁内做网络请求或睡眠，否则会严重拖慢整个程序。

掌握这些锁的差异与正确写法，才能在Python多线程程序中真正保证线程安全，避免难以排查的数据错乱问题。

Python线程安全锁机制修改时间：2026-08-03 16:36:34