Python多进程启动Process报cannot pickle系列错误求助

第一个错误：TypeError: cannot pickle '_thread.lock' object #

原因 #

Windows系统下Python的multiprocessing默认采用spawn方式启动子进程，启动时需要对Process子类的整个实例做序列化（pickle）传递给子进程。你代码里初始化RadioServer时传入的是queue.Queue（线程队列），该队列内部持有不可序列化的线程锁对象，导致序列化失败。

解决方法 #

将线程队列替换为multiprocessing.Queue，后者是专门为跨进程通信设计的，支持序列化，你已经完成了这一步的修正。

第二个错误：AudioCodecContext cannot be converted to a Python object for pickling 及后续EOFError #

原因 #

还是和spawn启动的序列化机制有关：

1. 你在RadioServer的__init__方法或者主进程中提前初始化了依赖av库（aiortc底层依赖的音视频处理库）的对象，比如MediaPlayer、CustomRadioStream实例、音视频编解码器上下文等。这类对象底层是Cython封装的C指针/结构体，完全不支持序列化。
2. 序列化阶段失败后，子进程无法收到完整的序列化数据，加载时就会抛出EOFError: Ran out of input。

解决方法 #

核心原则是：所有不可序列化的对象初始化操作全部挪到run()方法内执行，__init__方法中仅存放可序列化的基础参数

• RadioServer的__init__只传基础配置、multiprocessing.Queue这类支持序列化的对象，不要初始化任何aiortc、av相关的实例，也不要把主进程中创建的av相关对象作为属性传给RadioServer。
• 涉及音视频流、编解码器、自定义流类（比如CustomRadioStream）的初始化全部放在子进程启动后执行的run()方法中完成。
• 主进程和子进程之间仅通过multiprocessing.Queue传递可序列化的纯数据（比如原始音频帧的字节内容、numpy数组、控制指令字符串等），不要传递任何音视频处理相关的实例对象。

修正后的代码结构示例 #

from multiprocessing import Process, freeze_support, Queue # 注意导入的是multiprocessing的Queue
import os
# 其他导入保持不变

class RadioServer(Process):
    def __init__(self, q, host:str, port:int, root_path:str):
        super().__init__()
        # __init__中仅存放可序列化的基础参数
        self.q = q
        self.host = host
        self.port = port
        self.ROOT = root_path
        # 不要在这里初始化pcs、stream_offers这类后续会持有不可序列化对象的列表，也不要提前创建任何av/aiortc相关实例
    
    def run(self):
        # 所有不可序列化的对象全部在run方法内初始化
        self.pcs = []
        self.channels = []
        self.stream_offers = []
        self.requests = []
        
        self.app = web.Application()
        self.app.on_shutdown.append(self.on_shutdown)
        self.app.router.add_get("/", self.index)
        self.app.router.add_get("/radio.js", self.javascript)
        self.app.router.add_get("/jquery-3.5.1.min.js", self.jquery)
        self.app.router.add_post("/offer", self.offer)
        
        threading.Thread(target=self.fill_the_queues).start()
        web.run_app(self.app, access_log=None, host=self.host, port=self.port, ssl_context=None)
    
    # 其他方法保持不变，注意所有涉及aiortc/av的对象创建都在run启动后执行
    def fill_the_queues(self):
        while(True):
            frame = self.q.get()
            for stream_offer in self.stream_offers:
                stream_offer.q.put(frame)

    # 剩下的index、javascript、offer、on_shutdown方法保持原有逻辑即可

if __name__ == "__main__":
    freeze_support()
    q = Queue()
    # 初始化时仅传可序列化的参数
    custom_server_child_process = RadioServer(
        q=q,
        host="192.168.1.20",
        port=8080,
        root_path=os.path.dirname(__file__)
    )
    custom_server_child_process.start()

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