Python multiprocessing如何检测已结束进程并获取返回值

核心思路 #

直接实例化multiprocessing.Process启动进程的写法存在两个明显短板：一是无法直接获取任务函数的返回值，二是批量检测进程存活状态、匹配任务身份的逻辑需要手动实现，重复造轮子。
最优实现方式基于官方内置的进程池+异步任务提交接口完成需求，核心逻辑：

• 初始化固定最大并发数的进程池，避免无限制创建进程打满系统资源
• 用apply_async异步提交任务，该方法非阻塞，提交后立刻返回任务结果句柄
• 维护一个在途任务字典，以自定义的唯一任务ID为键，任务结果句柄为值，实现任务身份和进程任务的绑定
• 主while True循环每次迭代先完成新数据采集、新任务提交，再遍历在途任务字典，非阻塞检测哪些任务已经执行完成，取出返回值做后续处理后，把已完成的任务从字典中移除，避免内存泄漏

完整可运行示例代码 #

import random
import time
import multiprocessing


def process_task(task_id: int, collect_data: float) -> tuple[int, float, str]:
    """模拟数据处理的工作进程函数
    :param task_id: 任务唯一标识，用于识别进程身份
    :param collect_data: 主循环采集到的待处理数据
    :return: 任务ID、处理耗时、处理结果
    """
    cost_time = random.uniform(0.5, 3)
    time.sleep(cost_time)
    process_result = f"数据{collect_data:.2f}处理完成"
    return task_id, cost_time, process_result


if __name__ == '__main__':
    # 初始化进程池，最大同时运行4个进程，可根据机器CPU核数、任务类型调整
    pool = multiprocessing.Pool(processes=4)
    # 存放在途任务：key=任务ID，value=异步任务结果句柄
    running_tasks = dict()
    task_id_counter = 0

    try:
        while True:
            # 1. 模拟采集新数据
            new_collect_data = random.uniform(10, 100)
            task_id_counter += 1
            current_task_id = task_id_counter

            # 2. 提交新任务到进程池，非阻塞
            task_res = pool.apply_async(process_task, args=(current_task_id, new_collect_data))
            running_tasks[current_task_id] = task_res
            print(f"[主循环] 提交新任务ID:{current_task_id}, 待处理数据:{new_collect_data:.2f}, 当前在途任务数:{len(running_tasks)}")

            # 3. 检测已完成的任务，非阻塞遍历
            for tid, res in list(running_tasks.items()):
                # ready()方法返回True代表任务已经执行结束
                if res.ready():
                    # 确认任务结束后get()可立刻拿到返回值，不会阻塞
                    task_id, cost, result = res.get()
                    print(f"[任务完成] 任务ID:{task_id}, 耗时:{cost:.2f}s, 返回结果:{result}")
                    # 从在途字典中删除已完成任务，释放内存
                    del running_tasks[tid]

            # 主循环短暂休眠，避免空转占满CPU
            time.sleep(0.2)

    except KeyboardInterrupt:
        print("\n接收到退出信号，正在关闭进程池...")
        # 关闭进程池，不再接收新任务
        pool.close()
        # 等待所有在途任务执行完成
        pool.join()
        print("程序退出")

关键注意点 #

• 检测任务状态用AsyncResult.ready()是完全非阻塞的，不会卡住主循环的数据采集逻辑
• 必须在确认任务ready()之后再调用get()拿返回值，否则get()会一直阻塞直到任务结束，打断主循环运行节奏
• 遍历在途任务时要遍历字典的副本（示例中用list(running_tasks.items())实现），否则遍历过程中删除字典元素会抛出遍历异常
• 进程池的最大并发数可根据任务类型调整：CPU密集型任务不要超过CPU核心数，IO密集型任务可适当调大
• 自定义任务ID可以根据业务场景替换，比如用采集时间戳、数据批次号等，和任务的绑定关系存在字典中不会混乱
• 程序退出时必须调用进程池的close()和join()方法，避免产生僵尸进程

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