如何通过multiprocessing.Process获取函数返回值（少用额外函数）

在StackOverflow的《How to get the return value of a function passed to multiprocessing.Process?》帖子里，有不少多进程取值方案：

• vartec和Nico Schlömer提到的共享状态方案，用Value和Array存储数据，但二者支持的对象类型有限，没法兼容sympy、numpy这类复杂对象；
• 看到的Manager类示例缺少初始化语句，无法直接复用。

我的实际场景需要运行如下结构的多进程：

def function(Input):
    Output = computation(Input)
    return Output

p1 = multiprocessing.Process(target=function, args=(input_1,))
p2 = multiprocessing.Process(target=function, args=(input_2,))

p1.start()
p2.start()

p1.join()
p2.join()

或者带索引的版本，返回带标签的结果：

def function(Input, ix):
    Output = computation(Input)
    return [ix, Output]

p1 = multiprocessing.Process(target=function, args=(input_1, 1,))
p2 = multiprocessing.Process(target=function, args=(input_2, 2,))

最终需要按进程启动顺序拿到返回值列表，或者按索引整理结果。

我考虑过几种方案但都有问题：

• 全局列表追加：担心多进程同时写入引发内存冲突或性能损耗；
• 直接用Queue：需要大幅修改原函数，导致函数无法单独正常调用；
• Pool虽然简洁，但需求限定必须使用Process类。

现在想知道：能不能不修改原函数，用Process类运行多组输入并获取返回值？怎么用最少的额外多进程对象实现？

完全可以做到不修改原函数，核心思路是用包装函数承接原函数的返回值，再通过进程安全的队列传递结果，原函数可以保持完全独立。

方案1：原生multiprocessing.Queue（最少额外对象） #

这是最轻量化的实现，只需要一个Queue对象，无需额外服务进程：

import multiprocessing

# 原函数完全不变，支持返回任意可序列化对象
def function(Input):
    import numpy as np
    return np.array([Input, Input*2])

# 包装函数：调用原函数并将结果存入队列
def wrapper(func, args, queue):
    result = func(*args)
    queue.put(result)

if __name__ == "__main__":
    input_1 = 10
    input_2 = 20

    # 初始化进程安全队列
    result_queue = multiprocessing.Queue()

    # 创建进程，将原函数、参数、队列传给包装器
    p1 = multiprocessing.Process(target=wrapper, args=(function, (input_1,), result_queue))
    p2 = multiprocessing.Process(target=wrapper, args=(function, (input_2,), result_queue))

    p1.start()
    p2.start()
    p1.join()
    p2.join()

    # 按进程启动顺序获取结果
    result_1 = result_queue.get()
    result_2 = result_queue.get()
    print("按启动顺序的结果：", [result_1, result_2])

如果是带索引的版本，原函数同样不用改，只需调整参数传递：

def function(Input, ix):
    import numpy as np
    return [ix, np.array([Input, Input*2])]

# 包装器无需修改，直接传递带索引的参数
p1 = multiprocessing.Process(target=wrapper, args=(function, (input_1, 1), result_queue))
p2 = multiprocessing.Process(target=wrapper, args=(function, (input_2, 2), result_queue))

# 取结果后按索引排序
results = [result_queue.get() for _ in range(2)]
results_sorted = sorted(results, key=lambda x: x[0])
print("按索引排序的结果：", results_sorted)

方案2：Manager.Queue（跨进程更稳定） #

如果需要在嵌套子进程等复杂场景下使用，可以用Manager创建队列，本质是通过服务进程实现跨进程共享，用法和原生队列几乎一致：

import multiprocessing

if __name__ == "__main__":
    # 用上下文管理器自动管理Manager生命周期
    with multiprocessing.Manager() as manager:
        result_queue = manager.Queue()

        p1 = multiprocessing.Process(target=wrapper, args=(function, (input_1,), result_queue))
        p2 = multiprocessing.Process(target=wrapper, args=(function, (input_2,), result_queue))

        p1.start()
        p2.start()
        p1.join()
        p2.join()

        results = [result_queue.get() for _ in range(2)]

方案优势 #

• 原函数完全无需修改，依然可以单独调用（比如function(10)正常返回结果）；
• 仅用到一个Queue对象（或加一个Manager），额外多进程对象最少；
• 队列是进程安全的，底层已处理同步逻辑，不用担心写入冲突；
• 支持任意可pickle序列化的Python对象，numpy、sympy等第三方库的复杂对象都兼容。

注意事项 #

• 如果要保证结果和进程启动顺序一致，需按启动顺序调用get()，因为进程执行速度可能不同，队列内的存储顺序不一定等于启动顺序；
• 若原函数返回的对象无法被pickle序列化，需要自定义序列化逻辑，但绝大多数Python对象默认支持pickle。

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