Python Future.add_done_callback 怎样集中处理线程任务结果

批量提交线程池任务后，很多代码只保存了 Future，却没有统一处理任务完成、取消和异常。add_done_callback() 适合把这段收尾逻辑集中起来：任务结束时回调自动收到对应的 Future，再由回调读取状态并写入结果记录。

官方地址：https://docs.python.org/3/library/concurrent.futures.html

add_done_callback() 只负责在 Future 完成或取消后通知回调；它不会改变任务调度顺序，也不会替你把异常重新抛回提交线程。回调中应先判断取消，再读取异常或结果。

先看懂 add_done_callback 的触发边界

官方文档规定，add_done_callback(fn) 会在 Future 被取消或执行结束后调用 fn(future)。如果注册回调时 Future 已经完成，回调会立即被调用；因此回调函数不能假设“注册之后才会触发”，也不要在里面依赖某个尚未初始化的外部状态。

回调收到的不是任务返回值，而是 Future 对象。这样设计是为了让同一套回调同时处理成功、异常和取消三种结果。最小的读取顺序可以写成下面这样：

from concurrent.futures import Future

def report_done(future: Future) -> None:
    # 取消状态要先判断，否则 exception() 可能再次抛出取消异常。
    if future.cancelled():
        print("任务已取消")
        return

    error = future.exception()
    if error is not None:
        # exception() 返回任务异常对象，不会在这里自动重新抛出。
        print(f"任务失败：{error}")
        return

    # 没有异常时再读取结果，避免把失败任务当成成功处理。
    print(f"任务完成：{future.result()}")

用一个回调统一成功、取消和异常

实际批量处理中，回调通常不应该只打印文本，而应产生结构化记录。下面的例子用线程安全的 queue.Queue 收集完成信息，主线程最后再按记录类型汇总。这样回调只做很短的状态读取和入队，不在工作线程中执行复杂业务。

from concurrent.futures import Future
from queue import Queue

events: Queue[dict] = Queue()

def collect_done(future: Future) -> None:
    # 先记录取消，避免对已取消 Future 调用 result() 或 exception()。
    if future.cancelled():
        events.put({"status": "cancelled"})
        return

    error = future.exception()
    if error is not None:
        # 把异常对象转成记录，主线程可以统一决定是否重试。
        events.put({"status": "failed", "error": repr(error)})
        return

    # 成功结果只在确认没有异常后读取。
    events.put({"status": "ok", "value": future.result()})

这里没有把回调中的异常继续向提交方抛出。若回调自身抛出 Exception，官方文档规定该异常会被记录并忽略，所以回调内部也应保持简单，并在需要时自行捕获记录错误。

把回调接到线程池，并区分提交顺序与完成顺序

为每个 Future 注册同一个回调即可形成统一收口。任务完成快慢取决于输入和线程调度，因此事件队列里的顺序是完成顺序，不是 submit() 的提交顺序；如果业务要求按输入顺序输出，就要额外保存序号。

from concurrent.futures import ThreadPoolExecutor
from queue import Queue

def work(item: int) -> int:
    # 这里代表一个可能成功或失败的阻塞任务。
    if item == 3:
        raise ValueError("演示任务失败")
    return item * 10

events: Queue[dict] = Queue()

def collect(future) -> None:
    # 回调只负责收口；完成顺序不等于提交顺序。
    if future.cancelled():
        events.put({"status": "cancelled"})
        return
    error = future.exception()
    if error is not None:
        events.put({"status": "failed", "error": repr(error)})
        return
    events.put({"status": "ok", "value": future.result()})

with ThreadPoolExecutor(max_workers=3) as executor:
    # 用字典保留输入序号，后续需要原序输出时可以按 key 排序。
    futures = {}
    for index in range(1, 6):
        future = executor.submit(work, index)
        futures[index] = future
        future.add_done_callback(collect)

# 主线程在退出线程池后读取已完成记录，再决定重试或展示结果。
while not events.empty():
    print(events.get())

回调线程、共享状态和最终汇总怎么安排

回调函数会在创建它的进程所属线程中被调用，但不要把它当成固定的主线程回调。若回调要更新共享列表、计数器或日志缓冲区，应选择线程安全容器，或者只把小记录放入队列，让主线程统一处理。不要在回调里再次等待同一个线程池中依赖后续任务的结果，否则容易让线程池出现互相等待。

如果需要严格保持输入顺序，可将序号与 Future 对应保存，主线程取完记录后按序号排序；如果更关心哪个任务先完成，则直接消费队列即可。两种需求不要混用，否则读者会误以为回调天然提供有序结果。

常见问题

add_done_callback 会阻塞提交线程吗？

注册动作本身很短；回调真正执行时可能占用所属进程中的线程。回调应保持轻量，耗时工作继续提交给明确的执行器。

怎样拿到任务异常而不让回调抛出？

先判断 cancelled()，再调用 exception() 取得异常对象，最后把异常转换成日志或结构化记录。

怎样让结果保持提交顺序？

Future 完成回调不保证全局提交顺序。保存输入序号与 Future 的对应关系，主线程汇总后按序号排序即可。