Python 中的协程 (Coroutines) 与 asyncio 异步编程从入门到精通

在传统的单线程同步编程中，当程序执行到 I/O 操作（如网络请求、读写文件、数据库查询）时，CPU 会处于闲置状态，静静等待 I/O 操作完成。这种“阻塞”模式在面临高并发场景（如高频爬虫、实时通信、微服务 API）时，会导致系统吞吐量极低。

为了在不使用多线程/多进程（避免昂贵的线程上下文切换和锁竞争）的情况下解决高并发问题，Python 引入了基于事件循环的异步编程模型。其中，协程（Coroutines）与 asyncio 标准库是这一模型的核心武器。

本文将带你从零开始，理清异步编程中的核心概念，探索协程的演进历史，并通过实战对比，帮助你彻底掌握 Python 异步编程。

一、 核心概念：同步/异步与阻塞/非阻塞

在学习 asyncio 之前，我们必须厘清两组经常被混淆的概念：

• 同步 (Synchronous)：任务按顺序执行，前一个任务没结束，后一个任务就必须等待。
• 异步 (Asynchronous)：任务可以交替执行，无需等待前一个任务彻底完成，即可启动下一个任务。
• 阻塞 (Blocking)：当程序调用一个 I/O 操作时，在收到结果前，当前调用方的控制权会被系统收回，程序“卡住”无法继续执行。
• 非阻塞 (Non-blocking)：调用 I/O 操作后，系统立即返回控制权，调用方可以继续做其他事，后续通过轮询或通知来获取 I/O 结果。

异步编程的目的，就是用单线程实现异步非阻塞的执行流程。

二、 协程的演进历史

Python 对协程的支持经历了一个漫长的演进过程，主要分为三个阶段：

1. 基于生成器 (Generator-based) 的协程

早期的 Python 通过生成器中的 yield 关键字来实现协程。yield 不仅能返回数据，还能暂停函数的执行，并通过 send() 方法向函数内部传递数据。

def simple_coroutine():
    print("-> 协程开始启动")
    x = yield "等待外部输入..."
    print(f"-> 接收到外部输入: {x}")

coro = simple_coroutine()
# 启动协程（预激）
status = next(coro)
print(status)  # 输出: 等待外部输入...
# 向协程发送数据并恢复执行
try:
    coro.send(42)
except StopIteration:
    print("-> 协程执行结束")

2. yield from 语法 (Python 3.3)

为了方便协程的嵌套和调用，Python 3.3 引入了 yield from 语法，允许一个生成器将部分操作委托给另一个生成器，这也是现代 await 的雏形。

3. 原生协程 async / await (Python 3.5+)

为了让协程的定义和语义更加清晰，Python 3.5 正式引入了 async def 和 await 关键字，标志着原生协程的诞生。

async def my_coroutine():
    # 使用 await 挂起当前协程，将控制权交还给事件循环
    await asyncio.sleep(1)
    return "Done"

三、 asyncio 核心组件与基本用法

asyncio 库通过一个事件循环 (Event Loop) 来调度和执行所有的协程。其核心用法非常直观：

1. 声明与运行协程

使用 async def 声明的函数是一个协程函数，直接调用它不会执行函数体，而是会返回一个协程对象。我们必须通过事件循环来运行它。

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)  # 模拟非阻塞的延时
    print("World")

# 使用 asyncio.run() 启动事件循环并运行协程
asyncio.run(hello())

2. 并发运行多个任务：asyncio.gather

如果我们需要并发运行多个协程，可以使用 asyncio.gather() 将它们打包，事件循环会自动在它们之间进行非阻塞切换。

import asyncio
import time

async def fetch_data(id, delay):
    print(f"任务 {id}: 开始获取数据...")
    await asyncio.sleep(delay)
    print(f"任务 {id}: 数据获取成功！")
    return f"Data_{id}"

async def main():
    start = time.perf_counter()
    # 并发运行三个任务
    results = await asyncio.gather(
        fetch_data(1, 2),
        fetch_data(2, 3),
        fetch_data(3, 1)
    )
    end = time.perf_counter()
    print(f"所有任务执行完毕，结果: {results}")
    print(f"总耗时: {end - start:.2f} 秒") # 总耗时约为 3 秒（即最大延时）

asyncio.run(main())

四、 实战对比：同步 vs 异步网页下载器

为了展现异步编程的真实威力，我们对比一下传统的同步 requests 爬虫与基于 aiohttp 库的异步爬虫在下载多个网页时的速度差异。

1. 同步版本 (requests)

import requests
import time

urls = ["https://httpbin.org/delay/2"] * 5  # 模拟 5 个耗时 2 秒的请求

def download_sync():
    start = time.perf_counter()
    for i, url in enumerate(urls):
        resp = requests.get(url)
        print(f"同步下载完成 {i+1}: {resp.status_code}")
    end = time.perf_counter()
    print(f"同步总耗时: {end - start:.2f} 秒")

download_sync()  # 总耗时约为 10 秒以上

2. 异步版本 (aiohttp + asyncio)

import asyncio
import aiohttp
import time

urls = ["https://httpbin.org/delay/2"] * 5

async def download_page(session, i, url):
    async with session.get(url) as response:
        status = response.status
        print(f"异步下载完成 {i+1}: {status}")
        return status

async def download_async():
    start = time.perf_counter()
    async with aiohttp.ClientSession() as session:
        tasks = [download_page(session, i, url) for i, url in enumerate(urls)]
        await asyncio.gather(*tasks)
    end = time.perf_counter()
    print(f"异步总耗时: {end - start:.2f} 秒")

asyncio.run(download_async())  # 总耗时仅需约 2 秒多！

从上面的对比中可以看出，异步编程利用网络 I/O 等待的空闲时间并发处理其他请求，使得执行效率提升了数倍！

五、 异步编程的避坑指南

1. 绝对不要在协程中调用阻塞函数： 在 async def 中，如果你使用了阻塞库（如 time.sleep() 或 requests.get()），整个事件循环都会被挂起（卡死），异步退化为同步。必须使用 asyncio.sleep() 或 aiohttp。
2. 如果必须调用阻塞或 CPU 密集型任务： 如果使用的库没有异步版本，或者必须执行 CPU 密集型的计算，可以使用 loop.run_in_executor() 将其委托给多线程/多进程线程池执行，从而不阻塞主线程的事件循环。
3. 异常处理： 在使用 asyncio.gather() 时，如果其中一个协程抛出异常，默认情况下会直接向上抛出，但其他未运行完的协程仍会继续运行。可以使用 return_exceptions=True` 参数，将异常作为结果返回，避免中断。