Python ThreadPoolExecutor多线程编程如何避免特定代码段竞态条件

针对非线程安全操作的同步需求，最高效的实现方式是细粒度互斥锁，仅对存在线程安全风险的临界区代码加锁，锁外的业务逻辑依然可以完全并行执行，不会因为同步丢失多线程的效率优势。

核心实现原则 #

• 禁止给整个工作函数加锁，否则所有线程会串行执行完整逻辑，完全丧失并行能力
• 锁实例必须是所有工作线程共享的同一个对象，禁止在工作线程内部创建锁（每个线程独立的锁无法起到同步作用）
• 用上下文管理器（with语法）管理锁的获取与释放，避免代码异常时出现死锁
• 临界区代码尽可能精简，仅保留必须操作共享非线程安全资源的逻辑，减少锁竞争耗时
• 多个独立的共享资源（如同时操作共享列表、多个共享文件）分别配置独立的锁，进一步降低锁冲突概率

修正后可运行代码 #

首先修复你原有代码的两处基础逻辑错误（和竞态无关，但会直接导致运行报错）：

1. 提交任务时传入的是my_list[index]，工作函数内操作的是列表子元素，而非你预期的共享列表
2. 工作函数返回值是布尔值True，后续对返回值做[0]索引会触发类型错误

from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
from typing import List, Any

def my_multithreading_method(
    input_item: Any,
    shared_lock: threading.Lock,
    shared_write_list: List = None,
    shared_file = None
):
    # --------------------------
    # 锁外区域：无共享资源写入的逻辑全部放这里，完全并行执行
    # 可放数据计算、接口调用、数据清洗等无线程安全问题的业务代码
    # --------------------------
    process_result = "Hello"

    # --------------------------
    # 临界区：仅非线程安全的共享资源操作放锁内，串行执行规避竞态
    # --------------------------
    with shared_lock:
        # 共享列表写入
        if shared_write_list is not None:
            shared_write_list.append(process_result)
        # 共享文件写入
        if shared_file is not None:
            shared_file.write(f"{process_result}\n")

    # --------------------------
    # 锁释放后，后续无共享资源操作的逻辑依然可以并行执行
    # --------------------------
    return True

def parallel_execution(my_list: List):
    response_list = []
    future_list = []
    list_count = len(my_list)
    # 初始化所有线程共用的锁实例
    shared_lock = threading.Lock()

    # 如果需要操作共享文件，在此处打开文件句柄传入工作函数即可
    # with open("output.txt", "w", encoding="utf-8") as f:
    with ThreadPoolExecutor(max_workers=list_count) as executor:
        for index in range(list_count):
            future_list.append(
                executor.submit(
                    my_multithreading_method,
                    my_list[index],
                    shared_lock,
                    response_list
                    # f  # 传文件句柄到工作函数
                )
            )
        for future in as_completed(future_list):
            # 直接获取结果即可，共享列表的写入已经在锁保护下完成
            future.result()
    return response_list

补充说明 #

• 普通场景直接用threading.Lock即可，性能优于可重入锁RLock，仅当你明确需要在同一线程内重复获取锁时才用RLock
• 如果仅需要做线程间的结果收集，也可以用线程安全的queue.Queue替代手动加锁的列表，但对于文件操作、自定义非线程安全对象的操作，互斥锁是通用性最强的方案
• 不要依赖CPython的GIL保证操作原子性：比如list.append在CPython中碰巧是原子操作，但跨版本、跨解释器实现（如PyPy）不保证该特性，且文件写入、多步复合操作依然存在竞态，显式加锁是最可靠的方案

内容的提问来源于stack exchange，提问作者Vanshika