如何防止asyncio.Task被取消 解决优雅关闭任务丢失问题

这个写法从根上就有问题，踩了两个asyncio关闭流程的经典坑：

• 靠asyncio.Task.all_tasks()扫全量任务等完成本身就不可靠：进程收到SIGINT/SIGTERM信号时，asyncio默认的清理逻辑会第一时间给所有运行中任务投递CancelledError，没做取消保护的任务会直接抛错退出，很快就会从任务集合里被移除，你再调用all_tasks()自然拿不到。顺便提一句，Task.all_tasks()早就废弃了，新版本Python直接用asyncio.all_tasks()。你看到任务“消失”本质是默认取消逻辑触发后，任务没做异常捕获，直接抛CancelledError结束，生命周期走完自然不会出现在未完成任务列表里。
• 直接gather全量任务必现死锁：all_tasks()会把当前正在跑的关闭处理任务本身、事件循环的内部系统任务全捞出来，等这些任务结束等于自己等自己，直接卡到超时。

不要依赖运行时全量任务扫描做优雅关闭，正确的实现方案如下：

1. 自行托管必须等待完成的核心任务 #

创建任务时不要扔给事件循环就不管，单独维护一个核心任务集合，任务执行完成后自动从集合移除避免内存泄漏：

import asyncio
from typing import Set

# 可以放在全局或者你的应用上下文里
critical_tasks: Set[asyncio.Task] = set()

def create_critical_task(coro):
    task = asyncio.create_task(coro)
    critical_tasks.add(task)
    # 不管任务是成功、失败还是被取消，执行完就从集合里删掉
    task.add_done_callback(critical_tasks.discard)
    return task

后续所有需要在关闭前保证执行完的任务，统一用create_critical_task创建，不要直接调用asyncio.create_task。

2. 自定义信号处理器，替换默认取消逻辑 #

不要用asyncio自带的默认信号处理，自己注册SIGINT、SIGTERM的处理逻辑，触发退出时先等核心任务跑完，再清理其他非核心任务：

import signal

async def _shutdown_handler(sig, loop):
    # 先摘掉信号处理器，避免重复触发关闭
    for s in (signal.SIGINT, signal.SIGTERM):
        loop.remove_signal_handler(s)
    
    print(f"收到退出信号 {sig.name}，等待核心任务执行完毕...")
    # 等所有核心任务跑完，return_exceptions=True避免单个任务报错打断整体等待
    if critical_tasks:
        await asyncio.gather(*critical_tasks, return_exceptions=True)
    
    print("核心任务全部执行完成，清理剩余非核心任务")
    # 核心任务跑完后，再取消剩下的非核心任务
    remaining_tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    for t in remaining_tasks:
        t.cancel()
    await asyncio.gather(*remaining_tasks, return_exceptions=True)
    
    loop.stop()

# 事件循环启动后调用这个方法注册信号
def setup_shutdown_handlers(loop):
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(
            sig,
            lambda s=sig: asyncio.create_task(_shutdown_handler(s, loop))
        )

3. 给核心任务加取消保护（可选） #

如果你的核心任务哪怕收到取消通知也必须把收尾逻辑做完，可以在任务内部捕获CancelledError，执行完收尾操作再退出：

async def some_critical_operation():
    try:
        # 核心业务逻辑，比如刷日志、提交数据库事务、上报监控
        await asyncio.sleep(3)
        print("核心业务逻辑执行完成")
    except asyncio.CancelledError:
        print("核心任务收到取消通知，执行收尾操作...")
        # 收尾逻辑，比如关闭文件句柄、发完最后一条消息
        await asyncio.sleep(1)
        print("核心任务收尾完成")
        raise # 如果需要保留取消状态可以往上抛，不需要可以不raise

• 如果你用FastAPI、aiohttp这类自带生命周期管理的框架，不用自己写信号处理，直接在框架提供的shutdown生命周期钩子里await asyncio.gather(*critical_tasks, return_exceptions=True)就行，框架会帮你处理信号拦截的逻辑。
• Python 3.11及以上版本已经完全移除了asyncio.Task.all_tasks()这个旧接口，统一用asyncio.all_tasks()获取当前事件循环的任务。
• 永远不要在关闭逻辑里写await asyncio.gather(*asyncio.all_tasks())，100%会死锁。

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