如何在Multiprocessing Process中不使用Queue获取函数返回值

针对你的需求——顺序执行Pytest测试用例且每个测试在独立进程运行，同时不修改现有timer装饰器和do_stats类的前提下获取返回结果，以下是几种可行方案：

方案1：使用multiprocessing.Manager共享对象传递结果 #

利用multiprocessing.Manager创建跨进程可访问的共享容器（如字典），在子进程中执行函数并将结果存入共享容器，父进程直接读取即可，无需修改原函数的return逻辑。

修改后的示例代码：

import time
from multiprocessing import Process, Manager

def timer(func):
    def wrapper(*args, **kwargs):
        t1 = time.time()
        result = func(*args, **kwargs)
        t2 = time.time()
        elapsed = t2 - t1
        return result, elapsed
    return wrapper

class do_stats:
    def _solve(self):
        # 模拟业务逻辑
        return "test_result"

if __name__ == '__main__':
    # 创建Manager共享字典
    with Manager() as manager:
        result_dict = manager.dict()
        # 定义子进程执行的包装函数，将结果存入共享字典
        def run_task():
            stats = do_stats()
            result_dict["data"] = timer(stats._solve)()
        
        p = Process(target=run_task)
        p.start()
        p.join()
        
        # 从共享字典获取结果
        result, elapsed_time = result_dict["data"]
        print(f"结果: {result}, 耗时: {elapsed_time}")

方案2：使用multiprocessing.Pool的apply方法 #

Pool.apply是阻塞式同步调用，会顺序执行任务，配合maxtasksperchild=1可以让每个任务在新进程中运行，且能直接获取函数的返回值，无需额外的结果传递逻辑。

示例代码：

import time
from multiprocessing import Pool

def timer(func):
    def wrapper(*args, **kwargs):
        t1 = time.time()
        result = func(*args, **kwargs)
        t2 = time.time()
        elapsed = t2 - t1
        return result, elapsed
    return wrapper

class do_stats:
    def _solve(self):
        return "test_result"

if __name__ == '__main__':
    # 创建大小为1的进程池，每个进程只执行一个任务后销毁
    with Pool(processes=1, maxtasksperchild=1) as pool:
        stats = do_stats()
        # 同步调用，直接获取返回值
        result, elapsed_time = pool.apply(timer(stats._solve))
        
        print(f"结果: {result}, 耗时: {elapsed_time}")

方案3：基于Pytest插件实现（更贴合测试场景） #

如果是在Pytest框架中批量处理测试用例，推荐使用pytest-forked插件，它可以让每个测试用例在独立的子进程中运行，同时保持Pytest默认的顺序执行逻辑，无需手动编写进程管理代码。

步骤： #

1. 安装插件：

pip install pytest-forked

2. 执行测试时添加--forked参数：

pytest --forked your_test_file.py

3. 在测试用例中正常使用timer装饰器，Pytest会自动捕获测试函数的返回值和执行结果，无需额外处理进程间通信。

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