Python多进程超时后终止未完成任务并返回已完成结果的实现方案

需求可以实现，原有代码存在两个核心问题：

• 超时后没有调用进程池的terminate()方法，工作进程会持续运行直到所有任务处理完成，导致后台残留运行进程
• starmap_async返回的是批量任务的聚合结果对象，必须等全部任务执行完毕才能通过get()获取结果，无法单独读取已完成的单任务返回值

实现思路 #

• 逐个提交任务并按顺序保存每个任务的AsyncResult对象，方便后续逐个判断任务执行状态
• 超时后直接终止进程池，杀死所有运行中进程，未调度的排队任务也会直接丢弃
• 遍历所有任务的结果对象，执行成功的取返回值，未执行完成的统一填充None，最终结果顺序和参数提交顺序完全一致

完整可运行代码 #

import multiprocessing
import time

def func(x):
    time.sleep(1)
    return x * 2

def main():
    # 配置项：超时时间/工作进程数/任务总数
    timeout = 5
    process_num = 3
    task_total = 100

    pool = multiprocessing.Pool(processes=process_num)
    # 按参数顺序提交任务，保存每个任务的异步结果对象
    async_result_list = [pool.apply_async(func, args=(i,)) for i in range(1, task_total + 1)]
    # 等待超时
    time.sleep(timeout)
    # 终止所有进程，回收资源
    pool.terminate()
    pool.join()

    # 组装最终结果
    final_res = []
    for res in async_result_list:
        # 任务执行完成且无异常则取返回值，否则返回None
        if res.ready() and res.successful():
            final_res.append(res.get())
        else:
            final_res.append(None)
    
    # 测试输出：前15位结果会是 [2,4,6,...,28,30, None, None, None]（3进程5秒共完成15个任务）
    print(final_res[:15])
    return final_res

if __name__ == "__main__":
    main()

注意事项 #

如果你的任务存在打开文件、建立数据库连接等需要手动释放的资源操作，建议在任务函数内加信号处理逻辑做优雅退出，避免强制杀进程导致资源泄漏。纯计算类任务可直接使用上述实现。

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