Python asyncio.wait_for 超时后任务为什么还在跑：取消、shield 与资源回收

接口已经返回超时，后台日志却每隔一秒继续打印 sync tick，这是 Python asyncio 里很容易踩坑的常见问题。asyncio.wait_for() 超时后默认会取消它等待的任务，但被保护的任务、没有留存引用的任务，以及没有在 finally 中收尾的资源，都可能造成“超时没生效，任务还在后台跑”的假象。

先复现：超时异常和后台日志为什么会同时出现

下面的示例任务每秒打印一次心跳，外层只等待2.2秒。把任务对象显式保存下来，就能在超时后直接核验它的状态：是运行结束、已经被取消，还是仍然挂在事件循环里持续执行。

import asyncio

async def worker():
    try:
        for step in range(6):
            print("sync tick", step)
            await asyncio.sleep(1)
        return "finished"
    finally:
        print("worker cleanup")

async def main():
    task = asyncio.create_task(worker(), name="sync-worker")
    try:
        result = await asyncio.wait_for(task, timeout=2.2)
        print(result)
    except asyncio.TimeoutError:
        print("request timeout")
        print("done=", task.done(), "cancelled=", task.cancelled())

asyncio.run(main())

这段程序通常只会打印两次左右的心跳，然后输出 worker cleanup。TimeoutError 是外层等待拿到的结果，CancelledError 会在 worker 内部触发，进入 finally 执行清理逻辑后任务才真正结束。这两个节点不是同一个观测点，很容易给人造成“超时后任务还在跑”的错觉。

最小可用写法：让 wait_for 的取消边界可验证

执行时长可控的短任务可以直接交给 wait_for 托管。要注意别把超时分支写成静默吞掉所有异常的逻辑，也不要抛出超时异常后就直接认定任务已经完全结束、不需要做后续校验。

async def run_with_timeout(coro, seconds):
    task = asyncio.create_task(coro, name="bounded-job")
    try:
        return await asyncio.wait_for(task, timeout=seconds)
    except asyncio.TimeoutError:
        # wait_for 已经发起取消；这里做业务层记录即可
        print("timeout:", task.get_name())
        raise
    finally:
        if not task.done():
            task.cancel()
        # 把取消处理完，避免留下未回收任务
        if not task.done():
            try:
                await task
            except asyncio.CancelledError:
                pass

绝大多数场景下，wait_for 返回时任务已经走完完整的取消流程，所以 finally 里的二次检查不会产生多余开销。保留这段逻辑的意义是明确划定边界：无论内部协程后续怎么迭代修改，退出前都要确认没有遗留悬挂任务。

关键边界：shield 不是“取消失效”，而是把生命周期控制权交给调用方

有些操作不应该因为单次前端请求超时就被中断，比如写入审计日志、提交轻量事务或者把已经接收完的文件移动到暂存区。这类场景可以用 asyncio.shield 保护目标任务，但也要接受对应的结果：外层等待直接超时返回，内层被保护的任务会继续运行直到结束。

async def submit_audit():
    await asyncio.sleep(4)
    print("audit committed")

async def main():
    task = asyncio.create_task(submit_audit(), name="audit-commit")
    try:
        await asyncio.wait_for(asyncio.shield(task), timeout=1)
    except asyncio.TimeoutError:
        print("request timeout; audit continues")

    await task
    print("audit done:", task.done())

这里不存在绝对的“不可中断任务”。shield 只是拦截这一次外层发来的取消请求，不让它继续传给 task。如果进程退出、事件循环主动关闭，被保护的任务依然可能被打断；如果要求操作幂等，提交动作还要带上唯一业务键，不能只靠 shield 保证不会重复执行。

资源回收：finally 要覆盖连接、文件和消息消费者

超时控制管的是外层等待的最大时长，资源释放管的是对象的生命周期。很多人会犯的错误是把清理代码写在正常成功分支里，结果超时触发后，数据库连接或者文件句柄还留在池中没有回收。

async def read_job(pool):
    conn = await pool.acquire()
    try:
        return await conn.fetch_one("select id, status from jobs limit 1")
    except asyncio.CancelledError:
        print("job read cancelled")
        raise
    finally:
        await pool.release(conn)

CancelledError 分支里记录完日志要继续向上抛出，不能把任务被取消伪装成正常的业务返回结果。资源释放逻辑统一放在 finally 块里，这样正常返回、普通异常和超时取消三种场景都会经过同一个回收出口。

三个容易踩坑的变体场景

把同一个协程对象重复交给多个等待者

同一个协程对象只能被调度执行一次。需要多个逻辑同时观测它的状态时，要先用 create_task 把它包装成任务对象，再分配给谁等待、谁只查询状态；不然很容易抛出“cannot reuse already awaited coroutine”这类报错。

在取消处理逻辑里做无限等待

任务被取消时本身也会执行预设的清理逻辑。如果 finally 里再次等待一个永不返回的网络操作，wait_for 会一直卡着等取消处理完成，外层设定的超时就不再是硬截止时间。所有清理动作都要配置自己的短超时，确保路径是可中断的。

直接用 all_tasks 代替业务侧的任务登记

asyncio.all_tasks() 适合做调试排查，不适合直接当作业务任务队列来使用。生产环境更可靠的实现是维护一个全局集合，任务创建时登记进去，执行完成后自动移除，在服务关闭流程里逐个取消集合内的任务并等待清理完成。

一段可复现的完整示例

import asyncio

active = set()

def track(coro, name):
    task = asyncio.create_task(coro, name=name)
    active.add(task)
    task.add_done_callback(active.discard)
    return task

async def persist():
    try:
        await asyncio.sleep(3)
        return "saved"
    finally:
        print("persist cleanup")

async def request():
    task = track(persist(), "persist-job")
    try:
        return await asyncio.wait_for(asyncio.shield(task), timeout=0.5)
    except asyncio.TimeoutError:
        return {"status": "accepted", "job": task.get_name()}

async def main():
    result = await request()
    print(result)
    print("active after request:", len(active))
    await asyncio.sleep(3.2)
    print("active after finish:", len(active))

asyncio.run(main())

这个示例把“请求超时返回”和“后台异步收尾”拆成两个明确的独立状态：请求直接返回 accepted，任务集合里暂时多一项任务；后台持久化操作完成后，回调函数会自动把任务从集合里移除。如果业务场景不允许任务在后台继续跑，直接去掉 shield，超时后主动等待任务走完取消收尾流程就行。

相关问题

wait_for 超时后一定会立刻返回吗？

不一定。它会等待被取消对象执行完取消和清理逻辑，所以协程里的收尾代码可能让实际返回时间晚于你设定的超时秒数。

什么场景下应该使用 shield？

只有当后台动作有明确的独立生命周期、可追踪的状态和幂等边界时才适合用。单纯为了“不让请求报超时”就随便加它，通常只会掩盖任务泄漏的问题。

怎么确认超时后是不是真的留下了后台任务？

在测试代码里留存任务引用，检查 done()、cancelled() 和业务登记集合的长度；关闭事件循环前再主动做一次全量取消和等待，多维度的日志远比单条异常信息可信。

总结

asyncio.wait_for 管的是等待者的时限，任务会不会继续运行、资源能不能正常释放，取决于取消传播逻辑、shield 的使用方式和 finally 的异常处理设计。先明确任务是否允许被取消，再选择直接等待还是用保护逻辑保留任务；最后通过业务任务登记和状态校验，把资源回收的逻辑加入常规测试用例。