Python asyncio.Future 怎样避免重复设置结果触发 InvalidStateError

如果一个低层回调把结果交给 asyncio.Future，但成功回调、超时回调或取消清理又同时尝试结束它，第二次调用 set_result() 或 set_exception() 就会抛出 InvalidStateError。解决思路不是捕获后忽略，而是让所有完成入口在提交结果前共享一次状态判断。

官方文档：https://docs.python.org/3/library/asyncio-future.html

本文只讨论 Future 的完成状态，不扩展到 TaskGroup 的异常聚合。下面的代码片段用于说明结构和边界，图示也是静态说明图，不代表本机运行截图。

为什么第二次 set_result 会触发 InvalidStateError

Future 只有一个终态入口：它可以得到结果、得到异常，或者被取消。官方文档中的 done() 会把这三类状态都视为“已完成”；而 set_result() 和 set_exception() 只负责把尚未完成的 Future 推入终态。

import asyncio

async def duplicate_completion():
    # 用当前事件循环创建 Future，避免手动绑定过时的 loop 参数。
    loop = asyncio.get_running_loop()
    fut = loop.create_future()

    # 第一次完成已经写入结果，Future 进入 done 状态。
    fut.set_result("first")
    # 第二次完成没有新的状态可写，因此会触发 InvalidStateError。
    fut.set_result("second")

这类问题通常不是 set_result 本身随机失败，而是同一个 Future 有两个生产者：例如 I/O 回调负责成功、定时器负责超时，清理分支又负责取消。只要其中一个先完成，其他分支就不能再改写结果。

用 done 把完成动作收口到一个入口

最小修复是把 set_result 与 set_exception 都放进同一个函数，并在写入前调用 done()。返回布尔值比静默捕获异常更有用：调用方可以记录“本次结果被谁先提交”，也可以在重复回调时直接结束清理。

def complete_once(fut, *, value=None, error=None):
    # done() 同时覆盖已有结果、已有异常和取消，避免只检查 cancelled()。
    if fut.done():
        return False

    # error 不为空时提交异常，否则提交正常结果；两条路径只能选一条。
    if error is None:
        fut.set_result(value)
    else:
        fut.set_exception(error)
    return True

这个判断适合多个回调都在同一个事件循环线程中执行的场景。事件循环一次只运行一个同步回调，因此第一个回调完成 Future 后，后续回调再次进入函数就会读到 done() == True。不要把 cancelled() 当成完整判断，因为“已经有结果”和“已经有异常”都不是 cancelled。

取消和跨线程回调要在哪一层处理

取消也是终态。超时分支调用 fut.cancel() 后，I/O 成功回调仍可能晚到；它应该先检查 done()，而不是继续写入结果。若结果来自另一个线程，不能直接操作 asyncio Future；应使用事件循环的线程安全调度方法，把最终判断放回 Future 所属的循环线程。

def publish_from_worker(loop, fut, value):
    # 其他线程只负责把提交动作排回事件循环，不直接改 Future。
    def commit():
        # 回调真正执行时再检查，覆盖超时或取消先到的情况。
        if fut.done():
            return
        fut.set_result(value)

    # call_soon_threadsafe 是跨线程提交回调的入口。
    loop.call_soon_threadsafe(commit)

这里的关键是“延迟到循环线程后再判断”。如果应用正在关闭，调度到已关闭的循环可能抛出 RuntimeError，这属于生命周期问题，应由上层停止工作线程或忽略关闭阶段的提交，而不是再次直接调用 set_result。

排查重复完成时看这四个位置

先搜索同一个 Future 的所有写入点，特别是 set_result、set_exception、cancel 和超时处理。然后确认是否存在两个独立生产者：正常回调与超时回调、成功与失败回调、业务返回与关闭清理。最后把日志放在统一入口的判断前后，记录 Future 标识、来源和 done() 状态，不要只在异常捕获处打印。

一句话记忆：Future 的结果只能写一次；结果、异常和取消都是终态；跨线程只投递回调，最终的 done() 判断必须贴着写入动作。

常见追问

只判断 fut.cancelled() 可以吗？

不够。Future 已经有正常结果或异常时，cancelled() 仍然是 False，但它同样不能再次设置结果。需要用 done() 覆盖全部终态。

捕获 InvalidStateError 后继续流程可以吗？

不建议把异常当作并发控制。捕获只能隐藏重复生产者，无法说明最终结果来自哪条路径。统一完成入口并返回是否提交成功，排查和监控都会更清楚。