Python中捕获SIGINT需在asyncio.run()前用signal.signal注册同步处理器，再触发异步退出；热退出须停止接收新连接、等待现有连接关闭、显式cancel并await后台任务、关闭资源；推荐手动管理事件循环以精准控制shutdown顺序，并加超时保护防卡死。
Python中捕获SIGINT的正确姿势
直接在主线程用 signal.signal(signal.SIGINT, handler) 没问题，但前提是主循环没被阻塞。常见陷阱是用了 asyncio.run() 后再注册信号——此时事件循环已启动，信号处理器可能不生效或被忽略。必须在调用 asyncio.run() 前注册，且 handler 函数需是同步的（不能是 async def）。
实际做法：
- 用
signal.signal(signal.SIGINT, sync_shutdown_handler)注册一个普通函数 - 该函数里触发异步退出逻辑：比如设置一个全局
asyncio.Event或调用loop.create_task(shutdown_coro()) - 确保事件循环对象可被访问（推荐通过
asyncio.get_running_loop()获取，而非保存引用）
asyncio服务热退出的关键步骤
热退出 ≠ 立即终止，而是等待正在处理的请求完成、释放资源、关闭监听套接字。核心是让协程有机会“善后”。asyncio.create_task() 启动的后台任务默认不会被自动取消，必须显式 await 它们。
典型 shutdown 流程应包含：
- 停止接收新连接：调用
server.close()（对asyncio.start_server返回的Server对象） - 等待已有连接自然结束：用
await server.wait_closed() - 逐个 cancel 并 await 所有活跃的后台 task：
asyncio.all_tasks()获取，过滤掉当前 shutdown task 本身 - 关闭 session、数据库连接池等：调用对应
.aclose()或.close()方法（注意区分同步/异步）
避免 asyncio.run() 导致的信号失效
asyncio.run() 内部会新建并管理一个事件循环，且在退出时强制关闭它——这会导致你在外部注册的 SIGINT 处理器无法控制 shutdown 顺序。更可控的方式是手动管理循环生命周期。
推荐写法：
- 用
loop = asyncio.new_event_loop()创建循环 - 调用
signal.signal(signal.SIGINT, lambda s, f: loop.create_task(shutdown())) - 用
loop.run_forever()启动，而不是asyncio.run() - 在
shutdown()协程末尾调用loop.stop()，让run_forever()退出
这样能完全掌控信号响应时机和 shutdown 顺序，也方便做超时兜底（比如 30 秒后强制 cancel）。
超时强制退出与资源泄漏风险
即使写了完整 shutdown 流程，仍可能卡住：某个协程死循环、await 了一个永不 resolve 的 Future、或 IO 阻塞未设 timeout。这时候必须加超时保护。
关键点：
- 给
await server.wait_closed()加asyncio.wait_for(..., timeout=10) - 对每个后台 task 的 await 也加 timeout，或统一用
asyncio.wait(tasks, timeout=10) - 强制 cancel 后，记得调用
task.cancelled()判断是否真被取消，避免假成功 - 特别注意：
asyncio.Lock、asyncio.Semaphore等对象若在 cancel 时正被 await，可能留下未释放状态——务必在 finally 块中检查并手动 release
真正难处理的从来不是“怎么关”，而是“关到哪一步才算安全”。比如 HTTP 连接池里的空闲连接要不要等它归还？数据库事务要不要回滚？这些得按业务语义定，代码只能提供可插拔的钩子。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
