如何在Python派生进程间传递或复制multiprocessing.Queue对象

你遇到的RuntimeError是multiprocessing.Queue的设计限制：原生Queue内部包含不可序列化的同步原语、管道文件描述符状态，只能在子进程创建时通过继承的方式共享，无法通过普通Queue的put/get操作序列化后跨进程传输。

方案1：通过Manager托管队列（最低改造成本） #

multiprocessing.Manager创建的队列是可序列化的代理对象，支持在任意进程间传递，完全兼容原生Queue的API，只需少量修改代码即可实现需求：

import multiprocessing
from multiprocessing.managers import SyncManager

class Example:
  def __init__(self):
    self.rx_queue = multiprocessing.Queue()

  def poll(self):
    queue_to_duplicate = self.rx_queue.get()
    print("received queue?", queue_to_duplicate)
    # 后续可以直接用收到的队列完成双工通信
    queue_to_duplicate.put("response from sub process")

if __name__ == "__main__":
  # 用Manager创建可传递的队列
  manager = SyncManager()
  manager.start()
  queue_to_duplicate = manager.Queue()

  ex = Example()
  ex_proc = multiprocessing.Process(target=ex.poll)
  ex_proc.start()

  ex.rx_queue.put(queue_to_duplicate)
  print(queue_to_duplicate.get())
  ex_proc.join()
  manager.shutdown()

这个方案的优势是不需要接触底层实现，兼容性好；劣势是多了Manager进程的中转开销，性能略低于原生Queue。

方案2：传递文件描述符手动重建队列（高性能原生实现） #

你提到的通过文件描述符复制重建队列的思路完全可行，Linux平台下可以通过multiprocessing.Connection的send_handle/recv_handle方法跨进程传递Queue内部的管道文件描述符，接收端拿到fd后即可重建原生Queue：

import multiprocessing
from multiprocessing.connection import Connection
import os

class Example:
  def __init__(self):
    # 这里用Pipe代替普通Queue方便传递fd，也可以提取Queue内部的_reader/_writer的fd
    self.rx_conn, self.tx_conn = multiprocessing.Pipe()

  def poll(self):
    # 接收主进程传递的Queue读写fd
    read_fd = self.rx_conn.recv_handle()
    write_fd = self.rx_conn.recv_handle()
    # 重建Connection对象
    read_conn = Connection(read_fd)
    write_conn = Connection(write_fd)
    # 手动构建Queue（简化实现，完整实现可参考multiprocessing.Queue源码）
    from multiprocessing.queues import Queue
    rebuilt_queue = Queue(ctx=multiprocessing.get_context())
    rebuilt_queue._reader = read_conn
    rebuilt_queue._writer = write_conn
    rebuilt_queue._closed = False
    print("rebuilt queue success", rebuilt_queue)
    rebuilt_queue.put("response from sub process")

if __name__ == "__main__":
  queue_to_duplicate = multiprocessing.Queue()
  ex = Example()
  ex_proc = multiprocessing.Process(target=ex.poll)
  ex_proc.start()

  # 传递Queue内部的读写文件描述符
  ex.tx_conn.send_handle(queue_to_duplicate._reader.fileno(), os.getpid())
  ex.tx_conn.send_handle(queue_to_duplicate._writer.fileno(), os.getpid())
  print(queue_to_duplicate.get())
  ex_proc.join()

这个方案的优势是完全保留原生Queue的性能，没有额外中转开销；劣势是依赖Unix/Linux平台特性，且用到了Queue的内部私有属性，不同Python版本可能存在兼容性差异。

方案3：预创建队列通过继承共享（最高性能稳定实现） #

如果你的双工通信逻辑是可提前规划的，不需要动态创建传递队列，可以在子进程启动时就把双工需要的所有队列通过Process的args参数传入，进程创建时会自动继承队列，不存在序列化问题：

import multiprocessing

class Example:
  def __init__(self, rx_queue, tx_queue):
    # 直接接收预创建的两个队列完成双工
    self.rx_queue = rx_queue
    self.tx_queue = tx_queue

  def poll(self):
    print("received from main:", self.rx_queue.get())
    self.tx_queue.put("response from sub process")

if __name__ == "__main__":
  # 预创建双工队列
  main_rx = multiprocessing.Queue()
  sub_rx = multiprocessing.Queue()

  ex = Example(sub_rx, main_rx)
  ex_proc = multiprocessing.Process(target=ex.poll)
  ex_proc.start()

  sub_rx.put("request from main")
  print("received from sub:", main_rx.get())
  ex_proc.join()

这个方案无额外开销，兼容性最好，是官方推荐的进程间共享Queue的实现方式。

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