Python多进程中共享队列内容不可见问题排查与解决

我在进程A中运行若干协程，在独立进程B中运行一个较重的无界任务，希望进程B将结果发送到队列供进程A消费，但测试发现进程A无法看到进程B放入队列的数据。测试代码如下：

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor


def process__heavy(pipe):
    print("[B] starting...")
    while True:
        print(f"[B] Pipe queue: {pipe.qsize()}")
        pipe.put_nowait(str(time.time()))
        time.sleep(0.5)

async def coroutine__stats(pipe):
    print("[A] starting...")
    while True:
        print(f"[A] Pipe queue: {pipe.qsize()}")
        await asyncio.sleep(1)

  
async def main():
    pipe = asyncio.Queue()
    executor = ProcessPoolExecutor()

    jobs = await asyncio.gather(
        asyncio.get_running_loop().run_in_executor(executor, process__heavy, pipe),
        coroutine__stats(pipe)
    )

    print(f"Finished with result: {jobs.result()}")


if __name__ == '__main__':
    asyncio.run(main())
    print("Bye.")

运行输出：

[A] starting...
[A] Pipe queue: 0
[B] starting...
[B] Pipe queue: 0
[B] Pipe queue: 1
[A] Pipe queue: 0 <--- why zero?
[B] Pipe queue: 2
[B] Pipe queue: 3
[A] Pipe queue: 0 <---
[B] Pipe queue: 4
[B] Pipe queue: 5
[A] Pipe queue: 0 <---
[B] Pipe queue: 6
[B] Pipe queue: 7
[A] Pipe queue: 0
[B] Pipe queue: 8

核心疑问：
为什么进程A看不到进程B放入队列的数据？Python中能否跨进程共享对象，还是仅能在进程退出时返回序列化结果？哪里出错了，创建进程间数据管道的最佳方式是什么？

问题原因 #

• asyncio.Queue不支持跨进程共享：asyncio.Queue是专为单进程内的协程通信设计的内存数据结构，没有实现跨进程同步机制。通过ProcessPoolExecutor传递队列时，会通过pickle序列化复制一份队列副本到子进程内存，两个进程操作的是完全独立的实例，自然看不到对方的修改。
• Python进程无法直接共享普通对象：每个进程拥有独立的内存空间，普通对象不能直接跨进程访问，必须通过专门的IPC（进程间通信）机制传递数据，或使用支持共享内存的同步原语。

解决方法 #

方案1：使用multiprocessing.Queue（推荐） #

multiprocessing.Queue是标准库中专门用于跨进程通信的队列，基于管道和锁实现了进程间安全的数据传递，可与asyncio结合使用：

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Queue


def process__heavy(pipe):
    print("[B] starting...")
    while True:
        print(f"[B] Pipe queue: {pipe.qsize()}")
        pipe.put_nowait(str(time.time()))
        time.sleep(0.5)

async def coroutine__stats(pipe):
    print("[A] starting...")
    loop = asyncio.get_running_loop()
    while True:
        # 用run_in_executor包装阻塞的qsize方法，避免卡断事件循环
        queue_size = await loop.run_in_executor(None, pipe.qsize)
        print(f"[A] Pipe queue: {queue_size}")
        await asyncio.sleep(1)

async def main():
    pipe = Queue()  # 替换为跨进程队列
    executor = ProcessPoolExecutor()

    # 创建异步任务，避免gather因无限循环永久阻塞
    task1 = asyncio.create_task(
        asyncio.get_running_loop().run_in_executor(executor, process__heavy, pipe)
    )
    task2 = asyncio.create_task(coroutine__stats(pipe))
    
    try:
        await asyncio.gather(task1, task2)
    except asyncio.CancelledError:
        pass

if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nBye.")

关键说明：

• multiprocessing.Queue原生支持跨进程通信，父子进程操作的是同一个共享队列（底层通过管道传递数据）。
• 由于multiprocessing.Queue的方法是阻塞的，在协程中必须用loop.run_in_executor包装调用，避免阻塞asyncio事件循环。

方案2：使用multiprocessing.Pipe（轻量备选） #

如果只需要单向通信，multiprocessing.Pipe是更轻量的选择，基于双向管道实现，性能略高于队列：

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Pipe


def process__heavy(conn):
    print("[B] starting...")
    while True:
        conn.send(str(time.time()))
        print(f"[B] Sent data")
        time.sleep(0.5)

async def coroutine__stats(conn):
    print("[A] starting...")
    loop = asyncio.get_running_loop()
    while True:
        # 异步读取管道数据
        data = await loop.run_in_executor(None, conn.recv)
        print(f"[A] Received: {data}")
        await asyncio.sleep(0.1)

async def main():
    parent_conn, child_conn = Pipe()
    executor = ProcessPoolExecutor()

    task1 = asyncio.create_task(
        asyncio.get_running_loop().run_in_executor(executor, process__heavy, child_conn)
    )
    task2 = asyncio.create_task(coroutine__stats(parent_conn))
    
    try:
        await asyncio.gather(task1, task2)
    except asyncio.CancelledError:
        pass

if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nBye.")

关键说明：

• Pipe创建一对连接对象，父进程用parent_conn，子进程用child_conn，支持双向数据传递（示例中用单向发送/接收）。
• 同样需要用run_in_executor包装阻塞的recv方法，避免阻塞事件循环。

关键总结 #

• 禁止用asyncio.Queue做跨进程通信，它仅适合同一进程内的协程间通信。
• 跨进程通信优先选multiprocessing.Queue（安全、易用），轻量场景可选用Pipe。
• 异步代码中调用阻塞的IPC方法时，必须用run_in_executor包装，避免破坏asyncio事件循环的非阻塞特性。

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