Python threading模块锁原语详解

Python threading模块的锁原语

threading模块提供多种锁：Lock、RLock、Semaphore、Event、Condition、Barrier。

Lock的基本使用：

import threading

lock = threading.Lock()
shared_counter = 0

def increment():
global shared_counter
for _ in range(100000):
lock.acquire()
shared_counter += 1
lock.release()

threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(shared_counter) # 1000000

acquire/release保护临界区。with lock更安全。

with lock的异常安全：

lock = threading.Lock()
with lock:
shared_counter += 1

with语句自动处理acquire和release。异常时也确保释放锁。

lock.acquire的超时和阻塞控制：

if lock.acquire(timeout=5):
try:
shared_counter += 1
finally:
lock.release()
else:
print("Failed to acquire lock within 5 seconds")

timeout=0时非阻塞尝试。timeout>0等特指定毫秒数。

RLock可重入锁：

lock = threading.RLock()

def recursive(n):
if n <= 0:
return
with lock:
print(f"n={n}, recurse_depth={n}")
recursive(n - 1)

recursive(5) # 正常执行

RLock允许同一线程多次获取。递归锁计数器跟踪获取次数，每次release减少。

Semaphore限制并发：

sem = threading.Semaphore(3)

def limited_task():
with sem:
print(f"Task running, active={threading.active_count()}")
time.sleep(1)

threads = [threading.Thread(target=limited_task) for _ in range(10)]
for t in threads: t.start()

BoundedSemaphore防止超量释放：

sem = threading.BoundedSemaphore(3)
sem.release() # ValueError: Semaphore released too many times

Event让一个线程通知其他线程：

event = threading.Event()

def waiter():
print("Waiting for event...")
event.wait()
print("Got event!")

def setter():
time.sleep(2)
event.set()
print("Event set")

threading.Thread(target=waiter).start()
threading.Thread(target=setter).start()

event.wait()阻塞直到event.set()被调用。event.is_set()检查状态。event.clear()重置。

Condition变量实现生产者消费者：

cv = threading.Condition()
queue = []

def consumer():
with cv:
while not queue:
cv.wait()
item = queue.pop(0)
print(f"Consumed: {item}")

def producer():
with cv:
queue.append("item")
cv.notify()

Barrier同步多个线程：

def worker(barrier, name):
time.sleep(random.random())
print(f"{name} reached barrier")
barrier.wait()
print(f"{name} passed barrier")

barrier = threading.Barrier(3)
threads = [threading.Thread(target=worker, args=(barrier, i)) for i in range(3)]
for t in threads: t.start()

Barrier(3)等待3个线程到达后同时释放。