Python脚本如何避免线程资源竞争

本文目录导读：

1. 使用threading.Lock（锁）
2. 使用threading.RLock（可重入锁）
3. 使用threading.Semaphore（信号量）
4. 使用queue.Queue（队列）
5. 使用threading.Event（事件）
6. 使用with语句自动管理锁
7. 原子操作（特定情况下）
8. 使用concurrent.futures模块
9. 最佳实践建议
10. 错误示例（需要避免）

在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)]

最佳实践建议

1. 尽量使用队列：queue.Queue、multiprocessing.Queue等
2. 最小化锁的范围：只锁定必要的代码
3. 避免死锁：使用with语句或按照固定顺序获取锁
4. 考虑使用concurrent.futures：高级接口更安全
5. 使用不可变对象：尽量避免共享可变状态

错误示例（需要避免）

# 错误的做法 - 存在竞争条件
counter = 0
def bad_increment():
    global counter
    # 没有锁保护
    temp = counter
    temp += 1
    counter = temp  # 可能被其他线程覆盖
# 需要改为：
def good_increment():
    global counter
    with lock:
        counter += 1

选择合适的同步机制取决于你的具体需求，最简单的原则是：只要多个线程访问同一个可变对象，就要考虑线程安全。