Python concurrent.futures.Future.add_done_callback 如何整理异步结果：回调线程、异常读取与关闭时机

批量调用外部服务时，主线程往往只负责提交任务，真正的结果要等工作线程陆续回来。把归档逻辑塞进 Future.add_done_callback 可以让每个任务完成就收口，但回调不是“自动吞掉异常”的魔法：它需要和 future.result()、future.exception() 以及 ThreadPoolExecutor.shutdown() 配合，才能知道任务到底成功、失败还是被取消。

最稳妥的做法是：回调只负责记录完成状态，先用 future.cancelled() 区分取消，再用 future.result() 读取结果或异常；所有任务提交完后，让 ThreadPoolExecutor 在明确的关闭点等待收尾。

先把 Future 的三种完成状态分开

一个 Future 不是只有成功和失败两种结果。任务可能正常返回，也可能抛异常，或者在真正运行前被取消。回调触发时，这三种情况都会到达同一个入口，所以第一行判断应该是状态，而不是直接打印 future.result()。

from concurrent.futures import ThreadPoolExecutor

def fetch_piece(piece_id):
    if piece_id == 2:
        raise ValueError("bad piece")
    return f"piece-{piece_id}"

def collect_done(future):
    if future.cancelled():
        print("cancelled")
        return
    try:
        value = future.result()
    except Exception as exc:
        print(f"failed: {exc}")
    else:
        print(f"done: {value}")

with ThreadPoolExecutor(max_workers=2) as pool:
    futures = []
    for piece_id in range(3):
        future = pool.submit(fetch_piece, piece_id)
        future.add_done_callback(collect_done)
        futures.append(future)

这里的控制流是 Future 完成后进入 collect_done，先经过 future.cancelled()，未取消才调用 future.result()。第 2 个任务的异常会在 result() 处重新抛出，再由回调自己的 except 分支转成完成记录。

回调线程不是固定的主线程

add_done_callback 的回调由创建这些 Future 的进程中的线程执行。在线程池场景里，任务完成后通常由工作线程直接调用回调；如果 Future 在注册前已经完成，注册动作又可能立即触发回调。这个细节会影响共享列表、日志上下文和非线程安全对象的使用。

因此，回调更适合做轻量的状态归档。例如只把结果放入线程安全队列，或者写入一个带锁的计数器。不要在回调中再做长时间网络请求，也不要假设它一定运行在提交任务的线程。

from concurrent.futures import ThreadPoolExecutor
from threading import Lock

completed = []
completed_lock = Lock()

def record_done(future):
    if future.cancelled():
        state = "cancelled"
    else:
        error = future.exception()
        state = "failed" if error else "succeeded"
    with completed_lock:
        completed.append(state)

这段代码把 future.exception() 当作状态探针：成功时返回 None，任务失败时返回异常对象。真正需要返回值时，仍然应该在主流程或专门的结果收集处调用 future.result()。

异常应该在哪里被读取

回调函数自己抛出的普通 Exception 会被记录并忽略，这意味着“回调失败”不一定会让提交任务的主流程失败。更可靠的做法是把异常读取放在明确的结果汇总阶段：

errors = []
for future in futures:
    try:
        future.result()
    except Exception as exc:
        errors.append(exc)

if errors:
    raise RuntimeError(f"{len(errors)} tasks failed") from errors[0]

回调可以提前写入“已完成”事件，但最终是否成功，仍由 future.result() 的读取结果决定。若只依赖回调里的打印语句，任务失败很容易变成日志里一行不起眼的文本。

关闭 Executor 时要留出收尾边界

使用 with ThreadPoolExecutor(...) 时，离开代码块会调用关闭逻辑并等待已提交任务完成。手动管理时则应把“停止提交”和“等待读取”分成两个动作：先不再调用 submit()，再调用 shutdown(wait=True)，最后汇总每个 Future。

pool = ThreadPoolExecutor(max_workers=4)
futures = []
try:
    for item in items:
        futures.append(pool.submit(process_item, item))
finally:
    pool.shutdown(wait=True)

for future in futures:
    print(future.result())

如果任务还可能失败，最后的 future.result() 要放在自己的异常处理边界内。不要把 shutdown() 当成成功确认；它只负责等待执行器收尾，不能替你判断每个任务的业务结果。

三个容易混淆的判断

常见问题

Future 已经完成后再注册回调，会漏掉通知吗？

不会。官方文档规定，Future 已完成或已取消时注册回调，会立即调用它；代码仍应避免在回调里做不可预期的重活。

add_done_callback 能替代 future.result() 吗？

不能。回调适合做事件通知，result() 才是读取返回值并重新抛出任务异常的明确接口。

future.exception() 会把失败异常抛出来吗？

任务已完成时，它返回异常对象；任务成功时返回 None。如果 Future 还没完成，调用它会等待，取消状态则会抛出取消异常。

把完成通知和结果确认拆开

这个模式的价值不在于少写几行代码，而在于把两个问题分开：回调负责“什么时候结束”，结果收集负责“结束得是否正确”。当线程池规模扩大、失败需要重试或结果要落库时，这个边界比在回调里堆业务逻辑更容易测试，也更容易在关闭前确认没有遗漏。