Python 中使用 asyncio.create_task() 启动的任务会在 asyncio.run() 结束时被自动取消，但若需资源清理或状态持久化，仍应显式捕获 CancelledError 并执行 finally 清理逻辑，而非依赖隐式终止。
python 中使用 `asyncio.create_task()` 启动的任务会在 `asyncio.run()` 结束时被自动取消，但若需资源清理或状态持久化，仍应显式捕获 `cancellederror` 并执行 `finally` 清理逻辑，而非依赖隐式终止。
在异步 Python 应用中，理解任务生命周期与程序退出行为至关重要。如示例所示，asyncio.run(main()) 在 main() 协程返回后会自动关闭事件循环，并向所有仍在运行的后台任务（包括 create_task 创建的）发送取消信号——这并非“静默忽略”，而是标准的、受控的取消流程。任务会收到 asyncio.CancelledError 异常（继承自 BaseException），中断当前 await，并沿调用栈向上冒泡。
要验证这一点，只需在任务中添加异常捕获：
import asyncio
import random
async def forever_while_loop_task():
try:
while True:
sleep_seconds = random.randint(5, 30)
print(f"sleeping for {sleep_seconds} seconds")
await asyncio.sleep(sleep_seconds)
print("looping")
except asyncio.CancelledError:
print("✅ The task was cancelled gracefully")
# 可在此处释放连接、保存状态、记录日志等
raise # 必须重新抛出，确保取消传播完成
finally:
print("? Cleanup completed (e.g., closing DB connections)")
async def main():
task = asyncio.create_task(forever_while_loop_task())
await asyncio.sleep(20)
print("✔️ Main logic finished; event loop will now shut down")
# 运行后将看到：'✅ The task was cancelled gracefully' 和 '? Cleanup completed'
asyncio.run(main())
⚠️ 关键注意事项：
- ❌ 不要静默吞掉 CancelledError（例如 except asyncio.CancelledError: pass），否则任务无法真正终止，可能引发资源泄漏或死锁；
- ✅ 始终 raise 或 re-raise 该异常（即使已做清理），以配合 asyncio 的取消协议；
- ✅ 利用 finally 块执行必行清理操作（如关闭网络连接、释放锁、刷新缓冲区）；
- ? 若需更精细控制（如等待多个任务协同退出），推荐使用 asyncio.TaskGroup（Python 3.11+）：
async def main(): async with asyncio.TaskGroup() as tg: tg.create_task(forever_while_loop_task()) await asyncio.sleep(20) # TaskGroup 退出时自动 cancel + wait for all tasks
总结而言：asyncio.run() 的自动取消机制是可靠且符合规范的，但“自动取消”不等于“无需管理”。真正的优雅退出 = 显式响应取消信号 + 确保清理逻辑执行 + 遵循协程取消约定。将 CancelledError 视为正常控制流的一部分，而非错误，是编写健壮异步服务的核心实践。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
