Python 异步编程测验：从生成器到 asyncio

2026-08-31 42 预计阅读时间: 1 分钟
来源: realpython.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间：7 分钟

Python 的异步编程并不是把所有函数前面加上 async 就结束了。理解生成器、async/await、阻塞与非阻塞操作，以及事件循环之间的关系，才能判断一段代码究竟能否并发执行。

这篇文章把这些概念串起来，并用一个可以直接运行的示例帮助你验证理解。

生成器为何是重要起点

普通函数通过 return 返回结果后就结束了，而生成器使用 yield 暂停执行，并在下一次迭代时从暂停位置继续。它不会一次性把所有数据加载到内存中，也体现了“暂停、恢复”的执行模型。

def numbers():
    print("生成 1")
    yield 1
    print("生成 2")
    yield 2

for number in numbers():
    print("收到:", number)

运行时可以看到，函数会在每次 yield 处暂停。这种可暂停的控制流，为后来的协程和异步任务提供了重要思想基础。不过，生成器本身并不等于异步代码；它不会自动让阻塞的网络或文件操作变成非阻塞操作。

async 和 await 到底做什么

使用 async def 定义的函数称为协程函数。调用它时不会立即执行函数体，而是返回一个协程对象。只有把它交给事件循环，或者在另一个协程中使用 await，代码才会真正运行。

await 的含义不是“创建一个新线程”，而是告诉事件循环：当前协程暂时没有可继续执行的工作，可以把执行机会交给其他任务。

下面的例子模拟两个非阻塞等待任务：

import asyncio


async def fetch(name: str, delay: float) -> str:
    print(f"{name} 开始")
    await asyncio.sleep(delay)
    print(f"{name} 完成")
    return f"{name} 的结果"


async def main() -> None:
    results = await asyncio.gather(
        fetch("任务 A", 1.0),
        fetch("任务 B", 0.5),
    )
    print(results)


if __name__ == "__main__":
    asyncio.run(main())

将代码保存为 async_demo.py，然后运行：

python async_demo.py

任务 B 会先完成，但整个程序大约只需要等待 1 秒，而不是把两个等待时间简单相加。这是因为 asyncio.sleep() 会暂停当前协程，同时让事件循环运行其他就绪任务。

阻塞代码与非阻塞代码

异步程序的并发能力取决于协程是否愿意把控制权交还给事件循环。await asyncio.sleep() 是非阻塞等待；而在协程中直接调用 time.sleep()，会阻塞整个事件循环。

import asyncio
import time


async def blocking_task() -> None:
    print("阻塞任务开始")
    time.sleep(2)
    print("阻塞任务完成")


async def non_blocking_task() -> None:
    print("非阻塞任务开始")
    await asyncio.sleep(0.5)
    print("非阻塞任务完成")


async def main() -> None:
    await asyncio.gather(blocking_task(), non_blocking_task())


asyncio.run(main())

虽然这里使用了 gather()，但 time.sleep(2) 运行期间事件循环无法调度其他协程，非阻塞任务也会被拖住。实际项目中，常见的阻塞来源包括同步 HTTP 客户端、同步数据库驱动、文件 I/O，以及复杂的 CPU 密集型计算。

可以这样改造阻塞调用：

import asyncio
import time


def blocking_work() -> str:
    time.sleep(2)
    return "完成"


async def main() -> None:
    result = await asyncio.to_thread(blocking_work)
    print(result)


asyncio.run(main())

asyncio.to_thread() 会把同步函数放到线程中执行，避免它直接卡住事件循环。它更适合 I/O 型阻塞操作；对于真正的 CPU 密集型任务，进程池或专门的任务队列通常更合适。

事件循环与常见判断题

事件循环负责管理协程和任务：当一个任务遇到可等待的异步操作时，它会暂停当前任务，并尝试运行其他已经准备好的任务。

理解下面几条规则，可以解决大多数异步编程测验题：

  • async def 定义的是协程函数，调用它只会得到协程对象。
  • 协程对象需要通过 await、asyncio.run() 或任务调度 API 执行。
  • asyncio.create_task() 会把协程安排到当前运行中的事件循环中。
  • asyncio.gather() 可以并发等待多个可等待对象，并按传入顺序返回结果。
  • await 不等于开启线程，也不等于并行执行 CPU 计算。
  • 单线程事件循环可以并发处理 I/O 等待，但不能消除同步阻塞代码的影响。
  • 一个事件循环中不能随意嵌套调用 asyncio.run()；在已经运行的协程里通常应直接使用 await。

使用前的检查清单

把同步代码改成异步代码时，可以逐项确认：

  1. 外层调用链是否已经进入异步上下文。
  2. 网络、数据库等依赖是否提供真正的异步 API。
  3. 是否误用了 time.sleep() 或同步客户端。
  4. 多个独立 I/O 操作是否适合使用 gather()。
  5. 任务是否需要超时、取消和异常处理。
  6. CPU 密集型工作是否应该交给进程池或后台任务系统。

异步编程的关键不是语法数量，而是控制权如何在事件循环中的任务之间流动。掌握“协程何时暂停、谁负责恢复、哪些调用会阻塞”这三件事，才能从生成器自然过渡到 asyncio，并准确判断一段 Python 异步代码的实际行为。

# AI # Python # 后端

相关推荐