Python Async/Await 异步编程深度指南

前言：告别阻塞，拥抱效率

在现代软件开发中，尤其是涉及大量 I/O 操作（如网络请求、数据库查询、文件读写）的应用中，传统的同步编程模式常常会遇到性能瓶颈。一个操作的等待会阻塞整个程序的执行，导致资源无法得到充分利用，响应速度变慢。为了解决这个问题，并发编程应运而生。

Python 提供了多种并发机制：

多进程 (Multiprocessing): 利用操作系统级别的多进程，可以充分利用多核 CPU，适合 CPU 密集型任务。进程间内存独立，通信开销相对较大。
多线程 (Threading): 在同一进程内创建多个线程。线程间共享内存，通信方便，但受限于 Python 的全局解释器锁 (GIL)，对于 CPU 密集型任务无法实现真正的并行。然而，对于 I/O 密集型任务，当一个线程等待 I/O 时，GIL 会释放，允许其他线程执行，因此多线程在某些 I/O 场景下也能提升效率。
异步编程 (Asynchronous Programming): 通过在等待 I/O 完成时切换执行其他任务，而不是阻塞等待，从而提高单线程或少量线程的并发能力。Python 的异步编程主要通过 asyncio 库和 async/await 语法来实现。

本文将聚焦于第三种方式：基于 asyncio 库和 async/await 关键字的 Python 异步编程。我们将详细探讨其核心概念、工作原理、使用方法以及最佳实践。

1. 什么是异步编程？同步与异步、阻塞与非阻塞

理解异步编程，首先要区分几个概念：

同步 (Synchronous): 代码按照顺序一条一条执行，一条代码没有执行完，后面的代码必须等待。当遇到一个耗时操作（如网络请求），程序会暂停在此，直到操作完成并返回结果。
异步 (Asynchronous): 代码执行时，遇到耗时操作不会原地等待，而是注册一个“回调”或“将来完成时通知”的机制，然后继续执行后续代码。当耗时操作完成后，系统会通知程序，程序再回头处理之前注册的任务。
阻塞 (Blocking): 当一个操作导致程序暂停执行，直到操作完成，这就是阻塞。同步操作通常是阻塞的。
非阻塞 (Non-blocking): 当一个操作不会导致程序暂停，即使操作尚未完成，程序也可以继续执行其他任务。异步操作通常是非阻塞的。

异步编程的核心思想在于：在等待 I/O 操作的时间里，不让 CPU 闲着，而是切换去做其他“准备好”的任务。它不是通过增加线程或进程来实现并行（同时执行），而是通过协作式多任务 (Cooperative Multitasking) 来实现并发（看似同时进行）。

想象一下：

同步阻塞: 你去餐厅点餐，点完一道菜，必须站在柜台前等着这道菜做好，拿到手了，才能点下一道菜。
异步非阻塞 (使用回调): 你去餐厅点餐，点完一道菜，服务员告诉你好了会叫你，你就可以找个座位坐下或者去干点别的（比如看菜单准备点下一道菜）。菜做好了，服务员叫你，你再去取。
异步非阻塞 (使用 async/await): 你去餐厅点餐，点完一道菜，服务员告诉你好了会叫你，你就可以去干点别的。当你“await”这道菜时，你就把控制权交给了餐厅（事件循环）。餐厅看到你在等待，就去处理其他人的点餐。你的菜做好了，餐厅（事件循环）会通知你，并把控制权交还给你，你就可以继续享用你的菜了。

async/await 语法是 Python 实现异步非阻塞编程的一种更具可读性的方式，它让异步代码看起来很像同步代码，避免了回调地狱 (callback hell)。

2. Python 中的异步编程：`asyncio` 和 `async/await`

Python 3.4 引入了 asyncio 库，提供了基于事件循环 (event loop) 的异步编程框架。Python 3.5 引入了 async 和 await 关键字，让异步代码的编写变得更加直观和易读。从此，asyncio + async/await 成为了 Python 官方推荐的异步编程方式。

核心概念：

事件循环 (Event Loop): 异步编程的心脏。它是一个循环，不断地检查有没有任务准备好运行（例如，某个 I/O 操作完成了）。它负责调度协程的执行。
协程 (Coroutine): 使用 async def 定义的函数。协程是可暂停和可恢复的函数。当协程内部遇到 await 关键字时，它会暂停执行，将控制权交还给事件循环，等待某个操作完成。操作完成后，事件循环会在合适的时机恢复该协程的执行。协程本身并不会立即执行，它需要被事件循环调度。
可等待对象 (Awaitable): 可以在 await 表达式后面使用的对象。主要包括：
- 协程 (Coroutine)
- 任务 (Task)
- 未来对象 (Future)
  await expression 的作用是暂停当前协程的执行，等待 expression 的结果，然后恢复当前协程的执行。
任务 (Task): asyncio 中，任务是协程的包装。事件循环通过任务来调度协程的执行。你可以使用 asyncio.create_task() 来创建一个任务，并将其放入事件循环中等待执行。任务也是可等待对象。
未来对象 (Future): 表示一个尚未完成的操作的最终结果。当操作完成时，Future 对象会被设置一个结果或一个异常。任务是 Future 的子类。在编写大部分异步代码时，你直接与协程和任务打交道居多，但 Future 是底层机制。

`async def`：定义协程函数

使用 async def 关键字来定义一个协程函数：

“`python
import asyncio

async def my_coroutine():
print(“Coroutine started”)
# 模拟一个耗时的异步 I/O 操作
await asyncio.sleep(1) # await 是关键，表示暂停并让出控制权
print(“Coroutine finished after 1 second”)

直接调用 async def 函数会返回一个协程对象，而不是执行它

coro_obj = my_coroutine()
print(f”Type of direct call result: {type(coro_obj)}”) #

如何运行协程？需要事件循环

“`

一个 async def 函数中，你可以使用 await 关键字来等待另一个可等待对象（协程、任务或 Future）。请注意：await 只能用在 async def 函数内部。

`await`: 暂停与等待

await 关键字是异步编程的核心操作符。当你在协程中使用 await some_awaitable 时：

当前协程的执行会暂停。
控制权被交还给事件循环。
事件循环会检查是否有其他准备好的任务可以运行。
当 some_awaitable 完成并产生结果或异常时，事件循环会在合适的时机恢复之前暂停的协程，并将 await 表达式的结果作为 some_awaitable 的结果。

“`python
import asyncio

async def task1():
print(“Task 1 started”)
await asyncio.sleep(2) # 模拟耗时 2 秒
print(“Task 1 finished”)

async def task2():
print(“Task 2 started”)
await asyncio.sleep(1) # 模拟耗时 1 秒
print(“Task 2 finished”)

async def main():
print(“Main coroutine started”)
# 同步调用 awaitables 会按顺序执行
# await task1()
# await task2()
# 上面这样写虽然用了 await，但仍然是串行的！总耗时 3 秒

# 要实现并发，需要创建任务并同时等待它们
print("Creating tasks...")
task_a = asyncio.create_task(task1()) # 创建任务，并将其放入事件循环
task_b = asyncio.create_task(task2()) # 创建任务，并将其放入事件循环

print("Tasks created, waiting using asyncio.gather...")
# asyncio.gather() 并发运行可等待对象，并等待它们全部完成
await asyncio.gather(task_a, task_b) # await gather() 等待所有任务完成

print("Main coroutine finished")

使用 asyncio.run() 运行顶层协程 (Python 3.7+)

asyncio.run() 负责创建和关闭事件循环

if name == “main“:
print(“Running with asyncio.run…”)
asyncio.run(main())

输出大致会是这样 (顺序可能因调度略有不同):

Running with asyncio.run…

Main coroutine started

Creating tasks…

Tasks created, waiting using asyncio.gather…

Task 1 started

Task 2 started

Task 2 finished

Task 1 finished

Main coroutine finished

注意 Task 2 (1秒) 在 Task 1 (2秒) 之前完成，证明是并发执行

“`

上面的例子中，asyncio.sleep() 是一个协程，它模拟了耗时的异步操作。当你 await asyncio.sleep(1) 时，当前协程会暂停执行，控制权回到事件循环，事件循环可以去执行其他任务。1秒后，事件循环会“唤醒”这个协程，让它从暂停的地方继续执行。

通过 asyncio.create_task() 和 asyncio.gather()，我们并发地运行了 task1 和 task2。asyncio.gather() 等待所有的任务完成，并收集它们的结果（如果协程有返回值的话）。

`asyncio.run()`：运行顶层协程

asyncio.run(coro, *, debug=False) 是 Python 3.7+ 推荐的运行异步程序的入口点。它负责：

获取当前线程的事件循环（如果还没有，则创建一个新的）。
运行传入的协程 coro。
管理事件循环，直到协程 coro 完成。
关闭事件循环。

这使得启动一个简单的异步程序变得非常方便。对于更复杂的场景（如在已有同步框架中集成异步代码），可能需要手动获取和管理事件循环。

手动管理事件循环 (了解)

在 Python 3.7 之前或特定场景下，你可能需要手动管理事件循环：

“`python
import asyncio

async def simple_coro():
print(“Hello”)
await asyncio.sleep(1)
print(“World”)

Python 3.6 及更早版本或需要精细控制时

loop = asyncio.get_event_loop() # 获取事件循环

try:

loop.run_until_complete(simple_coro()) # 运行直到协程完成

finally:

loop.close() # 关闭事件循环

Python 3.7+ 推荐使用 asyncio.run()，它内部做了类似的事情

“`

3. `asyncio` 的主要组件和功能

除了 async/await 语法和事件循环，asyncio 还提供了许多用于构建复杂异步应用的工具：

3.1 `asyncio.sleep()`

前面已经用到，这是模拟异步操作的最简单方式。它是一个可等待的协程，当 await asyncio.sleep(n) 时，当前协程会暂停 n 秒，然后恢复。

“`python
import asyncio

async def sleepy_task(name, delay):
print(f”Task {name}: Starting sleep {delay}”)
await asyncio.sleep(delay)
print(f”Task {name}: Finished sleep {delay}”)
return f”Task {name} result”

async def main_sleep():
print(“Main sleep start”)
results = await asyncio.gather(
sleepy_task(“A”, 3),
sleepy_task(“B”, 1),
sleepy_task(“C”, 2)
)
print(f”Main sleep finished, results: {results}”)

asyncio.run(main_sleep())

Output:

Main sleep start

Task A: Starting sleep 3

Task B: Starting sleep 1

Task C: Starting sleep 2

Task B: Finished sleep 1

Task C: Finished sleep 2

Task A: Finished sleep 3

Main sleep finished, results: [‘Task A result’, ‘Task B result’, ‘Task C result’]

注意任务完成的顺序与它们的 delay 相关，而不是启动顺序

“`

3.2 `asyncio.create_task()` 和 `asyncio.gather()`

asyncio.create_task(coro): 将一个协程包装成一个 Task 并安排它在事件循环中运行。返回 Task 对象。Task 会立即被调度，但只有当控制权回到事件循环时才可能真正开始执行。
asyncio.gather(*awaitables, return_exceptions=False): 并发地运行 *awaitables（协程、任务等）。它是一个协程，当所有 awaitables 完成后，它也会完成。结果是一个列表，包含每个 awaitable 的结果，顺序与传入的顺序一致。如果 return_exceptions=False (默认)，任何一个 awaitable 发生异常都会立即取消其他等待的 awaitables，并传播异常。如果 return_exceptions=True，即使发生异常，其他 awaitables 也会继续运行，异常会在结果列表中以异常对象的形式返回。

这是实现并发的两种主要方式：创建一个个任务，然后用 gather 等待它们。或者直接将多个协程传给 gather，gather 会自动为它们创建任务并等待。

3.3 同步原语 (Synchronization Primitives)

在并发编程中，多个协程可能需要访问共享资源。为了避免竞态条件 (race conditions)，需要使用同步原语。asyncio 提供了异步版本的同步原语：

asyncio.Lock: 异步锁。用于保护共享资源的访问。当一个协程获取了锁，其他试图获取该锁的协程会被暂停，直到锁被释放。

“`python
import asyncio

async def critical_section(lock, name):
print(f”Task {name}: Trying to acquire lock”)
async with lock: # 使用 async with 语法获取锁，更安全
print(f”Task {name}: Acquired lock”)
await asyncio.sleep(0.1) # 模拟在临界区内的操作
print(f”Task {name}: Releasing lock”)

async def main_lock():
lock = asyncio.Lock()
await asyncio.gather(
critical_section(lock, “A”),
critical_section(lock, “B”),
critical_section(lock, “C”)
)

asyncio.run(main_lock())

Output shows tasks acquiring lock one by one, not simultaneously

Task A: Trying to acquire lock

Task A: Acquired lock

Task A: Releasing lock

Task B: Trying to acquire lock

Task B: Acquired lock

Task B: Releasing lock

Task C: Trying to acquire lock

Task C: Acquired lock

Task C: Releasing lock

“`
asyncio.Semaphore: 异步信号量。用于控制同时访问某个资源的协程数量。

“`python
import asyncio

async def limited_resource(semaphore, name):
print(f”Task {name}: Trying to acquire semaphore”)
async with semaphore:
print(f”Task {name}: Acquired semaphore”)
await asyncio.sleep(0.5) # 模拟使用资源
print(f”Task {name}: Releasing semaphore”)

async def main_semaphore():
# 只允许最多 2 个协程同时访问资源
semaphore = asyncio.Semaphore(2)
await asyncio.gather(
limited_resource(“A”, semaphore),
limited_resource(“B”, semaphore),
limited_resource(“C”, semaphore),
limited_resource(“D”, semaphore)
)

asyncio.run(main_semaphore())

Output shows only 2 tasks acquired semaphore at any time

“`
asyncio.Event: 异步事件。一个简单的标志，可以被设置 (set) 和清除 (clear)。协程可以等待事件被设置。

“`python
import asyncio

async def wait_for_event(event, name):
print(f”Task {name}: Waiting for event”)
await event.wait() # 暂停直到事件被设置
print(f”Task {name}: Event received, continuing”)

async def set_the_event(event, delay):
print(f”Setter: Sleeping for {delay} before setting event”)
await asyncio.sleep(delay)
print(“Setter: Setting the event”)
event.set() # 设置事件，唤醒所有等待 event.wait() 的协程

async def main_event():
event = asyncio.Event()
await asyncio.gather(
wait_for_event(event, “Reader1”),
wait_for_event(event, “Reader2”),
set_the_event(event, 1)
)

asyncio.run(main_event())

Output:

Task Reader1: Waiting for event

Task Reader2: Waiting for event

Setter: Sleeping for 1 before setting event

Setter: Setting the event

Task Reader1: Event received, continuing

Task Reader2: Event received, continuing

“`
asyncio.Condition: 异步条件变量。通常与锁一起使用，允许协程在某个条件满足之前暂停，并在条件可能满足时被其他协程唤醒。
asyncio.Queue: 异步队列。用于协程之间进行通信和数据传递。put() 和 get() 方法都是可等待的。

“`python
import asyncio
import random

async def producer(queue, num_items):
for i in range(num_items):
item = f”item-{i}”
print(f”Producer: Putting {item}”)
await queue.put(item) # 放入队列，如果队列满了则等待
await asyncio.sleep(random.random() * 0.1) # 模拟生产延迟
await queue.put(None) # 发送结束信号

async def consumer(queue, name):
print(f”Consumer {name}: Starting”)
while True:
item = await queue.get() # 从队列获取，如果队列空了则等待
if item is None:
print(f”Consumer {name}: Received stop signal, exiting”)
queue.task_done() # 表示 None 处理完成
break
print(f”Consumer {name}: Got {item}”)
await asyncio.sleep(random.random() * 0.2) # 模拟处理延迟
queue.task_done() # 表示该 item 处理完成

async def main_queue():
queue = asyncio.Queue() # 创建一个无限大的队列
# queue = asyncio.Queue(maxsize=5) # 创建一个有最大容量的队列
```
await asyncio.gather(
    producer(queue, 10),
    consumer(queue, "C1"),
    consumer(queue, "C2")
    # Note: You might need to put None multiple times for multiple consumers
    # Or use a different mechanism to signal completion if consumers should all exit
    # For simplicity here, sending None once per consumer is implied for clean exit
    # A more robust approach involves waiting for queue.join() and cancelling consumers
)
# A more robust consumer exit:
# await producer(queue, 10)
# await queue.join() # Wait until all items in the queue have been received and processed
# for _ in range(2): # Send stop signal for 2 consumers
#     await queue.put(None)
```
asyncio.run(main_queue())

“`

3.4 处理阻塞代码 (`loop.run_in_executor`)

asyncio 的事件循环是单线程的。如果在协程中执行一个同步的、阻塞的函数（例如，调用 time.sleep() 而不是 asyncio.sleep()，或者执行一个耗时的计算），将会阻塞整个事件循环，导致其他协程无法运行，丧失异步的优势。

如果必须执行阻塞的代码，应该将其放到单独的线程或进程中运行，然后通过 Future 或 Task 来等待其结果。asyncio 提供了 loop.run_in_executor() 方法来方便地做到这一点：

“`python
import asyncio
import time
import concurrent.futures # 导入标准库的线程池或进程池

def blocking_function(seconds):
print(f”Executing blocking function for {seconds} seconds…”)
time.sleep(seconds) # 这是一个阻塞调用
print(f”Blocking function finished.”)
return f”Result from blocking after {seconds} seconds”

async def main_executor():
loop = asyncio.get_event_loop()
# 使用默认的 ThreadPoolExecutor
print(“Starting non-blocking part”)
# 将阻塞函数提交到线程池中运行，返回一个 Future 对象
future = loop.run_in_executor(None, blocking_function, 2) # 第一个参数 None 表示使用默认执行器
print(“Submitted blocking function to executor, continuing other tasks”)

# 可以继续执行其他非阻塞的异步任务
asyncio.create_task(asyncio.sleep(1)) # 另一个任务
asyncio.create_task(asyncio.sleep(1.5)) # 另一个任务

print("Waiting for blocking function to complete...")
result = await future # await 等待 Future 完成，但不会阻塞事件循环

print(f"Blocking function result: {result}")
print("Main executor finished")

asyncio.run(main_executor())

Output shows non-blocking parts can run while the blocking function is in the background thread

Starting non-blocking part

Submitted blocking function to executor, continuing other tasks

Waiting for blocking function to complete…

Executing blocking function for 2 seconds…

Blocking function finished.

Blocking function result: Result from blocking after 2 seconds

Main executor finished

Note: Sleep messages from asyncio.sleep tasks might interleave depending on exact timing

“`

run_in_executor 默认使用一个 ThreadPoolExecutor。你也可以配置使用 ProcessPoolExecutor 来执行 CPU 密集型任务，以绕过 GIL。

3.5 取消任务 (`task.cancel()`)

异步任务是可以被取消的。你可以调用任务对象的 cancel() 方法来请求取消。

“`python
import asyncio

async def cancellable_task():
print(“Cancellable task starting…”)
try:
# 模拟一个长时间运行的任务
await asyncio.sleep(1000) # 会在此暂停很长时间
except asyncio.CancelledError:
print(“Cancellable task was cancelled!”)
# 在这里进行清理工作（如果有的话）
finally:
print(“Cancellable task exiting.”)
print(“This line will not be reached if cancelled”)

async def main_cancel():
print(“Main cancel start”)
task = asyncio.create_task(cancellable_task())

# 等待一段时间，然后取消任务
await asyncio.sleep(1)
print("Main cancel: Cancelling the task...")
task.cancel()

# 等待任务真正完成（被取消）
try:
    await task
except asyncio.CancelledError:
    print("Main cancel: Task successfully cancelled and awaited.")

print("Main cancel finished")

asyncio.run(main_cancel())

Output:

Main cancel start

Cancellable task starting…

Main cancel: Cancelling the task…

Cancellable task was cancelled!

Cancellable task exiting.

Main cancel: Task successfully cancelled and awaited.

Main cancel finished

“`

当任务被取消时，它会在等待点（即 await 表达式处）抛出一个 asyncio.CancelledError 异常。协程应该捕获这个异常来执行必要的清理工作。如果没有捕获，异常会向上冒泡。await task 在任务被取消时会重新抛出 CancelledError。

4. 异步生态系统：常用的异步库

asyncio 只是一个框架，要实现真正的异步 I/O，你需要使用支持 asyncio 的库。许多流行的同步库都有异步版本：

网络请求: aiohttp (异步版的 requests) 是最常用的异步 HTTP 客户端/服务器库。
数据库访问:
- asyncpg (异步 PostgreSQL 驱动)
- aiomysql (异步 MySQL 驱动)
- aioodbc (异步 ODBC 驱动)
- asyncio-redis, aioredis (异步 Redis 客户端)
- motor (异步 MongoDB 驱动)
Web 框架:
- aiohttp (同时是异步 Web 框架)
- FastAPI (基于 Starlette 和 Pydantic 的高性能框架，广泛使用 async/await)
- Starlette (轻量级异步 Web 框架)
- Quart (异步版的 Flask)
消息队列: aio-pika (异步 AMQP 客户端)
文件系统: aiofiles (异步文件操作)
底层网络: asyncio 本身提供了 open_connection, start_server 等函数用于创建异步 TCP/UDP 客户端和服务端。

重要提示: 不要在 async def 协程中直接使用同步的、阻塞的 I/O 库 (例如 requests, time.sleep, psycopg2, mysql.connector, mongodb 同步驱动等)，除非你使用 loop.run_in_executor() 将它们放入线程池或进程池。否则，它们会阻塞事件循环。

5. 什么时候使用 `asyncio`？

asyncio + async/await 最适合处理大量 I/O 密集型且并发数量很高的任务。

优点:
- 高效: 在单线程内管理大量并发连接，上下文切换开销远小于多线程。
- 资源占用少: 每个协程的内存开销很小。
- 易于推理: async/await 语法使得异步代码结构清晰，避免了回调地狱。
缺点:
- 生态系统仍需完善: 并非所有库都提供了异步版本，有时需要适配或寻找替代品。
- 病毒性: 一旦开始使用 async/await，相关的代码（调用链上的函数）往往也需要变成协程。
- 不适合 CPU 密集型任务: asyncio 是单线程的，CPU 密集型任务会阻塞事件循环。CPU 密集型任务应该使用多进程或 loop.run_in_executor(ProcessPoolExecutor, ...)。
- 调试相对复杂: 调试异步代码比同步代码或传统多线程代码稍复杂一些。

总结使用场景:

构建高性能网络服务（Web 服务器、API 网关、爬虫）
需要同时连接和处理大量客户端的服务器
需要同时访问多个外部服务（微服务调用、并行 API 请求）
处理大量数据库连接或文件读写
需要高效的消息队列消费者

6. 异步编程的最佳实践和常见陷阱

不要在协程中执行阻塞同步代码: 这是最常见的错误。使用 asyncio 提供的异步函数（如 asyncio.sleep, await socket.recv 等），或使用 loop.run_in_executor() 处理必须使用的同步阻塞函数。
理解 await 的作用: await 是暂停和让出控制权的关键。如果你 await 一个可等待对象，当前协程会暂停。如果你忘了 await 一个协程或 Task，它可能根本不会运行，或者不会等待它完成。
正确启动异步任务: 使用 asyncio.run() 运行顶层协程。在协程内部，使用 asyncio.create_task() 创建并调度新的并发任务，然后使用 asyncio.gather() 或单独 await 任务来等待它们完成。
使用 async with 和 async for: 许多异步库提供了支持异步上下文管理器 (async with) 和异步迭代器 (async for) 的对象。它们提供了更安全、更简洁的资源管理和迭代方式。
处理取消 (Cancellation): 优雅地处理 asyncio.CancelledError，确保在任务被取消时能释放资源（锁、文件句柄、网络连接等）。
使用异步同步原语: 在多个协程访问共享资源时，务必使用 asyncio.Lock, asyncio.Semaphore 等异步原语，而不是 threading.Lock。
异常处理: 在协程中使用 try...except 块。asyncio.gather 默认会在第一个异常发生时取消其他任务并抛出异常。可以使用 return_exceptions=True 来改变此行为。
命名任务: 使用 asyncio.create_task(coro, name="my_task") 为任务命名，这有助于调试和监控。
调试模式: 开启 asyncio 的调试模式 (asyncio.run(..., debug=True) 或设置环境变量 PYTHONASYNCIODEBUG=1) 可以获得更详细的日志信息和警告。

7. 进一步学习

官方文档: Python asyncio 的官方文档是权威的学习资源，虽然有时比较底层，但非常详细。
asyncio 源代码: 阅读 asyncio 库的源代码是理解其工作原理的绝佳方式。
第三方异步库文档: 学习 aiohttp, asyncpg, FastAPI 等库的文档，了解如何在实际应用中使用异步。
实践项目: 尝试用 asyncio 重写一个简单的爬虫、一个 Web 服务器或一个客户端程序。

总结

Python 的 asyncio 和 async/await 语法为处理 I/O 密集型并发任务提供了一种强大且高效的解决方案。通过事件循环、协程、任务和可等待对象等概念，它使得在单线程内管理成千上万个并发操作成为可能，极大地提高了程序的性能和可伸缩性，尤其适用于构建现代网络应用和高性能服务。掌握 async/await 是成为一名优秀 Python 开发者的重要一步，它将打开通往高性能异步世界的大门。虽然一开始可能需要适应新的编程范式，但一旦理解了其核心原理并遵循最佳实践，你就能写出更高效、更具响应性的 Python 程序。

Python Async/Await 异步编程深度指南

前言：告别阻塞，拥抱效率

1. 什么是异步编程？同步与异步、阻塞与非阻塞

2. Python 中的异步编程：asyncio 和 async/await

核心概念：

async def：定义协程函数

直接调用 async def 函数会返回一个协程对象，而不是执行它

如何运行协程？需要事件循环

await: 暂停与等待

使用 asyncio.run() 运行顶层协程 (Python 3.7+)

asyncio.run() 负责创建和关闭事件循环

输出大致会是这样 (顺序可能因调度略有不同):

Running with asyncio.run…

Main coroutine started

Creating tasks…

Tasks created, waiting using asyncio.gather…

Task 1 started

Task 2 started

Task 2 finished

Task 1 finished

Main coroutine finished

注意 Task 2 (1秒) 在 Task 1 (2秒) 之前完成，证明是并发执行

asyncio.run()：运行顶层协程

手动管理事件循环 (了解)

Python 3.6 及更早版本或需要精细控制时

loop = asyncio.get_event_loop() # 获取事件循环

try:

loop.run_until_complete(simple_coro()) # 运行直到协程完成

finally:

loop.close() # 关闭事件循环

Python 3.7+ 推荐使用 asyncio.run()，它内部做了类似的事情

3. asyncio 的主要组件和功能

3.1 asyncio.sleep()

asyncio.run(main_sleep())

Output:

Main sleep start

Task A: Starting sleep 3

Task B: Starting sleep 1

Task C: Starting sleep 2

Task B: Finished sleep 1

Task C: Finished sleep 2

Task A: Finished sleep 3

Main sleep finished, results: [‘Task A result’, ‘Task B result’, ‘Task C result’]

注意任务完成的顺序与它们的 delay 相关，而不是启动顺序

3.2 asyncio.create_task() 和 asyncio.gather()

3.3 同步原语 (Synchronization Primitives)

asyncio.run(main_lock())

Output shows tasks acquiring lock one by one, not simultaneously

Task A: Trying to acquire lock

Task A: Acquired lock

Task A: Releasing lock

Task B: Trying to acquire lock

Task B: Acquired lock

Task B: Releasing lock

Task C: Trying to acquire lock

Task C: Acquired lock

Task C: Releasing lock

asyncio.run(main_semaphore())

Output shows only 2 tasks acquired semaphore at any time

asyncio.run(main_event())

Output:

Task Reader1: Waiting for event

Task Reader2: Waiting for event

Setter: Sleeping for 1 before setting event

Setter: Setting the event

Task Reader1: Event received, continuing

Task Reader2: Event received, continuing

asyncio.run(main_queue())

3.4 处理阻塞代码 (loop.run_in_executor)

asyncio.run(main_executor())

Output shows non-blocking parts can run while the blocking function is in the background thread

Starting non-blocking part

Submitted blocking function to executor, continuing other tasks

Waiting for blocking function to complete…

Executing blocking function for 2 seconds…

Blocking function finished.

Blocking function result: Result from blocking after 2 seconds

Main executor finished

Note: Sleep messages from asyncio.sleep tasks might interleave depending on exact timing

3.5 取消任务 (task.cancel())

2. Python 中的异步编程：`asyncio` 和 `async/await`

`async def`：定义协程函数

`await`: 暂停与等待

`asyncio.run()`：运行顶层协程

3. `asyncio` 的主要组件和功能

3.1 `asyncio.sleep()`

3.2 `asyncio.create_task()` 和 `asyncio.gather()`

3.4 处理阻塞代码 (`loop.run_in_executor`)

3.5 取消任务 (`task.cancel()`)

5. 什么时候使用 `asyncio`？

发表评论取消回复