1 引言
在前面的文章中，Event Loop 通过 call_soon 和 call_later 手动注册回调函数（Callback）的方式在 Python 中进行异步编程，但是这种方式非常不直观。为了让代码看起来像同步代码一样直观，Python 引入了 Coroutine（协程）。协程本质上是让程序在等待 IO 时能够主动“让出”控制权。
2 Coroutine
2.1 如何定义 coroutine
在使用协程时，async 和 await 是核心关键字。
async def 声明一个函数是异步的，而 await 则是告诉 Event Loop：我现在要开始等一个耗时操作了，你可以先去忙别的事，等我等的东西好了再回来叫我。在本例中，await asyncio.sleep(1) 将保证函数在继续下一次循环迭代之前至少阻塞 1 秒。它不能保证是实时的。
def print_now()print(datetime.datetime.now())async def keep_printing(name: str = "") -> None:while True:print_now()await asyncio.sleep(1)
2.2 如何运行 coroutine
有三种方式运行协程：
第一种是直接使用 asyncio.run(keep_printing())，这会创建一个全新的 Event Loop 并无限期运行该协程。
第二种是通过 asyncio.wait_for 配合 asyncio.run，这给协程设定的运行上限，例如 5 秒后强制停止并抛出超时异常。
第三种则是在一个异步主函数中捕获这个超时异常，从而优雅地处理超时逻辑而不让程序崩溃。
import datetime
import asyncio
loop = asyncio.get_event_loop()
def print_now():print(datetime.datetime.now())async def keep_printing(name: str = "") -> None:while True:print_now()await asyncio.sleep(1)# Method 1
# asyncio.run(keep_printing())# Method 2
# asyncio.run(asyncio.wait_for(keep_printing("Test"), timeout=5.0))# Method 3
# async def async_main() -> None:
# try:
# await asyncio.wait_for(keep_printing("Test"), timeout = 5.0)
# except asyncio.TimeoutError:
# print("Timed out!")# asyncio.run(async_main())3 从顺序执行到并发执行
3.1 为什么我的协程是顺序执行的？
看下面的代码：
import asyncioasync def keep_printing(name: str="") -> None:while True:print(f"{name}") await asyncio.sleep(1)async def async_main():await keep_printing("First")await keep_printing("Second")await keep_printing("Third")asyncio.run(async_main())
输出将是：
First
First
First
...
为什么？因为 await 会等待被等待的 Coroutine 完成。在本例中，keep_printing("First") 是一个无限循环，所以它永远不会完成，后续的 Coroutine 永远不会被执行。
3.2 解决方案 A: asyncio.gather
如果你有一组预先定义好的任务，想要让它们同时起跑并等待结果，gather 是最简单的选择：
async def async_main() -> None:await asyncio.gather(keep_printing("First"),keep_printing("Second"),keep_printing("Third"),)
asyncio.run(async_main())输出：
First
Second
Third
First
Second
Third
...
3.3 解决方案 B: asyncio.Task
Task 是对 Coroutine 的进一步封装。当你调用 asyncio.create_task(coro) 时，该任务会被 立即排入 Event Loop 的日程表。
import asyncioasync def keep_printing(name: str="") -> None:while True:print(f"{name}") await asyncio.sleep(1)async def async_main():# 1. 创建任务：这会立即把协程提交给 Event Loop# 此时任务已经在后台“启动”了，代码会继续向下执行task1 = asyncio.create_task(keep_printing("First"))task2 = asyncio.create_task(keep_printing("Second"))task3 = asyncio.create_task(keep_printing("Third"))# 2. 等待任务：因为这些是无限循环，await 会让 main 一直守在这里# 如果不 await，main 函数执行完毕后，这些后台任务会被强制关闭await task1await task2await task3if __name__ == "__main__":asyncio.run(async_main())
输出：
First
Second
Third
First
Second
Third
...
3.4 三种方案对比
| 方式 | 特点 | 适用场景 |
|---|---|---|
| 直接 await | 串行执行，等待当前行结束才走下一行 | 有严格先后逻辑的操作 |
| gather | 批量提交，等待所有任务完成 | 已知数量的一组独立请求 |
| create_task | 立即提交到后台，返回句柄，随时可取消 | 后台监控、长连接、动态增加的任务 |
4 任务管理：Cancel 机制
4.1 使用 asyncio.gather（整体管理）
在 gather 模式下，所有的协程被打包成一个“大任务”。如果你想取消，通常是配合 wait_for 整体切断。
Python
import asyncio
import datetimeasync def keep_printing(name: str="") -> None:try:while True:print(f"{name}: {datetime.datetime.now()}")await asyncio.sleep(1)except asyncio.CancelledError:# 当 wait_for 超时取消时，gather 会向内部所有协程发送取消信号print(f"--- [清理] {name} 收到取消信号并停止 ---")async def async_main() -> None:print(">>> Gather 模式启动：整体打包运行 3 秒")try:# gather 将三个协程绑定在一起，作为一个整体被 wait_for 监控await asyncio.wait_for(asyncio.gather(keep_printing("First"),keep_printing("Second"),keep_printing("Third"),), 3)except asyncio.TimeoutError:print(">>> 结果：5秒超时已到，Event Loop 自动切断了 gather 整体")asyncio.run(async_main())
4.2 使用 asyncio.create_task（独立控制）
在 Task 模式下，我们拥有每个任务的“遥控器”（即 Task 对象）。我们可以让任务 1 继续跑，而单独取消任务 2。
import asyncio
import datetimeasync def keep_printing(name: str="") -> None:try:while True:print(f"{name}: {datetime.datetime.now()}")await asyncio.sleep(1)except asyncio.CancelledError:print(f"--- [清理] {name} 被手动取消了 ---")async def async_main() -> None:print(">>> Task 模式启动：独立精细化管理")# 1. 立即把协程包装成 Task 并启动t1 = asyncio.create_task(keep_printing("First"))t2 = asyncio.create_task(keep_printing("Second"))t3 = asyncio.create_task(keep_printing("Third"))# 2. 让它们先跑 3 秒await asyncio.sleep(3)# 3. 展示精细化控制：只取消其中两个print(">>> 动作：我们要手动关掉 First 和 Second，保留 Third")t1.cancel()t2.cancel()# 4. 给一点时间让取消逻辑执行完，并观察 Third 还在继续打印print(">>> 观察：你会发现 Third 还在跑...")await asyncio.sleep(3)# 5. 最后清理 Thirdt3.cancel()await asyncio.gather(t1, t2, t3, return_exceptions=True)print(">>> 结果：所有任务已清理完毕")asyncio.run(async_main())
gather像是一个“包裹”：你把东西塞进去，await它。你不需要管里面每个任务的句柄（Variable）。在你的原代码中，wait_for超时时，它会向整个gather发送取消指令，内部所有任务都会跟着停。Task像是“遥控赛车”：你通过create_task拿到每一辆车的遥控器。你可以决定让红车停（t1.cancel()），蓝车继续开。
4.3 关于 Cancel 机制的补充
当你调用 .cancel() 时：
- Python 会在对应的协程下一次
await的地方抛出一个asyncio.CancelledError。 - 如果你在协程内部用了
try...except Exception:（捕获了所有异常），务必要重新抛出CancelledError或者单独处理它，否则任务可能无法真正停止，变成“僵尸任务”。
5 理解 Awaitable
不是所有东西都能被 await。在 Python 中，只有 Awaitable 对象可以，主要包括：
- Coroutine: 调用异步函数返回的对象。协程函数本身不能被 await，只有它返回的协程对象才可以。
- Task: 已经被排入日程的协程。当你用 asyncio.create_task() 包装一个协程时，你就创建了一个 Task。Task 是 Awaitable 的子类，它不仅可以被 await 拿回结果，还多了一些管理功能（如取消、查询状态）。
- Future: 低层级的异步结果占位符（后面会讲到）