Python multiprocessing跨平台报错、Queue使用及append线程安全咨询

问题根因分析 #

• Windows下WinError 6 句柄无效、Linux下卡在queue.join()的核心原因有两个：
  1. multiprocessing.Queue/JoinableQueue的join()方法逻辑是阻塞直到队列中所有任务被标记为处理完成，要求消费者每次调用get()取到任务、处理完成后必须调用task_done()通知队列。原代码消费者逻辑中完全没有调用task_done()，导致join()永久阻塞。
  2. Windows下多进程默认使用spawn启动模式，会重新导入主模块代码，如果没有if __name__ == '__main__':保护块，会递归启动子进程触发句柄继承错误；同时设置daemon=True的子进程会在主进程退出时被强制终止，此时子进程持有的队列句柄未被正常释放，就会抛出句柄无效错误。
• list_dict始终为空的原因：多进程间内存是完全隔离的，主进程中定义的全局变量、作为参数传入子进程的普通list对象，子进程拿到的是独立的内存拷贝，子进程对list的修改只会作用在自己的拷贝上，主进程的原始对象不会有任何变化，加global声明也无效，因为global仅对当前进程内的全局变量生效。
• list.append()的安全性结论：
  1. 多线程场景（CPython环境）：list.append()是C实现的原子操作，执行过程中不会释放GIL，因此对同一个list对象的并发append是线程安全的，不会出现数据丢失、损坏。你之前观察到的异常，基本都是因为append之外的逻辑不是原子操作——比如dict1[some_key].append(func2(request(...)))这类写法，request()请求、func2()计算的过程是可以被线程切换打断的，或者你对list做了读取长度后按下标赋值、遍历中修改这类非原子操作，和append本身无关。
  2. 多进程场景：普通list在进程间不共享，不存在并发操作同一个对象的问题；如果使用multiprocessing.Manager提供的共享list，其append方法内部已经加了进程锁，也是进程安全的。

跨平台可运行修正代码 #

修正点如下：

1. 所有多进程初始化、启动逻辑放在if __name__ == '__main__':保护块下，兼容Windows spawn启动模式
2. 使用JoinableQueue做任务队列，消费者处理完任务后调用task_done()，避免join阻塞
3. 取消daemon强制杀进程的逻辑，处理完所有任务后通过哨兵值（None）通知消费者进程正常退出
4. 处理结果通过专用结果队列传回主进程收集，避免多进程内存隔离导致的数据丢失

import copy
import random
import multiprocessing
from typing import Any


def rand(key: str) -> int:
    if key == 'a':
        return random.randint(0, 5)
    else:
        return random.randint(6, 9)


def run1(task_queue: multiprocessing.JoinableQueue, result_queue: multiprocessing.Queue) -> None:
    while True:
        task_data = task_queue.get()
        # 收到哨兵值，退出进程
        if task_data is None:
            task_queue.task_done()
            break
        [data1] = task_data
        key = list(data1.keys())[0]
        data1[key].append(rand(key))
        result_queue.put(data1)
        print(f"子进程处理结果: {data1}")
        task_queue.task_done()


def qwer(app: Any) -> list:
    # 任务队列、结果队列都在函数内初始化，避免全局变量跨进程异常
    task_queue = multiprocessing.JoinableQueue()
    result_queue = multiprocessing.Queue()
    process_list = []
    # 启动2个消费者进程
    for _ in range(2):
        process = multiprocessing.Process(target=run1, args=(task_queue, result_queue))
        process_list.append(process)
        app.list_process1.append(process)
        app.list_process_names1.append(process.name)
        process.start()

    # 构造测试数据
    list_data = []
    for i in range(10):
        if i % 2 == 0:
            list_data.append({'a': []})
        else:
            list_data.append({'b': []})
    # 放入20个任务
    total_task_num = 20
    for i1 in range(total_task_num):
        idx = i1 % 10
        data2 = copy.deepcopy(list_data[idx])
        task_queue.put([data2])

    # 等待所有任务处理完成
    task_queue.join()
    # 给每个消费者发终止哨兵
    for _ in range(2):
        task_queue.put(None)
    task_queue.join()
    # 等待所有子进程正常退出
    for p in process_list:
        p.join()

    # 从结果队列收集所有处理后的数据
    list_dict = []
    while not result_queue.empty():
        list_dict.append(result_queue.get())

    # 校验结果
    for di in list_dict:
        key = list(di.keys())[0]
        if key == 'a':
            if di['a'][0] not in range(0, 6):
                print(di, False)
        else:
            if di['b'][0] not in range(6, 10):
                print(di, False)
    return list_dict


if __name__ == '__main__':
    # 测试用的模拟app对象，替换成你实际的app实例即可
    class MockApp:
        def __init__(self):
            self.list_process1 = []
            self.list_process_names1 = []
    test_app = MockApp()
    final_result = qwer(test_app)
    print(f"最终收集到结果数量: {len(final_result)}")

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