Python async异步代码内部无法捕获KeyboardInterrupt异常问题咨询

KeyboardInterrupt 是Python解释器收到SIGINT（Ctrl+C触发的信号）时，直接抛给当前主线程执行栈的异常。当协程执行到await asyncio.sleep(1)时，实际运行逻辑处于事件循环底层的selector系统调用等待阶段，异常会直接抛在事件循环的调度逻辑中，不会传入正在等待的协程上下文，因此协程内部写的except KeyboardInterrupt永远无法触发。
事件循环收到未捕获的KeyboardInterrupt后，会强制取消所有运行中的协程，这就是你观察到finally块可以正常执行，但异常最终穿透到最外层打印错误栈的原因。如果把异常捕获逻辑移到asyncio.run()外层，此时协程栈已经被事件循环销毁，自然拿不到协程内部的运行状态。

不要依赖asyncio默认的异常传播逻辑处理中断，不同Python小版本对中断的默认处理行为差异很大，最可控的方式是主动注册信号处理函数，收到中断信号时主动在协程上下文内触发取消，所有清理逻辑直接写在业务协程内部，完整保留上下文。
参考实现代码：

import asyncio
import signal

async def business_logic():
    count = 0
    try:
        while True:
            count += 1
            print(count)
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        # 此处可完整访问所有协程内部上下文，执行有序清理
        # 比如关闭数据库连接、断开MQ会话、保存运行状态等
        print("interrupt", count)
    finally:
        print("final count", count)

async def main():
    loop = asyncio.get_running_loop()
    main_task = asyncio.current_task()

    def trigger_shutdown():
        # 信号触发时仅做一件事：主动取消主任务，让取消异常进入协程调用栈
        main_task.cancel()

    # 同时注册Ctrl+C的SIGINT和服务终止的SIGTERM，适配容器/系统服务部署场景
    loop.add_signal_handler(signal.SIGINT, trigger_shutdown)
    loop.add_signal_handler(signal.SIGTERM, trigger_shutdown)

    await business_logic()

if __name__ == "__main__":
    asyncio.run(main())

运行上述代码，在输出第3次计数时按下Ctrl+C，会得到符合预期的输出，不会抛出未捕获的异常栈：

1
2
3
^Cinterrupt 3
final count 3

多任务场景扩展 #

如果程序同时运行多个后台协程（比如消息消费任务、连接保活任务、定时任务），可以在信号触发时收集所有运行中的任务，批量取消后等待所有任务的清理逻辑执行完成再退出，避免半关闭连接、数据落库不完整等问题：

def trigger_shutdown():
    all_tasks = asyncio.all_tasks()
    for task in all_tasks:
        task.cancel()

在main函数中捕获批量取消抛出的CancelledError即可：

async def main():
    loop = asyncio.get_running_loop()
    def trigger_shutdown():
        for task in asyncio.all_tasks():
            task.cancel()
    loop.add_signal_handler(signal.SIGINT, trigger_shutdown)
    loop.add_signal_handler(signal.SIGTERM, trigger_shutdown)
    
    try:
        await asyncio.gather(
            business_logic(),
            # 其他后台协程放这里
        )
    except asyncio.CancelledError:
        # 等待所有任务完成清理
        await asyncio.gather(*asyncio.all_tasks(), return_exceptions=True)

• 信号处理函数不要写重型清理逻辑，信号回调运行在事件循环的专用信号处理线程，重逻辑会阻塞整个事件循环，仅保留取消任务的逻辑即可
• 协程内捕获asyncio.CancelledError完成清理后，如果不需要特殊拦截，建议重新抛出异常，避免影响asyncio的任务取消传播逻辑
• Windows平台对SIGTERM的支持有限，本地Windows调试时可以仅注册SIGINT，Linux/macOS生产环境两个信号都需要注册

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