Python multiprocessing.Pool的map_async worker报错导致任务丢失问题排查

• 你观察到工作进程未终止是正常现象：multiprocessing.Pool的工作进程为常驻设计，单个任务执行时抛出的异常会被进程内部的调度逻辑捕获，不会导致工作进程退出，处理完异常后工作进程会继续拉取新的任务块执行。
• 部分任务被跳过的核心原因是Pool的任务分块调度机制：map系列方法会自动将输入的可迭代对象拆分为多个连续的任务块（chunk），每个工作进程一次拉取一整个块的任务执行。如果同一个块内的某一个任务抛出未捕获的异常，该块内剩余未执行的所有任务会被直接丢弃，不会继续调度。你的工作函数在a为3的倍数时主动抛出ValueError且未做捕获，直接触发了该机制。
• 你使用map_async后仅调用wait()方法，该方法仅等待异步任务执行结束，不会抛出工作进程内产生的异常，也不会返回执行结果，你无法感知到任务执行过程中出现的错误。
• 你传入的生成器表达式为一次性消费的可迭代对象，任务块拆分时会直接消费生成器内容，被丢弃的任务没有二次重试的数据源。

方案1：工作函数内捕获异常 #

在工作函数内部捕获所有业务异常，避免异常抛到Pool的调度层，打断同块其他任务的执行，示例如下：

from multiprocessing import Pool
import os

def wrk(a):
    try:
        if a % 3 == 0:
            print(a, os.getpid(), 'GONNA DIE')
            raise ValueError('ERROR')
        else:
            print(a, os.getpid())
    except Exception as e:
        # 按需处理异常，例如记录日志
        pass

if __name__ == '__main__':
    with Pool(processes=3) as pool:
        p = pool.map_async(wrk, range(50))
        p.wait()

方案2：调整分块大小为1 #

设置chunksize=1让每个任务单独作为一个分块，单个任务失败不会影响其他任务的调度，同时调用get()方法在主进程捕获异常：

from multiprocessing import Pool
import os

def wrk(a):
    if a % 3 == 0:
        print(a, os.getpid(), 'GONNA DIE')
        raise ValueError('ERROR')
    else:
        print(a, os.getpid())

if __name__ == '__main__':
    with Pool(processes=3) as pool:
        # 每个任务单独分块
        p = pool.map_async(wrk, range(50), chunksize=1)
        try:
            # get()会将工作进程内的异常抛到主进程
            res = p.get()
        except Exception as e:
            print(f"任务执行异常: {e}")

内容的提问来源于stack exchange，提问作者Sergii V.