Windows10下Python3.10.9 Multiprocessing功能异常排查求助

Windows和Linux的Python多进程实现机制存在本质差异：

• Linux默认用fork方式启动子进程，子进程会直接继承父进程的内存空间，无需重新导入主模块
• Windows默认用spawn方式启动子进程，会启动全新的Python解释器，重新导入主模块并执行所有模块级代码

你的代码没有对主执行逻辑做保护，Windows下子进程启动时会重复执行y = DedicatedProcessObject(...)和y.test('c')，导致子进程又创建新的子进程，最终触发异常，子进程直接退出（exitcode=1）。

给主执行代码加上if __name__ == '__main__':保护，避免子进程重复执行模块级逻辑，同时增加异常捕获方便调试：

import multiprocessing

def setup_subprocess(
    object_type,
    args: tuple,
    kwargs: dict,
    queue_in: multiprocessing.Queue,
    queue_out: multiprocessing.Queue,
):
    instance = object_type(*args, **kwargs)
    while True:
        try:
            method_name, method_args, method_kwargs = queue_in.get()
            result = getattr(instance, method_name)(
                *method_args,
                **method_kwargs
            )
            queue_out.put(result)
        except Exception as e:
            # 捕获异常并发送到输出队列，主进程可直接抛出
            queue_out.put(e)

class DedicatedProcessObject():
    def __init__(self, object_type, *args, **kwargs):
        self.queue_in = multiprocessing.Queue()
        self.queue_out = multiprocessing.Queue()

        self.process = multiprocessing.Process(
            target=setup_subprocess,
            args=(
                object_type,
                args,
                kwargs,
                self.queue_in,
                self.queue_out
            ),
        )

        self.process.start()

    def __getattr__(self, name):
        def method(*args, **kwargs):
            self.queue_in.put((name, args, kwargs))
            result = self.queue_out.get()
            if isinstance(result, Exception):
                raise result
            return result
        return method

class TestClass():
    def __init__(self, a, b='b'):
        self.a = a
        self.b = b
        
    def test(self, c, d='d'):
        return self.a,self.b,c,d

# 仅主进程执行以下逻辑，子进程启动时不会触发
if __name__ == '__main__':
    x = TestClass('a')
    y = DedicatedProcessObject(TestClass, 'a')

    print(x.test('c'))
    print(y.test('c'))

如果需要更详细的子进程错误信息，可以：

• 启用multiprocessing的日志功能，查看子进程启动和运行的细节：

import logging
multiprocessing.log_to_stderr(logging.DEBUG)

• 给子进程的Process对象设置daemon=True，避免主进程退出后残留僵尸进程

内容的提问来源于stack exchange，提问作者Koen T.