Python multiprocessing imap如何捕获超时进程并终止避免重复执行

问题根源 #

原代码触发超时后无限运行的核心原因：

• imap 按输入顺序提交、返回任务结果，调用 next(timeout) 抛出超时仅代表主线程停止等待结果，对应子进程的任务仍会继续运行，占用进程池资源
• 示例中进程池大小为1，超时任务占住唯一的工作进程后，后续所有任务都无法执行，下一次调用 next 仍会等待该超时任务返回，最终导致卡住。

解决代码（满足imap使用要求） #

import time
import multiprocessing as mp

def a_func(x):
    print(x)
    if x:
        return x
    # 触发超时的休眠逻辑
    else:
        time.sleep(2.0)
        return x

if __name__ == "__main__":
    solutions = []
    # 输入总和为4，预期最终solutions长度为4
    inputs = [1, 1, 0, 1, 1, 0]
    ctx = mp.get_context("spawn")

    with ctx.Pool(1) as pool:
        futures_res = pool.imap(a_func, inputs)
        for _ in inputs:
            try:
                res = futures_res.next(timeout=0.1)
                solutions.append(res)
            except mp.context.TimeoutError:
                # 终止所有正在运行的子进程，杀掉超时任务
                for p in pool._pool:
                    if p.is_alive():
                        p.terminate()
                        p.join()
                # 重新拉起工作进程，供后续任务使用
                pool._repopulate_pool()
                # 可自定义追加超时任务的默认返回值，这里直接丢弃不追加
                print("超时任务已丢弃")
    
    print(len(solutions))
    print(solutions)

实现说明 #

• 捕获超时异常后直接终止当前运行的子进程，彻底停止超时任务的执行
• 调用 _repopulate_pool 重新生成工作进程，不影响后续输入任务的正常执行
• 超时任务不会被重新提交到进程池，完全避免重复执行
• 全程使用 imap 实现，符合需求要求

注意事项 #

示例中用到的 _pool 属性和 _repopulate_pool 方法为进程池的内置非公开接口，Python 3所有正式稳定版本均支持该实现，无需担心通用场景下的兼容性问题。

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