Python 已经有 asyncio 很多年了。

async 和 await 语法也已经存在了很长时间。event loop 不再是什么陌生概念。FastAPI、aiohttp，以及许多数据库和 HTTP 库，都让异步编程成为 Python 开发中的常见组成部分。

但一个问题一直存在：

为什么异步 Python 还是让人觉得很难？

语法本身并不复杂。

真正困难的地方，是理解程序到底在做什么。

一旦应用程序变成异步的，你思考的就不再只是函数调用另一个函数。你还需要考虑 Task、调度、等待、取消、失败、资源生命周期，以及同步代码和异步代码之间的边界。

复杂性主要就来自这些地方。

async 和 await 是最简单的部分

一个简单的 coroutine 很容易理解：

import asyncio

async def fetch_data():
    await asyncio.sleep(1)
    return "done"

async def main():
    result = await fetch_data()
    print(result)

asyncio.run(main())

语法很直接。

async def 函数会创建一个 coroutine function，而调用它会产生一个 coroutine object。coroutine 不会因为你调用了它就直接开始执行。它必须被 await，或者被调度为一个 Task。

coroutine object 本身不会被 event loop 独立调度。只有直接 await 它，或者将它包装成一个可以由 event loop 调度的 Task，它才会开始运行。

这是异步代码与普通同步代码不同的第一个地方。

实际的模型更接近这样：

coroutine
    ↓
直接 await / 包装成 Task
    ↓
event loop
    ↓
运行直到 await
    ↓
等待 I/O
    ↓
运行另一个 Task

理解这个模型之后，async 和 await 就不会再显得神秘。

但这只是开始。

event loop 会改变你的思考方式

event loop 是 asyncio 的核心。

一个有用的思维模型是：一个 worker 管理许多 Task。

假设你有三个网络请求：

Task A → 等待网络
Task B → 等待网络
Task C → 等待网络

event loop 不需要一直阻塞在 A 上直到它完成。当 A 到达一个 awaitable 操作时，event loop 可以暂停 A，让 B 或 C 获得运行机会。

这就是 cooperative scheduling。

Python 文档对此有直接描述：event loop 一次运行一个 Task，当一个 Task 正在等待 Future 时，event loop 可以运行其他 Task、callback，或者执行 I/O 操作。

这就是为什么异步编程可以高效处理大量 I/O 操作。

但这里有一个重要后果：

event loop 必须保持可用，才能运行其他 Task。

看看下面的代码：

async def main():
    time.sleep(5)

这个函数虽然被声明为 async，但 time.sleep() 仍然是一个阻塞操作。

event loop 不会神奇地把同步阻塞代码变成异步代码。

如果 event loop 所在线程被阻塞五秒，其他 Task 在这段时间内就无法继续执行。

这是实际 asyncio 应用中最常见的错误之一。

基本规则很简单：

一个 async def 函数不会自动让里面的所有代码都变成异步代码。

你调用的库和你写的语法同样重要。

当你开始创建 Task 后，asyncio 就不那么简单了

当你不再这样写：

result = await fetch()

而开始这样写：

task = asyncio.create_task(fetch())

两者之间的区别就很重要了。

await fetch() 会等待这个 coroutine 完成，作为当前执行流程的一部分。create_task() 则会把这个 coroutine 调度成一个独立的 Task，让当前 coroutine 可以继续执行。

现在，你有了一个生命周期独立于当前执行位置的 Task。

这会立即带来一些新的问题。

这个 Task 什么时候完成？

谁负责等待它？

如果它失败了怎么办？

如果父操作被取消了怎么办？

这个 Task 会不会在创建它的请求结束之后还继续运行？

它占用的资源怎么办？

这些问题比 asyncio.create_task() 的语法本身重要得多。

过去，很多 asyncio 程序使用 asyncio.gather() 来协调多个 Task。

例如：

results = await asyncio.gather(
    fetch_users(),
    fetch_orders(),
    fetch_products(),
)

这个方法很有用，但并发编程有一个顺序编程经常隐藏的问题：

多个操作可能独立失败。

默认情况下，如果传给 gather() 的某个 awaitable 抛出异常，gather() 会把这个异常传播给调用者，但不会自动取消其他仍在运行的 awaitable。

Python 3.11 引入了 TaskGroup，作为一种更高级的 structured concurrency 机制。

async with asyncio.TaskGroup() as tg:
    users = tg.create_task(fetch_users())
    orders = tg.create_task(fetch_orders())
    products = tg.create_task(fetch_products())

这里重要的改进不只是语法更短。

这些 Task 现在拥有明确的生命周期。

它们属于一个 group，而离开 async with 块意味着这个 group 已经被处理完毕。如果其中一个 Task 抛出异常，其他 Task 会被取消，group 会等待它们完成，然后传播最终的 exception group。

这就是 structured concurrency 的核心思想：子 Task 与创建它们的作用域绑定。

Python 文档指出，对于嵌套的 Task 组，TaskGroup 相比 gather() 提供了更强的安全保证。

Cancellation 是 asyncio 真正变得复杂的地方

Cancellation 可能是异步程序变得复杂的最大原因之一。

在同步代码中，你通常会这样思考：

function starts
    ↓
function runs
    ↓
function returns

而在异步代码中，还存在另一条路径：

task starts
    ↓
task waits
    ↓
task gets cancelled
    ↓
cleanup
    ↓
task exits

当一个 asyncio Task 被取消时，Python 会在下一次有机会的时候抛出 asyncio.CancelledError。官方文档建议使用 try/finally 进行清理，并且通常应该让 CancelledError 在清理完成后继续向上传播。

例如：

async def worker():
    resource = await acquire_resource()

    try:
        await do_work(resource)
    finally:
        await release_resource(resource)

finally 块很重要，因为 cancellation 是异步应用程序正常控制流程的一部分。

Timeouts 可能导致 cancellation。

客户端断开连接可能导致 cancellation。

父 TaskGroup 可能导致 cancellation。

应用程序关闭时也可能导致 cancellation。

如果你的代码捕获了 CancelledError，通常应该执行必要的清理，然后重新抛出它，而不是静默吞掉 cancellation。

如果你的代码假设 coroutine 一定会从头运行到尾，那么迟早会遇到违反这个假设的情况。

Exception 也会变得更加复杂

假设有五个 Task 同时运行。

Task A 失败了。

几毫秒后 Task B 也失败了。

Task C 被取消。

Task D 成功。

Task E 还在运行。

那么程序到底应该报告什么？

这也是 Python 在 3.11 中引入 ExceptionGroup 和 except* 的原因之一。

PEP 654 解释了，并发系统可能需要同时传播多个互不相关的异常。传统的异常处理机制建立在一次传播一个异常的假设上，而这个假设并不总是适用于并发执行。

这也是异步编程并不只是“普通 Python 加上 await”的另一个原因。

并发会改变失败模型。

同步与异步之间的边界是另一个问题来源

一个现代 Python 应用很少会做到 100% 异步。

你可能会有这样的结构：

async HTTP server
        ↓
async database client
        ↓
synchronous library
        ↓
CPU-heavy processing
        ↓
file system

某个地方通常还是会出现同步代码。

例如：

async def handler():
    data = await fetch_data()

    result = legacy_library.process(data)

    return result

如果 legacy_library.process() 运行时间很长，event loop 可能会被阻塞。

有时候，正确的解决方式是把阻塞工作放到线程中：

result = await asyncio.to_thread(
    legacy_library.process,
    data,
)

asyncio.to_thread() 对 I/O-bound 的阻塞函数尤其有用，因为它会把工作移出 event loop 所在线程。在带 GIL 的标准 CPython 构建中，把普通 Python CPU-bound 工作移动到线程，并不会让这段计算通常获得真正的并行执行。

其他 CPU-heavy 工作可能需要使用进程或其他形式的并行。

重要的一点是，asyncio 主要提供的是 concurrency，而不是 parallel execution。多个 asyncio Task 通常会在同一个 event loop 线程上通过 cooperative scheduling 交替取得执行机会。

现代 Python 为开发者提供了更多选择，包括 free-threaded builds 和 subinterpreters，但这些属于不同的并发机制，各自有不同的权衡。

Asyncio 也有自己的同步原语

一旦多个 Task 开始共享状态，异步代码同样可能出现同步问题。

asyncio 提供了这些原语：

Lock
Event
Condition
Semaphore
BoundedSemaphore
Barrier

如果你使用过 threading，它们看起来会很熟悉。

但它们是为 asyncio Task 设计的，而不是用于 OS thread 同步。Python 文档明确指出，asyncio synchronization primitives 不是 thread-safe，不应该用来同步 OS threads。

例如，Semaphore 可以用于限制对某个资源的并发访问：

sem = asyncio.Semaphore(10)

async def fetch(url):
    async with sem:
        return await request(url)

这解决的是另一个问题，而不仅仅是让操作并发执行。

没有限制时，一个应用可能会从：

100 concurrent requests

变成：

10,000 concurrent requests

然后发现真正的瓶颈根本不是 Python 本身。

并发必须受到控制。

那么，Asyncio 还是很难吗？

是的。

但原因和表面上看起来有些不同。

asyncio 本身已经变得更加成熟。

基本语法已经稳定。

标准库提供了高级 API。

TaskGroup 提供了 structured concurrency。

真正困难的是，异步编程会暴露出一些顺序代码经常隐藏的问题。

你需要考虑：

Task lifetime
     ↓
Scheduling
     ↓
Blocking operations
     ↓
Cancellation
     ↓
Timeouts
     ↓
Exceptions
     ↓
Shared state
     ↓
Resource cleanup
     ↓
Shutdown

这些其实都不是语法问题。

它们是 生命周期和并发问题。

学习 Asyncio 的方式也变了

学习 asyncio 的一种常见方式，是记住各种 API：

async def
await
create_task()
gather()
sleep()
Queue
Lock
Semaphore

这当然有用，但它并不能提供一个完整的心智模型。

更好的学习方式，是理解一个异步操作的生命周期。

可以先从这个模型开始：

coroutine
    ↓
Task
    ↓
event loop
    ↓
await
    ↓
I/O completion
    ↓
resume
    ↓
result / exception / cancellation

然后学习多个 Task 是如何相互作用的。

接下来再学习：

TaskGroup
ExceptionGroup
timeouts
cancellation
resource cleanup
sync/async boundaries

一旦这些概念变得清晰，API 就会更容易理解。

你不再需要单独记住每个函数是做什么的。

你可以问自己一个更有用的问题：

这个 Task 从创建的那一刻开始，到彻底结束之前，究竟发生了什么？

这个问题已经非常接近异步编程的核心。

Asyncio 难，并不是因为它很新

asyncio 已经存在于 Python 中很多年了。

异步 I/O 背后的基本思想甚至更加古老。

真正发生变化的是，Python 的异步生态已经成熟到足以被越来越多地用于真实的生产系统。

而生产系统会暴露出那些真正困难的部分。

一个简单的示例可以教你这样写：

await fetch()

但一个真实的服务会迫使你回答：

What if fetch() takes too long?

What if the client disconnects?

What if the task is cancelled?

What if three tasks fail at once?

What if one library blocks the event loop?

What if a background task outlives its request?

What if the application shuts down while work is still running?

What if a resource needs asynchronous cleanup?

这些问题就是为什么这么多年过去了，asyncio 依然可能让人觉得困难。

好在现代 Python 已经提供了更好的工具来回答这些问题。

目标已经不再是让异步代码看起来像同步代码。

目标是理解 并发工作如何开始、等待、失败、被取消，以及最终结束。

一旦建立起这样的心智模型，asyncio 就不会再那么神秘。

更多相关技术内容咨询欢迎前往并持续关注好学星城论坛了解详情。

想高效系统的学习Python编程语言，推荐大家关注一个微信公众号：Python编程学习圈。每天分享行业资讯、技术干货供大家阅读，关注即可免费领取整套Python入门到进阶的学习资料以及教程，感兴趣的小伙伴赶紧行动起来吧。