本文目录导读：
- 使用
threading.Lock（锁） - 使用
threading.RLock（可重入锁） - 使用
threading.Semaphore（信号量） - 使用
queue.Queue（队列） - 使用
threading.Event（事件） - 使用
with语句自动管理锁 - 原子操作（特定情况下）
- 使用
concurrent.futures模块 - 最佳实践建议
- 错误示例（需要避免）
在Python中避免线程资源竞争（Race Condition）有多种方法,我来详细介绍几种常用方案：
使用threading.Lock（锁）
最基本的解决方案：
import threading
# 共享资源
counter = 0
lock = threading.Lock()
def increment():
global counter
for _ in range(100000):
with lock: # 获取锁，自动释放
counter += 1
# 创建多个线程
threads = []
for _ in range(10):
t = threading.Thread(target=increment)
threads.append(t)
t.start()
for t in threads:
t.join()
print(f"Counter value: {counter}") # 正确结果：1000000
使用threading.RLock（可重入锁）
当同一个线程需要多次获取锁时使用：
import threading
rlock = threading.RLock()
class Counter:
def __init__(self):
self.count = 0
def increment(self):
with rlock: # 第一次获取锁
self._do_increment()
def _do_increment(self):
with rlock: # 第二次获取锁（不会死锁）
self.count += 1
counter = Counter()
# 多个线程调用 counter.increment()
使用threading.Semaphore（信号量）
控制同时访问资源的线程数量：
import threading
import time
# 最多允许3个线程同时访问
semaphore = threading.Semaphore(3)
def worker(thread_id):
with semaphore:
print(f"Thread {thread_id} is working")
time.sleep(2)
print(f"Thread {thread_id} finished")
threads = []
for i in range(10):
t = threading.Thread(target=worker, args=(i,))
threads.append(t)
t.start()
使用queue.Queue（队列）
线程安全的FIFO队列,避免手动管理锁：
import threading
import queue
import time
# 线程安全队列
task_queue = queue.Queue()
result_queue = queue.Queue()
def producer():
for i in range(10):
task_queue.put(i)
print(f"Produced: {i}")
time.sleep(0.1)
def consumer():
while True:
try:
item = task_queue.get(timeout=1)
result = item * 2
result_queue.put(result)
task_queue.task_done()
except queue.Empty:
break
# 创建线程
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)
producer_thread.start()
consumer_thread.start()
producer_thread.join()
consumer_thread.join()
使用threading.Event（事件）
线程间同步,避免条件竞争：
import threading
import time
event = threading.Event()
shared_data = []
def writer():
time.sleep(2)
shared_data.append("Hello")
event.set() # 通知其他线程
def reader():
print("Waiting for data...")
event.wait() # 等待通知
print(f"Received: {shared_data[0]}")
threads = [
threading.Thread(target=writer),
threading.Thread(target=reader)
]
for t in threads:
t.start()
使用with语句自动管理锁
import threading
class ThreadSafeList:
def __init__(self):
self._list = []
self._lock = threading.Lock()
def append(self, item):
with self._lock:
self._list.append(item)
def pop(self):
with self._lock:
return self._list.pop()
safe_list = ThreadSafeList()
原子操作（特定情况下）
使用threading模块的原子操作避免锁：
import threading
# Python 3.8+ 可以使用 atomic
counter = threading.Value('i', 0) # 原子整数
def increment():
with counter.get_lock():
counter.value += 1
# 或者使用 threading 的 atomic 类型
from threading import AtomicInt
counter = AtomicInt(0)
def increment_safely():
counter.add(1)
使用concurrent.futures模块
高级接口,内部处理线程安全：
from concurrent.futures import ThreadPoolExecutor
import threading
shared_result = []
lock = threading.Lock()
def safe_worker(x):
with lock:
shared_result.append(x * 2)
with ThreadPoolExecutor(max_workers=4) as executor:
futures = [executor.submit(safe_worker, i) for i in range(10)]
最佳实践建议
- 尽量使用队列：
queue.Queue、multiprocessing.Queue等 - 最小化锁的范围：只锁定必要的代码
- 避免死锁：使用
with语句或按照固定顺序获取锁 - 考虑使用
concurrent.futures：高级接口更安全 - 使用不可变对象：尽量避免共享可变状态
错误示例（需要避免）
# 错误的做法 - 存在竞争条件
counter = 0
def bad_increment():
global counter
# 没有锁保护
temp = counter
temp += 1
counter = temp # 可能被其他线程覆盖
# 需要改为：
def good_increment():
global counter
with lock:
counter += 1
选择合适的同步机制取决于你的具体需求，最简单的原则是：只要多个线程访问同一个可变对象，就要考虑线程安全。
