Python多进程实现不同执行时长CPU密集型函数并行及参数更新问询

现有代码存在三个核心问题导致无法达到预期效果：

1. 调用pool.map_async时直接传入了my_class.execute(xxx)的执行结果，这部分逻辑实际是在主进程同步运行的，根本没有提交到子进程，完全达不到并行效果
2. 多进程场景下不同进程的对象内存空间独立，子进程中修改类实例的属性不会同步到主进程，update生成的新参数根本无法被后续的execute读取
3. 每次触发update时新建销毁进程池的开销极大，会严重打乱execute的执行时序，无法保证1s内完成10次execute的要求

我们采用共享参数空间+独立后台update进程的方案实现需求，主进程全程跑execute，后台update进程到预设时间点自动执行更新，更新完成后共享参数自动生效，全程无阻塞：

from multiprocessing import Process, Manager
import time
import numpy as np

def update_process(shared_params, update_input1, update_input2, update_time_set):
    """后台独立运行的update进程，到点自动更新参数"""
    start_time = time.time()
    while True:
        current_runtime = int(time.time() - start_time)
        # 到达预设更新时间点时执行更新
        if current_runtime in update_time_set:
            print(f'starting update function for time: {current_runtime}')
            time.sleep(1) # 模拟1s耗时的CPU密集更新逻辑
            # 更新共享参数
            shared_params.p1 += update_input1
            shared_params.p2 += update_input2
            print(f'exiting update function, new params: p1={shared_params.p1}, p2={shared_params.p2}')
            # 避免同一秒重复触发更新
            time.sleep(1.1)
        time.sleep(0.01) # 降低空转CPU占用

def execute(input1, input2, shared_params):
    """execute函数，每次读取最新的共享参数计算"""
    print(f'starting execute function, input1={input1}, input2={input2}')
    time.sleep(0.1) # 模拟0.1s耗时的CPU密集计算逻辑
    output = shared_params.p1 * input1 + shared_params.p2 + input2
    print('exiting execute function')
    return output

if __name__ == "__main__":
    # 初始化配置
    update_input1 = 3
    update_input2 = 4
    input1 = 0
    input2 = 1
    # 转换为集合提高查询效率
    update_time = set(np.array([10, 100, 15000, 200000], dtype=int))
    total_runtime = 30 # 测试用可改小，生产环境可按需调整为大数值

    # 创建共享参数空间
    manager = Manager()
    shared_params = manager.Namespace()
    shared_params.p1 = 1 # 初始参数1
    shared_params.p2 = 2 # 初始参数2

    # 启动后台update守护进程
    update_proc = Process(
        target=update_process,
        args=(shared_params, update_input1, update_input2, update_time),
        daemon=True
    )
    update_proc.start()

    # 主进程循环执行execute
    current_count = 0
    while current_count < total_runtime * 10: # 按execute每0.1s一次计算总运行次数
        output = execute(input1, input2, shared_params)
        # 此处可添加output的后续处理逻辑
        # 递增输入参数
        input1 +=1
        input2 +=2
        current_count +=1

核心设计说明 #

• 用multiprocessing.Manager创建共享参数命名空间，进程间修改全局可见，如果参数是复杂结构体可加Lock保证读写安全
• update进程作为守护进程后台运行，不会阻塞主进程的execute循环，update运行的1s内主进程可以正常跑完10次execute
• 所有进程只初始化一次，没有重复创建销毁的开销，时序控制精度高
• 当update执行完成更新共享参数后，后续的execute会自动读取新参数运行，完全符合预期要求

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