Python如何使用multiprocessing并行执行多函数并获取返回值

原有多进程代码问题说明 #

• 传参错误：getnpx、getnpx2仅需3个入参，原代码多传了certper，且入参age未对齐单进程逻辑，需改为age + int(certper)
• 缺少子进程返回值接收逻辑，导致npx、npx2变量未定义

推荐实现：使用multiprocessing.Pool（代码简洁） #

通过进程池的异步提交方法可以直接获取子进程返回值，无需手动维护通信管道，完整可运行代码如下：

import numpy as np
import time
import multiprocessing

# 原有三个计算函数
def getnpx(mt, age, interest):
    val = 1
    initval = 1
    for i in range(age, 55):
        val = val * mt[i]
        intval = val / (1 + interest) ** (i + 1 - age)
        initval = initval + intval
    return initval

def getnpx2(mt, age, interest):
    val = mt[age]
    initval = 1
    for i in range(age + 2, 55):
        val *= mt[i - 1]
        if mt[age]==0:
            intval =0
        else:
            intval = val / (1 + interest) ** (i - age - 1) / mt[age]
        initval = initval + intval
    return initval

def getnpxtocert(mt, age, maxvalue):
    val = mt[age]
    for i in range(age + 1, min(maxvalue, 7)):
        val = val * mt[i]
    return val

# 补全后的多进程主函数
def calcannfactprelim_v(pval, age, intrate, certper):
    calc_age = age + int(certper)
    # 创建2进程池
    with multiprocessing.Pool(processes=2) as pool:
        # 异步提交两个计算任务
        res_npx = pool.apply_async(getnpx, args=(pval, calc_age, intrate))
        res_npx2 = pool.apply_async(getnpx2, args=(pval, calc_age, intrate))
        # 获取返回值
        npx = res_npx.get()
        npx2 = res_npx2.get()
    
    if certper == 0:
        index = 1
        index2 = pval[calc_age]
    else:
        index = getnpxtocert(pval, age, age + int(certper)) 
        index2 = getnpxtocert(pval, age, age + int(certper) + 1)

    return index * npx + index2 * npx2

# 单进程基准函数
def calcannfactprelim(pval, age, intrate, certper):
    npx = getnpx(pval, age + int(certper), intrate)
    npx2 = getnpx2(pval, age + int(certper), intrate)
    if certper == 0:
        index = 1
        index2 = pval[age + int(certper)]
    else:
        index = getnpxtocert(pval, age, age + int(certper)) 
        index2 = getnpxtocert(pval, age, age + int(certper) + 1)
    return index * npx + index2 * npx2

if __name__ == "__main__":
    # 测试参数
    pval = np.array([0.000291,0.00027,0.000257,0.000294,0.000325,0.00035,0.000371,0.000388,0.000402,0.000414,0.000425,0.000437,0.011016,0.012251,0.013657,0.015233,0.016979,0.018891,0.020967,0.023209,0.025644,0.028304,0.03122,0.034425,0.037948,0.041812,0.046037,0.050643,0.055651,0.06108,0.066948,0.073275,0.080076,0.08737,0.095169,0.103455,0.112208,0.121402,0.131017,0.14103,0.151422,0.162179,0.173279,0.184706,0.196946,0.210484,0.225806,0.243398,0.263745,0.287334,0.314649,0.346177,0.382403,0.423813,0.470893])
    age=3
    intrate=0.04
    certper=1

    # 单进程测试
    start = time.time()
    res_single = calcannfactprelim(pval, age, intrate, certper)
    print(f"单进程结果：{res_single}，耗时：{time.time()-start:.6f}s")

    # 多进程测试
    start = time.time()
    res_multi = calcannfactprelim_v(pval, age, intrate, certper)
    print(f"多进程结果：{res_multi}，耗时：{time.time()-start:.6f}s")

    # 结果校验
    assert abs(res_single - res_multi) < 1e-15, "结果不一致"
    print("结果校验通过")

补充说明 #

当前示例中计算量极小，多进程的启动开销会大于并行收益，只有当实际业务中getnpx、getnpx2的循环量级大幅提升后，多进程版本才会体现出性能优势。

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