好的，各位观众，各位朋友，欢迎来到“Python asyncio 任务管理：取消、超时与异常处理”专场！今天咱们不聊虚的，直接上干货，教你如何像一位经验老道的船长一样，在asyncio的汪洋大海中掌控你的任务舰队，避免触礁沉没。
开场白：asyncio，你的任务，你的船！
asyncio，作为Python的异步IO框架，让你的程序在等待IO操作时不再傻等，而是可以切换到其他任务，提高效率。但是，这就好比你拥有了一支船队，每艘船（任务）都在执行不同的任务。如果管理不当，可能会出现船只迷航、超时搁浅、甚至遭遇风暴沉没的情况。
今天，我们就来聊聊如何有效地管理这些“船只”，确保它们安全、高效地完成任务。我们要关注的三个关键点是：
- 取消 (Cancellation)： 如何在任务不需要继续执行时，及时将其取消，释放资源。
- 超时 (Timeout)： 如何设置任务的执行时间限制，避免任务长时间阻塞，影响整个程序的运行。
- 异常处理 (Exception Handling)： 如何优雅地处理任务执行过程中可能出现的各种异常，保证程序的健壮性。
准备好了吗？ 让我们扬帆起航！
第一章：取消 (Cancellation)：当任务不再需要时…
想象一下，你派了一艘船去寻找宝藏，但后来你发现宝藏根本不存在，或者已经被人捷足先登了。这时，你肯定想把那艘船召回来，而不是让它继续漫无目的地漂流。这就是取消的作用。
在asyncio中，取消任务可以通过 task.cancel() 方法实现。这个方法会向任务发送一个 asyncio.CancelledError 异常。任务内部可以选择捕获这个异常并进行清理工作，或者直接让异常传播。
1.1 基础取消：简单粗暴，直接砍断！
最简单的取消方式就是直接调用 task.cancel()。
import asyncio
async def my_task(delay):
print(f"任务开始，延迟 {delay} 秒")
try:
await asyncio.sleep(delay)
print("任务完成") # 如果任务被取消，这行代码不会执行
except asyncio.CancelledError:
print("任务被取消了！")
finally:
print("清理工作...") # 无论是否被取消，都会执行
async def main():
task = asyncio.create_task(my_task(5))
await asyncio.sleep(2) # 等待2秒
print("取消任务！")
task.cancel()
try:
await task # 等待任务结束
except asyncio.CancelledError:
print("任务确实被取消了！")
if __name__ == "__main__":
asyncio.run(main())在这个例子中，my_task 会休眠5秒，但在休眠2秒后，我们调用了 task.cancel()，任务会被取消，并执行 except asyncio.CancelledError 和 finally 中的代码。
1.2 优雅取消：给任务一个缓冲的机会
直接砍断虽然有效，但有时候过于粗暴。如果任务正在进行一些重要的操作（例如，正在写入文件），直接取消可能会导致数据丢失。为了更优雅地取消任务，我们可以在任务内部检查是否被取消，并提前退出。
import asyncio
async def my_task(delay):
print(f"任务开始，延迟 {delay} 秒")
for i in range(delay):
if asyncio.current_task().cancelled():
print("任务检测到取消信号，准备退出...")
return # 提前退出
await asyncio.sleep(1)
print(f"已运行 {i+1} 秒")
print("任务完成")
async def main():
task = asyncio.create_task(my_task(5))
await asyncio.sleep(2)
print("取消任务！")
task.cancel()
try:
await task
except asyncio.CancelledError:
print("任务确实被取消了！")
if __name__ == "__main__":
asyncio.run(main())在这个例子中，任务在每次休眠后，都会检查 asyncio.current_task().cancelled() 是否返回 True。如果是，则提前退出，避免强制取消带来的问题。
1.3 取消的传播：一荣俱荣，一损俱损？
如果一个任务内部创建了子任务，取消父任务是否会影响子任务？答案是：默认情况下，不会。你需要手动取消子任务。
import asyncio
async def sub_task(delay):
print(f"子任务开始，延迟 {delay} 秒")
try:
await asyncio.sleep(delay)
print("子任务完成")
except asyncio.CancelledError:
print("子任务被取消了！")
finally:
print("子任务清理工作...")
async def parent_task():
print("父任务开始")
task = asyncio.create_task(sub_task(3))
try:
await asyncio.sleep(1)
print("父任务完成")
return # 父任务完成，不取消子任务
except asyncio.CancelledError:
print("父任务被取消了！")
task.cancel() # 手动取消子任务
await task # 等待子任务结束
finally:
print("父任务清理工作...")
async def main():
task = asyncio.create_task(parent_task())
await asyncio.sleep(2)
print("取消父任务！")
task.cancel()
try:
await task
except asyncio.CancelledError:
print("父任务确实被取消了！")
if __name__ == "__main__":
asyncio.run(main())在这个例子中，如果父任务被取消，它会手动取消子任务。否则，子任务会继续执行。
第二章：超时 (Timeout)：给任务设定一个截止日期
有时候，任务可能会因为各种原因而阻塞，例如，等待一个永远不会返回的网络请求。为了避免这种情况，我们需要设置任务的超时时间，如果任务在指定时间内没有完成，就自动取消它。
2.1 asyncio.wait_for()：简单好用，一招制敌！
asyncio.wait_for() 是实现超时的最简单方法。它可以等待一个协程完成，如果在指定的超时时间内没有完成，则抛出一个 asyncio.TimeoutError 异常。
import asyncio
async def my_task(delay):
print(f"任务开始，延迟 {delay} 秒")
await asyncio.sleep(delay)
print("任务完成")
async def main():
try:
await asyncio.wait_for(my_task(5), timeout=2) # 超时时间为2秒
print("任务成功完成！")
except asyncio.TimeoutError:
print("任务超时了！")
if __name__ == "__main__":
asyncio.run(main())在这个例子中，my_task 会休眠5秒，但我们设置了超时时间为2秒。因此，asyncio.wait_for() 会抛出 asyncio.TimeoutError 异常。
2.2 asyncio.Timeout() (Python 3.11+)：更灵活的超时控制
在 Python 3.11 中，asyncio 引入了 asyncio.Timeout() 上下文管理器，提供了更灵活的超时控制。你可以使用 async with 语句来包裹需要设置超时的代码块。
import asyncio
async def my_task(delay):
print(f"任务开始，延迟 {delay} 秒")
await asyncio.sleep(delay)
print("任务完成")
async def main():
try:
async with asyncio.Timeout(2): # 超时时间为2秒
await my_task(5)
print("任务成功完成！")
except asyncio.TimeoutError:
print("任务超时了！")
if __name__ == "__main__":
asyncio.run(main())asyncio.Timeout() 的优势在于，它可以精确地控制超时的范围，并且可以嵌套使用，实现更复杂的超时逻辑。
2.3 超时与取消：相辅相成，缺一不可
当 asyncio.wait_for() 或 asyncio.Timeout() 抛出 asyncio.TimeoutError 异常时，它实际上会尝试取消被超时的任务。因此，你需要在任务内部处理 asyncio.CancelledError 异常，并进行必要的清理工作。
第三章：异常处理 (Exception Handling)：防患于未然，稳如泰山
在asyncio的世界里，异常处理至关重要。一个未处理的异常可能会导致整个事件循环崩溃，甚至影响到其他任务的执行。所以，我们需要像老船长一样，时刻警惕风暴的来临，并做好充分的准备。
3.1 try...except：最基本的防线
最基本的异常处理方式就是使用 try...except 语句。你可以用它来捕获任务执行过程中可能出现的各种异常。
import asyncio
async def my_task():
print("任务开始")
try:
result = 1 / 0 # 故意引发一个除零错误
print(f"结果：{result}")
except ZeroDivisionError:
print("发生了除零错误！")
except Exception as e:
print(f"发生了其他错误：{e}")
finally:
print("任务结束")
async def main():
await my_task()
if __name__ == "__main__":
asyncio.run(main())在这个例子中，我们故意引发了一个 ZeroDivisionError 异常，并使用 try...except 语句捕获了它。
3.2 asyncio.gather() 的异常处理：集中火力，各个击破
asyncio.gather() 可以并发地运行多个协程，并返回它们的结果。但是，如果其中一个协程抛出了异常，asyncio.gather() 默认情况下会立即取消所有其他协程，并将异常传播给调用者。
import asyncio
async def task1():
print("任务1开始")
await asyncio.sleep(1)
print("任务1完成")
return "任务1结果"
async def task2():
print("任务2开始")
await asyncio.sleep(0.5)
raise ValueError("任务2出错了！") # 引发一个异常
async def task3():
print("任务3开始")
await asyncio.sleep(2)
print("任务3完成")
return "任务3结果"
async def main():
try:
results = await asyncio.gather(task1(), task2(), task3())
print(f"所有任务的结果：{results}")
except Exception as e:
print(f"发生了错误：{e}")
if __name__ == "__main__":
asyncio.run(main())在这个例子中，task2 抛出了一个 ValueError 异常，导致 asyncio.gather() 取消了 task1 和 task3，并将 ValueError 异常传播给 main 函数。
*3.3 `asyncio.gather(tasks, return_exceptions=True)`：允许部分失败**
如果你希望 asyncio.gather() 忽略部分任务的异常，并继续执行其他任务，可以使用 return_exceptions=True 参数。
import asyncio
async def task1():
print("任务1开始")
await asyncio.sleep(1)
print("任务1完成")
return "任务1结果"
async def task2():
print("任务2开始")
await asyncio.sleep(0.5)
raise ValueError("任务2出错了！") # 引发一个异常
async def task3():
print("任务3开始")
await asyncio.sleep(2)
print("任务3完成")
return "任务3结果"
async def main():
results = await asyncio.gather(task1(), task2(), task3(), return_exceptions=True)
print(f"所有任务的结果：{results}")
for result in results:
if isinstance(result, Exception):
print(f"发生了错误：{result}")
else:
print(f"任务结果：{result}")
if __name__ == "__main__":
asyncio.run(main())在这个例子中，asyncio.gather() 会继续执行 task1 和 task3，并将 task2 抛出的 ValueError 异常作为结果返回。你需要手动检查结果列表，判断哪些任务成功完成，哪些任务失败。
3.4 TaskGroup (Python 3.11+)：更强大的任务分组与异常处理
Python 3.11 引入了 TaskGroup，提供了一种更结构化的方式来管理和处理并发任务。TaskGroup 允许你在一个代码块中创建和管理多个任务，并自动处理任务之间的依赖关系和异常传播。
import asyncio
async def task1():
print("任务1开始")
await asyncio.sleep(1)
print("任务1完成")
return "任务1结果"
async def task2():
print("任务2开始")
await asyncio.sleep(0.5)
raise ValueError("任务2出错了！") # 引发一个异常
async def task3():
print("任务3开始")
await asyncio.sleep(2)
print("任务3完成")
return "任务3结果"
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(task1())
tg.create_task(task2())
tg.create_task(task3())
print("所有任务完成！") # 如果有任务抛出异常，这行代码不会执行
except ExceptionGroup as eg:
print("发生了多个错误：")
for e in eg.exceptions:
print(f" - {e}")
if __name__ == "__main__":
asyncio.run(main())在这个例子中，TaskGroup 会自动等待所有任务完成，并捕获所有异常。如果任何一个任务抛出异常，TaskGroup 会抛出一个 ExceptionGroup 异常，其中包含了所有异常的信息。
总结：asyncio 任务管理的葵花宝典
| 功能 | 方法/类 | 说明 | 适用场景 |
|---|---|---|---|
| 取消任务 | task.cancel() |
向任务发送 asyncio.CancelledError 异常，任务可以选择捕获并进行清理。 |
当任务不再需要继续执行时，释放资源。 |
| 优雅取消任务 | asyncio.current_task().cancelled() |
在任务内部检查是否被取消，并提前退出。 | 避免强制取消导致数据丢失等问题。 |
| 设置超时 | asyncio.wait_for() |
等待一个协程完成，如果在指定的超时时间内没有完成，则抛出一个 asyncio.TimeoutError 异常。 |
避免任务长时间阻塞，影响整个程序的运行。 |
| 设置超时 (3.11+) | asyncio.Timeout() |
上下文管理器，提供更灵活的超时控制。 | 需要精确控制超时范围，或者需要嵌套使用超时逻辑的场景。 |
| 异常处理 | try...except |
捕获任务执行过程中可能出现的各种异常。 | 所有需要进行异常处理的场景。 |
异常处理 (gather) |
asyncio.gather(..., return_exceptions=True) |
允许 asyncio.gather() 忽略部分任务的异常，并继续执行其他任务。 |
允许部分任务失败，但需要继续执行其他任务的场景。 |
| 异常处理 (3.11+) | asyncio.TaskGroup |
结构化的任务分组与异常处理，自动处理任务之间的依赖关系和异常传播。 | 需要管理和处理多个并发任务，并自动处理任务之间的依赖关系和异常传播的场景。 |
结束语：愿你乘风破浪，一帆风顺！
掌握了取消、超时和异常处理，你就掌握了asyncio任务管理的关键。希望今天的讲座能帮助你更好地驾驭asyncio，让你的程序在异步的世界里乘风破浪，一帆风顺！记住，好的船长不仅需要高超的驾驶技术，更需要对潜在风险的敏锐洞察和充分准备。
感谢大家的观看！我们下次再见！