生成器与协程

生成器是 Python 异步编程的基石，理解它才能真正理解 asyncio。

生成器基础

yield 暂停执行

def countdown(n):
    print("开始倒计时")
    while n > 0:
        yield n          # 暂停，返回值给调用者
        n -= 1
    print("倒计时结束")

gen = countdown(3)
print(type(gen))         # <class 'generator'>

print(next(gen))         # 开始倒计时 \n 3
print(next(gen))         # 2
print(next(gen))         # 1
# next(gen)              # StopIteration

生成器是迭代器

# 生成器实现了迭代器协议
gen = countdown(3)
print(hasattr(gen, '__iter__'))  # True
print(hasattr(gen, '__next__'))  # True
print(iter(gen) is gen)          # True

# 可以直接用 for 循环
for n in countdown(3):
    print(n)

生成器表达式

# 列表推导式 — 立即计算，全部存入内存
squares_list = [x**2 for x in range(1_000_000)]

# 生成器表达式 — 惰性计算，按需生成
squares_gen = (x**2 for x in range(1_000_000))

import sys
print(sys.getsizeof(squares_list))  # ~8MB
print(sys.getsizeof(squares_gen))   # ~104 bytes

send() — 双向通信

生成器不只能输出，还能接收值：

def accumulator():
    total = 0
    while True:
        value = yield total   # yield 既返回 total，又接收 send() 的值
        if value is None:
            break
        total += value

acc = accumulator()
next(acc)           # 启动生成器（必须先 next 到第一个 yield）
print(acc.send(10)) # 10
print(acc.send(20)) # 30
print(acc.send(5))  # 35

yield from — 委托生成器

def inner():
    yield 1
    yield 2
    yield 3

def outer():
    yield 0
    yield from inner()   # 委托给 inner，透明传递
    yield 4

print(list(outer()))  # [0, 1, 2, 3, 4]

yield from 的真正威力：透明传递 send() 和异常：

def sub_gen():
    x = yield "ready"
    yield f"got: {x}"

def main_gen():
    result = yield from sub_gen()  # 透明代理

gen = main_gen()
print(next(gen))          # ready
print(gen.send("hello"))  # got: hello

协程：从生成器到 async/await

基于生成器的协程（旧式，了解即可）

# Python 3.4 之前的协程写法
def old_coroutine():
    data = yield "waiting for data"
    result = yield f"processing {data}"
    return result

原生协程 async def

import asyncio

async def fetch(url):
    print(f"开始请求 {url}")
    await asyncio.sleep(1)   # 模拟 I/O 等待
    print(f"完成请求 {url}")
    return f"data from {url}"

async def main():
    # 并发执行多个协程
    results = await asyncio.gather(
        fetch("http://api1.example.com"),
        fetch("http://api2.example.com"),
        fetch("http://api3.example.com"),
    )
    print(results)

asyncio.run(main())
# 三个请求并发执行，总耗时约 1s 而非 3s

协程 vs 线程

import asyncio
import threading
import time

# 协程：单线程，协作式调度
async def async_task(name, delay):
    await asyncio.sleep(delay)
    return name

# 线程：多线程，抢占式调度
def thread_task(name, delay):
    time.sleep(delay)
    return name

# 协程更轻量：可以轻松创建 10 万个
async def many_coroutines():
    tasks = [async_task(i, 0.001) for i in range(100_000)]
    results = await asyncio.gather(*tasks)
    print(f"完成 {len(results)} 个协程")

实用生成器模式

无限序列

def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

from itertools import islice
print(list(islice(fibonacci(), 10)))  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

管道处理

def read_lines(filename):
    with open(filename) as f:
        yield from f

def filter_empty(lines):
    for line in lines:
        if line.strip():
            yield line.strip()

def parse_csv(lines):
    for line in lines:
        yield line.split(',')

# 组合成处理管道，内存高效
def process_file(filename):
    lines = read_lines(filename)
    lines = filter_empty(lines)
    rows = parse_csv(lines)
    return rows

上下文管理器生成器

from contextlib import contextmanager

@contextmanager
def managed_resource(name):
    print(f"获取资源: {name}")
    resource = {"name": name, "active": True}
    try:
        yield resource          # 暂停，将资源交给 with 块
    except Exception as e:
        print(f"处理异常: {e}")
        raise
    finally:
        resource["active"] = False
        print(f"释放资源: {name}")

with managed_resource("数据库连接") as res:
    print(f"使用资源: {res}")
# 输出：
# 获取资源: 数据库连接
# 使用资源: {'name': '数据库连接', 'active': True}
# 释放资源: 数据库连接

异步生成器

async def async_range(start, stop, delay=0.1):
    for i in range(start, stop):
        await asyncio.sleep(delay)
        yield i

async def main():
    async for value in async_range(0, 5):
        print(value)

    # 异步生成器表达式
    result = [v async for v in async_range(0, 3, delay=0)]
    print(result)  # [0, 1, 2]

asyncio.run(main())

关键理解

async/await 本质上是生成器的语法糖。await 相当于 yield from，事件循环通过 send() 驱动协程执行。理解了生成器，asyncio 就不再神秘。