Python多进程multiprocessing使用map_async终止Pool后如何获取AsyncResult已完成部分结果

阻塞根因 #

你遇到的get()阻塞问题，本质是map_async返回的AsyncResult对象的工作机制决定的：

• AsyncResult会一直等待进程池发来的全部任务完成的信号，才会解除阻塞、返回完整结果
• 一旦调用pool.terminate()，进程池的内部通信管道会被强制断开，永远不会发送那个“全部任务完成”的信号，所以无超时参数的get()会永久挂起
• 原生AsyncResult没有对外暴露读取“部分已完成结果”的接口，就算主进程已经通过管道收到了部分回传的结果，也没有合法方法直接读取，这就是为什么直接调用get()拿不到部分结果的核心原因。

能否基于map_async实现需求 #

可以完全基于map_async的框架实现，不需要切换到apply_async的逐任务提交逻辑，但需要绕开原生AsyncResult的结果收集管道，自主追踪任务完成状态。

具体实现步骤 #

• 引入进程安全的共享存储（比如multiprocessing.Manager托管的列表），用来独立存储每个任务的计算结果，和Pool自带的结果管道完全解耦
• 给每个任务绑定固定的任务索引，worker执行完当前计算后，第一时间把结果写入共享存储的对应位置，不依赖主进程的统一收集
• 触发terminate()终止进程池后，直接从共享存储中过滤掉未完成的空值，得到所有已完成任务的结果，全程不调用AsyncResult.get()，从根源上避免阻塞

改造后可运行的示例代码 #

import multiprocessing
import time

def example_run(i, shared_result, task_index):
    time.sleep(0.7)
    # 任务执行完成后，直接将结果写入共享存储的对应索引位置
    shared_result[task_index] = i
    return i

if __name__ == '__main__':
    terminate = False
    # 初始化进程安全的共享列表，长度和总任务数保持一致
    manager = multiprocessing.Manager()
    shared_result = manager.list([None] * 100)
    
    pool = multiprocessing.Pool(10)
    # 给每个任务绑定固定索引，确保结果顺序和原map逻辑一致
    task_args = [(i, shared_result, i) for i in range(100)]
    result_async = pool.map_async(example_run, task_args)
    
    i = 0
    while True:
        time.sleep(1.0)
        if i == 70:
            terminate = True
        print(result_async.ready(), terminate)
        if result_async.ready():
            break
        elif terminate:
            pool.terminate()
            break
        i += 10
    
    # 直接从共享存储读取已完成的结果，跳过阻塞式get()调用
    partial_result = [res for res in shared_result if res is not None]
    print(partial_result)
    pool.join()

关键注意事项 #

• map_async自带的单次全局回调，只会在所有任务全部执行完成后触发，不是每个任务完成时就调用，无法用来追踪中途的已完成结果
• 调用terminate()终止进程池后，不需要再执行pool.close()，因为terminate()已经强制终止了所有工作进程，再调用close()反而可能引发资源回收的异常
• 因为共享存储是按照任务索引存储结果的，后续读取时可以直接按索引遍历，拿到的结果顺序和正常map()返回的顺序完全一致，不需要做额外的乱序整理
• 不建议给get()加短超时来绕开阻塞，超时抛出的异常会直接丢失主进程已经收到的部分结果，完全达不到获取部分结果的目的

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