Python线程场景下对象内存复用引发竞态问题咨询

问题背景 #

• 应用架构：
  • 启动1个任务生成线程，该线程通过含5个工作线程的线程池生成任务，将任务放入FIFO队列
  • 同时启动含20个工作线程的线程池，从队列取出任务执行
• 异常现象：
  • 单任务运行正常，多任务场景下出现数据覆盖等异常
  • 日志显示队列中取出的任务对象id()、内存地址及内部automation id重复，多线程同时访问同一对象引发冲突

内存复用原因 #

Python中自定义对象不会自动复用内存，出现这种问题的核心原因是任务生成逻辑重复引用了同一个对象实例，而非Python主动复用内存：

• 若任务对象在循环/线程外部初始化，仅在内部修改属性后放入队列，所有队列元素会指向同一个内存地址
• 线程池的工作函数若共享外部定义的任务对象，多个生成线程会修改同一实例的属性，最终队列中全是同一对象的引用

强制分配新内存（生成独立对象）的方法 #

1. 每次生成任务时创建新实例 #

这是最直接的解决方案，确保每个任务都是独立的对象：

# 错误：复用同一个对象
task = Task()
for _ in range(task_count):
    task.automation_id = create_new_id()
    task_queue.put(task)

# 正确：每次循环实例化新对象
for _ in range(task_count):
    task = Task()  # 每次创建新对象，分配新内存
    task.automation_id = create_new_id()
    task_queue.put(task)

2. 使用深拷贝生成独立对象 #

如果需要基于某个基础模板生成任务，使用copy.deepcopy()创建完全独立的实例，避免引用共享：

import copy

base_task = Task()  # 基础模板对象
for _ in range(task_count):
    new_task = copy.deepcopy(base_task)  # 深拷贝生成新内存对象
    new_task.automation_id = create_new_id()
    task_queue.put(new_task)

3. 排查线程池任务生成逻辑 #

确认5个任务生成线程的工作函数，是否存在共享任务对象的情况。比如，线程池的任务函数不要引用外部作用域的同一个对象，而是在函数内部每次生成新实例：

from concurrent.futures import ThreadPoolExecutor

def generate_task():
    # 函数内部创建新对象，而非引用外部对象
    task = Task()
    task.automation_id = create_new_id()
    return task

# 任务生成线程池
with ThreadPoolExecutor(max_workers=5) as executor:
    for _ in range(task_count):
        future = executor.submit(generate_task)
        task_queue.put(future.result())

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