asyncio 任务取消与资源清理

功能说明

示例展示如何取消正在运行的 asyncio 工作协程，同时保证资源清理。三个 Worker 从同一个队列取任务，每个任务在 try/finally 中调用 queue.task_done()；外层再包一层 except asyncio.CancelledError，在清理后重新抛出，让取消语义保持正确。主协程运行 20 毫秒后对所有任务调用 cancel()，再用 gather(return_exceptions=True) 收集每个任务的结束方式。

关键代码

item = await queue.get()
try:
    ...
finally:
    queue.task_done()
...
except asyncio.CancelledError:
    raise
finally:
    self.cleaned_up = True

CancelledError 不能被吞掉，否则等待 task.cancel() 的一方会误以为任务正常结束。

使用注意

• 取消发生在 await queue.get() 时，finally 仍会执行，但不会对尚未取到的元素调用 task_done()。
• 如果随后调用 queue.join()，未被处理完的元素会让它永久等待；取消场景不要依赖 join()。
• 工作者在收到 None 哨兵时正常退出，这是另一种优雅停止方式，示例并未使用。
• 取消是协作式的：阻塞在同步代码里的任务只有回到事件循环才会响应取消。
• 生产环境建议在 finally 中关闭连接、刷新缓冲或释放锁，而不仅仅是设置标志位。

验证方式

运行脚本应看到 cancelled=3、processed 为取消前已完成的数量、cleaned_up=True。重复运行会因调度差异得到不同的 processed，但取消数量和清理标志应始终稳定。