使用ProcessPoolExecutor传递asyncio队列遇pickle错误的技术咨询

我是asyncio和concurrent.futures的新手，在理解结合asyncio队列与concurrent.futures的复杂代码时，复现了一个极简示例错误：TypeError: cannot pickle '_queue.SimpleQueue' object。当前问题出在尝试向ProcessPoolExecutor传递asyncio队列对象时，我想明确两个问题：

1. 是否可以将asyncio队列用于或传递给多进程？
2. 该问题是源于多进程与asyncio的结合机制，还是与zip函数的使用及pickle序列化有关？

复现代码 #

import os
import asyncio
import traceback
import datetime
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor


def runner(coro_fn, *args):
    loop1 = asyncio.new_event_loop()
    try:
        print(f"PID FN: {os.getpid()}")
        coro = coro_fn(*args)
        asyncio.set_event_loop(loop1)
        ret = loop1.run_until_complete(coro)
        return ret
    finally:
        loop1.close()


async def run_add(nums):
    task1 = asyncio.create_task(add(nums[0]))
    task2 = asyncio.create_task(add(nums[1]))
    await asyncio.gather(task1, task2)


async def add(num: int):
    await asyncio.sleep(20)
    print(f"num: {num}, pid: {os.getpid()}, tm: {datetime.datetime.now()}")


class TestClass:
    def __init__(self):
        self.max_queue_size = 24

    async def initialize(self):

        try:
            loop = asyncio.get_event_loop()
            val1 = [1, 2, 3, 4, 5]
            val2 = [11, 12, 13, 14, 15]
            val3 = [asyncio.Queue(5)]  # Dummy part to reproduce error.
            val4 = [asyncio.Queue(5)]  # Dummy part to reproduce error.
            executor = ProcessPoolExecutor()
            # tasks = [executor.submit(run_fn, run_add, (i, j)) for i, j in zip(val1, val2)] # works
            # tasks = [loop.run_in_executor(executor, runner, run_add, (i, j)) for i, j in zip(val1, val2)] # works
            tasks = [loop.run_in_executor(executor, runner, run_add, (i, j)) for i, j in
                     zip(val3, val4)]  # Throws error
            await asyncio.gather(*tasks)
        except Exception as _ex:
            print(traceback.format_exc())
            raise _ex


if __name__ == "__main__":
    test_object = TestClass()
    loop = asyncio.get_event_loop()
    loop.set_default_executor(ThreadPoolExecutor(max_workers=32))
    try:
        loop.run_until_complete(test_object.initialize())
    except Exception as ex:
        print(traceback.format_exc())
        raise ex
    finally:
        loop.close()

问题解答 #

1. 是否可以将asyncio队列用于或传递给多进程？ #

不行。asyncio.Queue是专为单进程内异步任务间通信设计的工具，它依赖当前进程的事件循环，内部使用的_queue.SimpleQueue包含无法被pickle序列化的组件（比如线程锁、事件循环引用）。即使强行绕过序列化，不同进程的事件循环相互独立，传递到子进程的队列也无法和父进程的异步任务联动，完全起不到跨进程通信的作用。

如果需要多进程间通信，应该使用multiprocessing.Queue或multiprocessing.Manager().Queue()这类专门为多进程场景设计的队列实现。

2. 该问题是源于多进程与asyncio的结合机制，还是与zip函数的使用及pickle序列化有关？ #

问题核心是pickle序列化的限制，和zip函数无关。

使用ProcessPoolExecutor时，所有传递给子进程的参数必须经过pickle序列化。asyncio.Queue本身不支持pickle序列化，因此在尝试传递它时触发了TypeError。zip函数只是将队列对象打包成元组，本身不会导致序列化问题——哪怕直接传递单个asyncio.Queue给executor，同样会报错。

多进程与asyncio结合本身是可行的（比如你之前传递整数的代码就能正常运行），问题出在传递了不支持跨进程序列化的asyncio队列对象。

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