生成器与协程

1. 学习目标

本章节对标 MIT 6.001 结构与解释、Stanford CS143 编译原理、CMU 15-410 操作系统、UC Berkeley CS61A 程序设计等顶级高校课程对控制流抽象与协程理论的教学水准，系统讲解 Python 生成器与协程的工程化使用与底层理论。完成本章节学习后，读者应能够：

1.1 Bloom 认知层级目标

1.2 知识地图

[理论基础] 子例程 vs 协程 | CPS 变换 | 有限状态机 | 惰性求值
    ↓
[Python 实现] 生成器函数 | 生成器对象 | yield 表达式 | send/throw/close
    ↓
[委托机制] yield from | 子生成器 | 返回值传播 | 异常透明转发
    ↓
[工程实战] 流式处理 | 管道组合 | 分页 | 树遍历 | 状态机
    ↓
[高级话题] 异步生成器 | 协程调度 | 生成器装饰器 | 暂停-恢复模型

1.3 前置知识检查

学习本章节前，请确认你已掌握：

1. Python 函数定义、参数传递、作用域规则；
2. 迭代器协议（__iter__、__next__、StopIteration）；
3. 装饰器的基本用法（参考《装饰器进阶》）；
4. 上下文管理器与 with 语句（参考《上下文管理器》）；
5. 异常处理机制（try/except/finally、自定义异常）；
6. 集合论与有限状态机的基本概念。

2. 历史动机与发展脉络

生成器与协程并非 Python 独创，其理论根源可追溯至 1958 年的协程概念与 1970 年代的惰性求值思想。理解这一脉络有助于把握 Python 设计的取舍。

2.1 协程的理论起源

• 1958：Melvin Conway 提出”协程”（coroutine）概念，用于 COBOL 编译器的词法分析。其核心是”对称的多入口函数”，与子例程（subroutine）的”调用-返回”模型相对；
• 1963：Joel W. W. 在 Simula 67 中实现”准并发”协程；
• 1972：CLU 语言（Barbara Liskov）引入迭代器抽象，奠定生成器的语言基础；
• 1975：Peter J. Landin 在 ISWIM 中使用流（stream）实现惰性求值；
• 1976：Rod Burstall 与 John Darlington 在 NPL 中明确”生成器即惰性列表”；
• 1980：C. D. Marlin 在《Coroutines: A Programming Methodology, a Language Design and an Implementation》中系统形式化协程语义；
• 1988：SML/NJ 引入”延迟序列”（lazy sequence）；
• 1995：Icon 语言将生成器作为一等公民，所有表达式皆可回溯；
• 1999：Python 1.5.1 引入 __getitem__ 协议的隐式迭代；
• 2001：Python 2.2（PEP 255）正式引入生成器。

2.2 Python 生成器与协程演进时间线

2.3 PEP 255 的设计目标

PEP 255（Simple Generators，由 Neil Schemenauer、Tim Peters、Magnus Lie Hetland 于 2001 年提出）明确生成器的设计动机：

1. 简化迭代器编写：传统迭代器需手写 __iter__ 与 __next__，逻辑分散；生成器自动保存局部状态；
2. 惰性求值：仅在 next() 调用时计算下一值，避免一次性生成大量数据；
3. 流式处理：处理无穷序列或大文件时无需全部加载到内存；
4. 状态机简化：复杂状态可保存在生成器局部变量中，无需显式状态字段；
5. 协程雏形：虽 2.2 版本未提供 send，但 yield 的暂停-恢复语义已奠定协程基础。

2.4 PEP 342 与协程觉醒

PEP 342（Coroutines via Enhanced Generators，由 Guido van Rossum、Phillip J. Eby 于 2005 年提出）将生成器升级为协程：

1. yield 表达式化：x = yield value 允许双向通信；
2. send(value)：从外部向生成器注入值；
3. throw(type, value, traceback)：从外部向生成器注入异常；
4. close()：显式关闭生成器，触发 GeneratorExit；
5. GeneratorExit 异常：标识生成器被关闭；
6. return 在生成器中合法：返回值通过 StopIteration.value 传递（Python 3.3+ 明确）。

PEP 342 的核心洞察：协程 = 生成器 + 双向通信 + 异常注入。Eby 在《Generator Tricks for Systems Programmers》中展示了基于生成器的协程可用于状态机、管道、协程调度器，催生了 Twisted 的 inlineCallbacks、Tornado 的 gen.coroutine、asyncio 的早期实现。

2.5 PEP 380 与 yield from

PEP 380（Syntax for Delegating to a Subgenerator，由 Greg Ewing 于 2011 年提出）引入 yield from：

1. 委托子生成器：yield from subgen() 自动转发 send、throw、close；
2. 返回值传播：return value 在子生成器中的值成为 yield from 表达式的值；
3. 简化组合：避免手写 for x in subgen: yield x 的样板代码；
4. asyncio 基石：在 async def 出现前，yield from 是 asyncio 的核心语法；
5. 协程组合原语：为后来的 await 提供了语义基础。

2.6 PEP 492 与协程语法升级

PEP 492（Coroutines with async and await syntax，由 Yury Selivanov 于 2015 年提出）将协程从生成器机制中独立出来：

1. async def：原生协程，与生成器显式区分；
2. await：取代 yield from 用于异步上下文；
3. async for/async with：异步迭代与异步上下文管理；
4. __await__ 协议：统一 awaitable 对象接口；
5. PEP 525 异步生成器：async def + yield，允许在协程中产出值。

2.7 与其他语言的对比

关键差异：

• 有栈协程（Lua、Ruby Fiber、Go goroutine）：每个协程有独立栈，可在任意嵌套深度挂起；
• 无栈协程（Python、JavaScript、Rust、C#）：协程状态保存在堆上的对象中，只能在 yield/await 处挂起；
• 对称协程（Lua、Go）：协程间平等切换；
• 非对称协程（Python、Ruby）：协程与调用者之间是主从关系，yield 总是返回到调用点。

3. 形式化定义

3.1 协程的代数定义

设 $S$ 为程序状态集合（包括局部变量、指令指针、操作数栈），$V$ 为值集合，$E$ 为事件集合（包括 send、throw、close）。协程（coroutine）定义为五元组：

其中：

• $S$ 为可枚举的状态集合；
• $s_0 \in S$ 为初始状态；
• $E$ 为事件集合；
• $T : S \times E \to S$ 为状态转移函数；
• $Y : S \to V \times S$ 为 yield 函数，产出值并返回新状态。

子例程（subroutine） 是协程的特例，其转移函数满足”严格调用-返回”约束：

协程放宽了这一约束：允许在任意 yield 点挂起，并从该点恢复执行。

3.2 生成器状态机

Python 生成器对象在执行过程中处于以下状态之一：

状态转移图：

形式化地，生成器的状态转移可表示为：

关键性质：

1. GEN_RUNNING 状态下不可再次调用 next/send（防止递归调用同一生成器）；
2. GEN_CLOSED 状态下任何操作均抛出 StopIteration；
3. 生成器对象不可重入，迭代完毕后需重新创建。

3.3 yield 表达式的形式化语义

yield expr 是一个表达式，其求值规则可形式化为：

执行流程：

1. 求值 expr 得到 $v_{\text{yield}}$；
2. 保存当前帧栈状态 $s'$ 到生成器对象；
3. 挂起执行，向调用者返回 $v_{\text{yield}}$（封装为 (value, done=False)）；
4. 等待恢复事件 $e$；
5. 当 send(value) 或 next() 触发恢复时，yield 表达式的值为：
   • send(v) 时：$v_{\text{resume}} = v$；
   • next() 时：$v_{\text{resume}} = \text{None}$。

3.4 send / throw / close 的操作语义

设生成器 $g$ 当前状态为 $s$：

send(value)：

throw(type, value, traceback)：

close()：

3.5 yield from 的委托语义

yield from expr 形式化为：

完整等价语义（PEP 380 官方说明）：

_i = iter(EXPR)
try:
    _y = next(_i)
except StopIteration as _e:
    _r = _e.value
else:
    while 1:
        try:
            _s = yield _y
        except GeneratorExit as _e:
            _i.close()
            raise _e
        except BaseException as _e:
            _m = _i.throw
            if _m:
                _y = _m(_e)
            else:
                raise _e
        else:
            try:
                _y = _i.send(_s)
            except StopIteration as _e:
                _r = _e.value
                break
RESULT = _r

关键语义：

1. send、throw、close 透明转发到子生成器；
2. 子生成器的 return value 成为 yield from 表达式的值；
3. GeneratorExit 不被透明转发，而是触发子生成器的 close()。

3.6 协程与生成器的关系

Python 中协程与生成器的关系经历了三个阶段：

阶段一（Python 2.5 - 3.4）：基于生成器的协程

阶段二（Python 3.5 - 3.10）：原生协程与生成器并存

阶段三（Python 3.11+）：原生协程主导

形式化地，原生协程与生成器的核心区别：

4. 理论推导与原理解析

4.1 CPS 变换与生成器状态机

CPS（Continuation-Passing Style）变换 是协程编译的核心技术。给定一个直接风格的函数：

def f(x):
    y = g(x)
    return y + 1

CPS 变换后：

def f_cps(x, k):
    def k_g(y):
        k(y + 1)
    g_cps(x, k_g)

每个”剩余计算”被显式封装为续延（continuation）k。Python 生成器通过保存帧栈实现了类似效果：yield 点相当于”分界点”，恢复执行相当于”调用续延”。

形式化地，生成器的状态转移可视为 CPS 中的续延链：

每次 yield 保存当前续延，next/send 调用该续延。

4.2 协作式调度的数学分析

协程采用协作式调度（cooperative scheduling）：

设 $n$ 个协程组成的协作调度系统，每个协程 $c_i$ 的执行时间为 $t_i$，yield 频率为 $f_i$：

• 平均响应时间：$T_{\text{resp}} = \sum_{i} t_i / f_i$；
• 公平性：若所有 $f_i$ 相等，则 $T_{\text{resp}}^{(i)} = T_{\text{resp}}^{(j)}, \forall i, j$；
• 饥饿风险：若某协程 $c_k$ 长时间不 yield，其他协程被阻塞。

与抢占式调度的对比：

4.3 生成器的内存模型

CPython 中生成器对象 PyGenObject 的内存布局（CPython 3.12）：

typedef struct {
    PyObject_HEAD
    PyCodeObject *gi_code;
    PyObject *gi_weakreflist;
    PyObject *gi_name;
    PyObject *gi_qualname;
    char gi_hooks_initing;
    char gi_hooks_active;
    char gi_resume_frame_depth;
    _PyGenObjectFrame *gi_frame;
    PyObject *gi_origin_or_finalizer;
    char gi_frame_state;  // CREATED, SUSPENDED, RUNNING, COMPLETED
    PyObject *gi_cr_name;
    ...
} PyGenObject;

内存开销：

• PyObject_HEAD：16 字节；
• gi_code 指针：8 字节（共享函数代码对象，不重复存储）；
• gi_frame 指针：8 字节（仅挂起时分配，包含局部变量、操作数栈、指令指针）；
• 其他字段：约 32 字节；
• 总计：~80 字节基础开销 + 帧栈（按需分配）。

帧栈保存策略：

1. 生成器调用 next() 时，从 gi_frame 恢复执行；
2. yield 时，保存当前帧栈到 gi_frame；
3. 生成器结束时，释放 gi_frame；
4. 与函数调用栈不同，生成器帧栈在堆上分配，生命周期独立于调用者。

4.4 yield from 的代数性质

yield from 满足以下代数性质：

1. 单位律（Identity）：

2. 结合律（Associativity）：

3. 分配律（Distributes over composition）：

其中 $\oplus$ 表示 itertools.chain。

4. 返回值传播：

这些性质使 yield from 成为协程组合的代数基础，与 await 的语义一致。

4.5 生成器与单子

生成器可视为 状态单子（State Monad）的实例：

• return/pure：yield value 将值注入生成器序列；
• bind/flatMap：yield from gen 将子生成器展平到父生成器。

满足单子三定律：

1. 左单位律：yield from (lambda x: yield x)(v) ≡ yield v；
2. 右单位律：yield from (lambda: yield v)() ≡ yield v；
3. 结合律：yield from (yield from g1) g2 ≡ yield from g1 then yield from g2。

这使得生成器可以像列表一样组合，同时保持惰性求值。

4.6 生成器与惰性求值

生成器实现了”按需计算”（call-by-need）：

其中 thunk 是延迟计算的”索求”。每次 next() 强制求值一个 thunk，得到下一个值与新 thunk。

与 Haskell 的对比：

-- Haskell 惰性列表
ones = 1 : ones
take 3 ones  -- [1, 1, 1]

# Python 生成器
def ones():
    while True:
        yield 1

import itertools
list(itertools.islice(ones(), 3))  # [1, 1, 1]

Python 通过显式 next() 控制 thunk 求值时机，Haskell 通过运行时自动管理。

5. 代码示例

5.1 基本生成器

"""生成器基础示例。Python 3.12+。"""
from __future__ import annotations

import sys
from collections.abc import Generator, Iterator


def count_up(n: int) -> Generator[int, None, None]:
    """生成 0 到 n-1 的整数序列。

    Args:
        n: 上界（不含）。

    Yields:
        0, 1, 2, ..., n-1
    """
    print(f"[count_up] 启动，n={n}")
    i = 0
    while i < n:
        print(f"[count_up] yield {i}")
        yield i
        i += 1
    print(f"[count_up] 结束")


def basic_generator_demo() -> None:
    """基本生成器演示。"""
    gen = count_up(3)
    print(f"类型: {type(gen).__name__}")  # generator
    print(f"是否为迭代器: {isinstance(gen, Iterator)}")  # True
    print(f"内存大小: {sys.getsizeof(gen)} 字节")  # ~112 字节

    # 逐个获取值
    print(next(gen))  # 触发执行到第一个 yield
    print(next(gen))  # 从上次 yield 处恢复
    print(next(gen))

    try:
        next(gen)  # 触发 StopIteration
    except StopIteration:
        print("生成器已结束")


if __name__ == "__main__":
    basic_generator_demo()

5.2 send 双向通信

"""send 方法：双向通信。Python 3.12+。"""
from __future__ import annotations

from collections.abc import Generator


def accumulator() -> Generator[float, float, None]:
    """累加器协程：接收值并返回当前总和。

    Yields:
        当前累加值。

    Receives:
        要累加的新值。
    """
    total: float = 0.0
    while True:
        value = yield total
        if value is None:
            break
        total += value


def send_demo() -> None:
    """send 方法演示。"""
    gen = accumulator()

    # 启动生成器（必须先 next 或 send(None)）
    initial = next(gen)
    print(f"初始值: {initial}")  # 0.0

    # 向生成器发送值
    print(gen.send(10))   # 10.0
    print(gen.send(20))   # 30.0
    print(gen.send(5.5))  # 35.5

    # 关闭生成器
    gen.close()


if __name__ == "__main__":
    send_demo()

5.3 throw 与 close

"""throw 与 close 方法。Python 3.12+。"""
from __future__ import annotations

from collections.abc import Generator


def robust_processor() -> Generator[str, str, None]:
    """可处理异常的协程。

    接收字符串，返回大写形式。
    遇到 ValueError 时记录并继续。
    遇到 GeneratorExit 时清理资源。
    """
    print("[processor] 启动")
    buffer: list[str] = []
    try:
        while True:
            try:
                value = yield
                if value is None:
                    continue
                upper = value.upper()
                buffer.append(upper)
                print(f"[processor] 处理: {value} -> {upper}")
            except ValueError as e:
                print(f"[processor] 警告: {e}")
    except GeneratorExit:
        print(f"[processor] 关闭，已处理 {len(buffer)} 条")
        # 这里可以执行清理逻辑
        raise  # 必须重新抛出 GeneratorExit


def throw_close_demo() -> None:
    """throw 与 close 演示。"""
    gen = robust_processor()
    next(gen)  # 启动

    gen.send("hello")
    gen.send("world")

    # 注入异常
    gen.throw(ValueError, "无效输入")
    gen.send("recover")

    # 关闭生成器
    gen.close()


if __name__ == "__main__":
    throw_close_demo()

5.4 yield from 委托

"""yield from 委托语义。Python 3.12+。"""
from __future__ import annotations

from collections.abc import Generator


def inner() -> Generator[str, str, str]:
    """子生成器：接收值并返回最终结果。"""
    print("[inner] 启动")
    received = []
    while True:
        try:
            value = yield f"echo: {value}" if (value := (yield "ready")) else "idle"
        except GeneratorExit:
            break
        received.append(value)
    print(f"[inner] 收到: {received}")
    return f"inner_done({len(received)})"


def simpler_inner() -> Generator[str, str, str]:
    """简化版子生成器。"""
    print("[simpler_inner] 启动")
    count = 0
    try:
        while True:
            value = yield f"step {count}"
            count += 1
            if value == "stop":
                break
    except GeneratorExit:
        pass
    return f"completed_at_{count}"


def outer() -> Generator[str, str, str]:
    """外层生成器：委托给 inner。"""
    print("[outer] 启动")
    # yield from 透明转发 send/throw/close
    result = yield from simpler_inner()
    print(f"[outer] inner 返回: {result}")
    yield "outer_final"
    return "outer_done"


def yield_from_demo() -> None:
    """yield from 演示。"""
    gen = outer()
    print(next(gen))  # 启动，输出 "step 0"
    print(gen.send(None))  # "step 1"
    print(gen.send("continue"))  # "step 2"
    print(gen.send("stop"))  # 触发 inner 结束，输出 "outer_final"
    try:
        next(gen)
    except StopIteration as e:
        print(f"outer 返回: {e.value}")  # "outer_done"


if __name__ == "__main__":
    yield_from_demo()

5.5 生成器管道

"""生成器管道：流式数据处理。Python 3.12+。"""
from __future__ import annotations

from collections.abc import Generator, Iterable


def read_lines(filepath: str) -> Generator[str, None, None]:
    """逐行读取文件。"""
    with open(filepath, encoding="utf-8") as f:
        yield from f


def strip_lines(lines: Iterable[str]) -> Generator[str, None, None]:
    """去除每行首尾空白。"""
    for line in lines:
        yield line.strip()


def filter_comments(lines: Iterable[str]) -> Generator[str, None, None]:
    """过滤注释行（以 # 开头）。"""
    for line in lines:
        if not line.startswith("#"):
            yield line


def parse_csv(rows: Iterable[str]) -> Generator[list[str], None, None]:
    """解析 CSV 行。"""
    for row in rows:
        if row:
            yield row.split(",")


def transform_records(
    records: Iterable[list[str]],
) -> Generator[dict[str, str], None, None]:
    """将 CSV 行转换为字典。"""
    headers: list[str] | None = None
    for record in records:
        if headers is None:
            headers = record
            continue
        yield dict(zip(headers, record))


def pipeline_demo() -> None:
    """组合生成器管道。"""
    # 嵌套调用：数据流式处理
    filepath = "data.csv"
    pipeline = transform_records(
        parse_csv(filter_comments(strip_lines(read_lines(filepath))))
    )

    for record in pipeline:
        print(record)


# 更优雅的管道组合方式
class Pipeline:
    """可组合的生成器管道。"""

    def __init__(self, source: Iterable) -> None:
        self._source = source
        self._stages: list[callable] = []

    def pipe(self, stage: callable) -> "Pipeline":
        """添加处理阶段。"""
        self._stages.append(stage)
        return self

    def __iter__(self) -> Generator:
        data = self._source
        for stage in self._stages:
            data = stage(data)
        yield from data


def fluent_pipeline_demo() -> None:
    """流式管道 API 演示。"""
    pipeline = (
        Pipeline(["  name,age  ", "# comment", "Alice,30", "Bob,25"])
        .pipe(strip_lines)
        .pipe(filter_comments)
        .pipe(parse_csv)
        .pipe(transform_records)
    )

    for record in pipeline:
        print(record)
    # {'name': 'Alice', 'age': '30'}
    # {'name': 'Bob', 'age': '25'}


if __name__ == "__main__":
    fluent_pipeline_demo()

5.6 协程实现状态机

"""协程实现有限状态机。Python 3.12+。"""
from __future__ import annotations

from collections.abc import Generator
from enum import Enum


class TCPState(str, Enum):
    """TCP 连接状态。"""
    CLOSED = "CLOSED"
    LISTEN = "LISTEN"
    SYN_SENT = "SYN_SENT"
    SYN_RECEIVED = "SYN_RECEIVED"
    ESTABLISHED = "ESTABLISHED"
    FIN_WAIT_1 = "FIN_WAIT_1"
    FIN_WAIT_2 = "FIN_WAIT_2"
    CLOSE_WAIT = "CLOSE_WAIT"
    CLOSING = "CLOSING"
    LAST_ACK = "LAST_ACK"
    TIME_WAIT = "TIME_WAIT"


def tcp_fsm() -> Generator[TCPState, str, None]:
    """TCP 状态机协程。

    接收事件名，返回当前状态。
    """
    state = TCPState.CLOSED
    while True:
        event = yield state
        state = _transition(state, event)


def _transition(state: TCPState, event: str) -> TCPState:
    """状态转移函数。"""
    transitions = {
        (TCPState.CLOSED, "passive_open"): TCPState.LISTEN,
        (TCPState.CLOSED, "active_open"): TCPState.SYN_SENT,
        (TCPState.LISTEN, "send_syn"): TCPState.SYN_SENT,
        (TCPState.LISTEN, "rcv_syn"): TCPState.SYN_RECEIVED,
        (TCPState.SYN_SENT, "rcv_syn_ack"): TCPState.ESTABLISHED,
        (TCPState.SYN_RECEIVED, "rcv_ack"): TCPState.ESTABLISHED,
        (TCPState.ESTABLISHED, "close"): TCPState.FIN_WAIT_1,
        (TCPState.FIN_WAIT_1, "rcv_ack"): TCPState.FIN_WAIT_2,
        (TCPState.FIN_WAIT_2, "rcv_fin"): TCPState.TIME_WAIT,
        (TCPState.TIME_WAIT, "timeout"): TCPState.CLOSED,
    }
    return transitions.get((state, event), state)


def state_machine_demo() -> None:
    """状态机演示。"""
    fsm = tcp_fsm()
    next(fsm)  # 启动

    print(fsm.send("active_open"))   # SYN_SENT
    print(fsm.send("rcv_syn_ack"))   # ESTABLISHED
    print(fsm.send("close"))          # FIN_WAIT_1
    print(fsm.send("rcv_ack"))        # FIN_WAIT_2
    print(fsm.send("rcv_fin"))        # TIME_WAIT
    print(fsm.send("timeout"))        # CLOSED


if __name__ == "__main__":
    state_machine_demo()

5.7 无限序列与惰性求值

"""无限序列与惰性求值。Python 3.12+。"""
from __future__ import annotations

import itertools
from collections.abc import Generator
from typing import TypeVar

T = TypeVar("T")


def fibonacci() -> Generator[int, None, None]:
    """无限斐波那契数列。"""
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b


def primes() -> Generator[int, None, None]:
    """无限素数序列（埃拉托色尼筛法变体）。"""
    seen: list[int] = []
    candidate = 2
    while True:
        if all(candidate % p != 0 for p in seen):
            seen.append(candidate)
            yield candidate
        candidate += 1


def naturals(start: int = 0) -> Generator[int, None, None]:
    """自然数序列。"""
    n = start
    while True:
        yield n
        n += 1


def cycle(items: list[T]) -> Generator[T, None, None]:
    """循环序列。"""
    while True:
        yield from items


def infinite_demo() -> None:
    """无限序列演示。"""
    # 取前 10 个斐波那契数
    fib_10 = list(itertools.islice(fibonacci(), 10))
    print(f"Fibonacci(10): {fib_10}")
    # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

    # 取前 10 个素数
    prime_10 = list(itertools.islice(primes(), 10))
    print(f"Primes(10): {prime_10}")
    # [2, 3, 5, 7, 11, 13, 17, 19, 23, 29]

    # 无限循环 + 索引
    for i, value in zip(range(5), cycle(["A", "B", "C"])):
        print(f"{i}: {value}")

    # 流式处理：前 1000 个自然数的平方和
    squares_sum = sum(x * x for x in naturals())
    first_1000_sum = sum(itertools.islice((x * x for x in naturals()), 1000))
    print(f"前 1000 个自然数平方和: {first_1000_sum}")


if __name__ == "__main__":
    infinite_demo()

5.8 生成器与 itertools 组合

"""生成器与 itertools 组合。Python 3.12+。"""
from __future__ import annotations

import itertools
from collections.abc import Generator, Iterable


def batched(iterable: Iterable, size: int) -> Generator[list, None, None]:
    """分批处理可迭代对象。

    Args:
        iterable: 输入序列。
        size: 每批大小。

    Yields:
        长度为 size 的列表（最后一批可能更小）。
    """
    iterator = iter(iterable)
    while True:
        batch = list(itertools.islice(iterator, size))
        if not batch:
            break
        yield batch


def sliding_window(
    iterable: Iterable, window_size: int
) -> Generator[tuple, None, None]:
    """滑动窗口。

    Args:
        iterable: 输入序列。
        window_size: 窗口大小。

    Yields:
        长度为 window_size 的元组。
    """
    iterator = iter(iterable)
    window = list(itertools.islice(iterator, window_size))
    if len(window) == window_size:
        yield tuple(window)
    for item in iterator:
        window.pop(0)
        window.append(item)
        yield tuple(window)


def grouped_by(
    iterable: Iterable, key_func
) -> Generator[tuple, None, None]:
    """按键分组（要求输入已排序）。"""
    for key, group in itertools.groupby(iterable, key_func):
        yield key, list(group)


def chunked_ranges(
    start: int, end: int, chunk_size: int
) -> Generator[range, None, None]:
    """分块范围。"""
    for chunk_start in range(start, end, chunk_size):
        chunk_end = min(chunk_start + chunk_size, end)
        yield range(chunk_start, chunk_end)


def itertools_demo() -> None:
    """itertools 组合演示。"""
    # 分批
    for batch in batched(range(10), 3):
        print(f"批次: {batch}")

    # 滑动窗口
    for window in sliding_window([1, 2, 3, 4, 5], 3):
        print(f"窗口: {window}")

    # 分组
    data = [("A", 1), ("A", 2), ("B", 3), ("B", 4), ("C", 5)]
    for key, group in grouped_by(data, lambda x: x[0]):
        print(f"{key}: {group}")

    # 分块范围
    for chunk in chunked_ranges(0, 100, 30):
        print(f"块: {list(chunk)}")


if __name__ == "__main__":
    itertools_demo()

5.9 异步生成器

"""异步生成器：异步流式数据。Python 3.12+。"""
from __future__ import annotations

import asyncio
from collections.abc import AsyncGenerator


async def async_range(n: int) -> AsyncGenerator[int, None]:
    """异步范围生成器。"""
    for i in range(n):
        await asyncio.sleep(0.01)  # 模拟异步操作
        yield i


async def async_read_lines(filepath: str) -> AsyncGenerator[str, None]:
    """异步逐行读取文件。"""
    # 实际项目使用 aiofiles
    try:
        with open(filepath, encoding="utf-8") as f:
            for line in f:
                await asyncio.sleep(0)  # 让出控制权
                yield line.rstrip()
    except FileNotFoundError:
        return


async def async_fetch_pages(
    url: str, page_size: int = 100
) -> AsyncGenerator[dict, None]:
    """异步分页获取 API 数据。"""
    page = 1
    while True:
        # 模拟异步 HTTP 请求
        await asyncio.sleep(0.05)
        data = {
            "page": page,
            "items": [{"id": page * page_size + i} for i in range(page_size)],
            "has_next": page < 5,
        }
        yield data
        if not data["has_next"]:
            break
        page += 1


async def async_pipeline() -> AsyncGenerator[dict, None]:
    """异步管道：分页 -> 过滤 -> 转换。"""
    async for page in async_fetch_pages("https://api.example.com"):
        for item in page["items"]:
            if item["id"] % 2 == 0:  # 仅保留偶数 ID
                yield {"processed_id": item["id"] * 10}


async def async_generator_demo() -> None:
    """异步生成器演示。"""
    # 异步迭代
    async for value in async_range(5):
        print(f"async value: {value}")

    # 异步管道
    results = [item async for item in async_pipeline()]
    print(f"处理结果: {len(results)} 条")

    # 异步生成器理解
    agen = async_range(3)
    print(f"类型: {type(agen).__name__}")  # async_generator

    # 手动迭代
    print(await agen.__anext__())
    print(await agen.__anext__())
    print(await agen.__anext__())

    # 关闭
    await agen.aclose()


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

5.10 实战：流式日志分析

"""实战：流式日志分析。Python 3.12+。"""
from __future__ import annotations

import re
from collections import Counter, defaultdict
from collections.abc import Generator, Iterable
from dataclasses import dataclass
from datetime import datetime


@dataclass(frozen=True)
class LogEntry:
    """日志条目。"""
    timestamp: datetime
    level: str
    message: str
    raw: str


LOG_PATTERN = re.compile(
    r"(?P<ts>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})"
    r"\s+(?P<level>INFO|WARN|ERROR|DEBUG)"
    r"\s+(?P<message>.*)"
)


def parse_log_lines(lines: Iterable[str]) -> Generator[LogEntry, None, None]:
    """解析日志行为 LogEntry 对象。"""
    for line in lines:
        match = LOG_PATTERN.match(line.strip())
        if match:
            yield LogEntry(
                timestamp=datetime.strptime(match["ts"], "%Y-%m-%d %H:%M:%S"),
                level=match["level"],
                message=match["message"],
                raw=line,
            )


def filter_by_level(
    entries: Iterable[LogEntry], levels: set[str]
) -> Generator[LogEntry, None, None]:
    """按级别过滤。"""
    for entry in entries:
        if entry.level in levels:
            yield entry


def window_by_time(
    entries: Iterable[LogEntry], window_seconds: int
) -> Generator[list[LogEntry], None, None]:
    """按时间窗口分组。"""
    window: list[LogEntry] = []
    window_start: datetime | None = None
    for entry in entries:
        if window_start is None:
            window_start = entry.timestamp
        elif (entry.timestamp - window_start).total_seconds() > window_seconds:
            yield window
            window = []
            window_start = entry.timestamp
        window.append(entry)
    if window:
        yield window


def count_levels(
    windows: Iterable[list[LogEntry]],
) -> Generator[dict, None, None]:
    """统计每个窗口的级别分布。"""
    for window in windows:
        counter = Counter(entry.level for entry in window)
        yield {
            "start": window[0].timestamp if window else None,
            "total": len(window),
            "levels": dict(counter),
        }


def analyze_log_stream(log_lines: Iterable[str]) -> Generator[dict, None, None]:
    """完整日志分析管道。"""
    entries = parse_log_lines(log_lines)
    filtered = filter_by_level(entries, {"ERROR", "WARN"})
    windows = window_by_time(filtered, window_seconds=60)
    yield from count_levels(windows)


def log_analysis_demo() -> None:
    """日志分析演示。"""
    sample_logs = """
2026-01-15 10:00:00 INFO Server started
2026-01-15 10:00:05 ERROR Connection refused
2026-01-15 10:00:10 WARN Slow query detected
2026-01-15 10:01:00 ERROR Disk full
2026-01-15 10:01:30 WARN Memory usage high
2026-01-15 10:02:00 ERROR Database timeout
2026-01-15 10:02:30 WARN CPU usage high
""".strip().split("\n")

    for stats in analyze_log_stream(sample_logs):
        print(f"窗口开始: {stats['start']}, 总数: {stats['total']}, 分布: {stats['levels']}")


if __name__ == "__main__":
    log_analysis_demo()

6. 对比分析

6.1 生成器 vs 迭代器

6.2 生成器协程 vs asyncio.Task

6.3 Python 生成器 vs JavaScript 生成器

6.4 Python 协程 vs Lua 协程

6.5 Python 生成器 vs Go goroutine

7. 常见陷阱与最佳实践

7.1 常见陷阱

陷阱 1：忘记启动生成器

"""错误：直接 send 非 None 值。"""
def gen():
    while True:
        value = yield
        print(f"收到: {value}")

g = gen()
g.send("hello")  # TypeError: can't send non-None value to a just-started generator

# 正确做法
g = gen()
next(g)         # 或 g.send(None)，启动生成器
g.send("hello")  # OK

陷阱 2：生成器不可重入

"""错误：迭代完毕后再次迭代。"""
def gen():
    yield 1
    yield 2

g = gen()
print(list(g))  # [1, 2]
print(list(g))  # [] - 空列表，因为生成器已耗尽

# 正确做法：每次重新创建
def make_gen():
    yield 1
    yield 2

print(list(make_gen()))  # [1, 2]
print(list(make_gen()))  # [1, 2]

陷阱 3：在生成器中捕获 GeneratorExit 后 yield

"""错误：捕获 GeneratorExit 后继续 yield。"""
def bad_gen():
    try:
        yield 1
    except GeneratorExit:
        yield 2  # RuntimeError: generator ignored GeneratorExit

g = bad_gen()
next(g)
g.close()  # 抛出 RuntimeError

# 正确做法：GeneratorExit 必须重新抛出
def good_gen():
    try:
        yield 1
    except GeneratorExit:
        print("清理资源")
        raise  # 必须重新抛出

陷阱 4：生成器中的资源泄漏

"""错误：未关闭的生成器导致资源泄漏。"""
def read_file(path):
    f = open(path)
    try:
        for line in f:
            yield line
    finally:
        f.close()

g = read_file("data.txt")
next(g)  # 读取第一行
# 如果不调用 close()，文件句柄不会立即关闭

# 正确做法 1：显式 close
g = read_file("data.txt")
try:
    print(next(g))
finally:
    g.close()

# 正确做法 2：使用 with 语句（Python 3.7+ contextlib.closing 或直接 with）
from contextlib import closing
with closing(read_file("data.txt")) as g:
    for line in g:
        print(line)
        break  # 提前退出也会触发 close

陷阱 5：递归生成器的栈深度

"""错误：深度递归导致栈溢出。"""
def traverse_deep(tree):
    """深度优先遍历。"""
    yield tree.value
    for child in tree.children:
        yield from traverse_deep(child)  # 递归深度受栈限制

# 深度 1000+ 时会 RecursionError

# 正确做法：显式栈模拟
def traverse_iter(tree):
    """迭代式深度优先遍历。"""
    stack = [tree]
    while stack:
        node = stack.pop()
        yield node.value
        stack.extend(reversed(node.children))

陷阱 6：生成器协程与原生协程混用

"""错误：在 async 函数中使用 yield from。"""
async def bad_async():
    yield from some_generator()  # SyntaxError: 'yield from' inside async function

# 正确做法：使用 async for 或 await
async def good_async():
    async for item in async_generator():
        process(item)

async def good_async_await():
    result = await some_coroutine()

陷阱 7：生成器中的闭包变量

"""错误：闭包捕获变量在延迟求值时已变。"""
funcs = [(lambda: i) for i in range(3)]
print([f() for f in funcs])  # [2, 2, 2] - 全是 2

# 生成器版本同样有此问题
def gen_funcs():
    for i in range(3):
        yield lambda: i

funcs = list(gen_funcs())
print([f() for f in funcs])  # [2, 2, 2]

# 正确做法：使用默认参数绑定
funcs = [(lambda i=i: i) for i in range(3)]
print([f() for f in funcs])  # [0, 1, 2]

7.2 最佳实践

实践 1：使用类型注解

from collections.abc import Generator

# Generator[YieldType, SendType, ReturnType]
def counter() -> Generator[int, str, bool]:
    count = 0
    while count < 10:
        msg = yield count
        if msg == "stop":
            return False
        count += 1
    return True

实践 2：始终在 finally 中清理资源

def safe_resource():
    resource = acquire_resource()
    try:
        while True:
            data = yield resource.process()
            resource.feed(data)
    finally:
        # 无论正常结束、异常还是 close，都会执行
        resource.release()

实践 3：使用 yield from 替代手动迭代

# 不推荐
def chain_manual(*iterables):
    for it in iterables:
        for item in it:
            yield item

# 推荐
def chain_elegant(*iterables):
    for it in iterables:
        yield from it

实践 4：异步生成器使用 async with

async def safe_async_gen():
    async with aiofiles.open("data.txt") as f:
        async for line in f:
            yield line

实践 5：限制生成器递归深度

import sys

def safe_traverse(tree, max_depth=None):
    if max_depth is None:
        max_depth = sys.getrecursionlimit() // 2
    yield from _traverse(tree, 0, max_depth)

def _traverse(node, depth, max_depth):
    if depth > max_depth:
        raise RecursionError(f"深度超过 {max_depth}")
    yield node
    for child in node.children:
        yield from _traverse(child, depth + 1, max_depth)

实践 6：使用 itertools 工具函数

from itertools import islice, takewhile, dropwhile, groupby

# 取前 N 个
list(islice(infinite_gen(), 10))

# 满足条件时停止
list(takewhile(lambda x: x < 5, naturals()))

# 跳过满足条件的
list(dropwhile(lambda x: x < 5, range(10)))

# 分组
for key, group in groupby(sorted(data, key=key_func), key=key_func):
    process(key, list(group))

实践 7：生成器装饰器

from functools import wraps
from collections.abc import Generator

def safe_close(func):
    """确保生成器被正确关闭的装饰器。"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        gen = func(*args, **kwargs)
        try:
            yield from gen
        finally:
            if isinstance(gen, Generator):
                gen.close()
    return wrapper

实践 8：使用 dataclass + 生成器构建管道

from dataclasses import dataclass

@dataclass
class PipelineStage:
    name: str
    func: callable

    def apply(self, data):
        return self.func(data)

class PipelineBuilder:
    def __init__(self):
        self.stages: list[PipelineStage] = []

    def add(self, name: str, func: callable) -> "PipelineBuilder":
        self.stages.append(PipelineStage(name, func))
        return self

    def build(self, source):
        data = source
        for stage in self.stages:
            data = stage.apply(data)
        yield from data

8. 工程实践

8.1 项目结构

project/
├── src/
│   ├── generators/
│   │   ├── __init__.py
│   │   ├── core.py          # 核心生成器
│   │   ├── pipelines.py     # 管道组合
│   │   ├── async_gens.py    # 异步生成器
│   │   └── decorators.py    # 生成器装饰器
│   ├── utils/
│   │   ├── itertools_ext.py # itertools 扩展
│   │   └── types.py         # 类型定义
│   └── api/
│       ├── routes.py
│       └── streaming.py     # 流式响应
├── tests/
│   ├── test_generators.py
│   ├── test_pipelines.py
│   └── test_async_gens.py
├── pyproject.toml
└── README.md

8.2 pyproject.toml 配置

[project]
name = "generator-utils"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "typing-extensions>=4.7.0",
]

[project.optional-dependencies]
async = [
    "aiofiles>=23.0",
    "httpx>=0.25.0",
]
dev = [
    "pytest>=7.4",
    "pytest-asyncio>=0.21",
    "mypy>=1.5",
    "ruff>=0.1.0",
]

[tool.mypy]
python_version = "3.12"
strict = true
warn_return_any = true

[tool.ruff]
target-version = "py312"
line-length = 100

[tool.pytest.ini_options]
asyncio_mode = "auto"

8.3 流式 HTTP 响应

"""FastAPI 流式响应。Python 3.12+。"""
from __future__ import annotations

import json
from collections.abc import AsyncGenerator

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


async def generate_large_dataset() -> AsyncGenerator[str, None]:
    """生成大型数据集（SSE 格式）。"""
    for i in range(10000):
        data = {"id": i, "value": i * i}
        yield f"data: {json.dumps(data)}\n\n"
        # 模拟异步处理
        import asyncio
        await asyncio.sleep(0.001)


@app.get("/stream")
async def stream_data() -> StreamingResponse:
    """流式返回数据。"""
    return StreamingResponse(
        generate_large_dataset(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@app.get("/csv")
async def stream_csv() -> StreamingResponse:
    """流式返回 CSV。"""
    async def csv_generator() -> AsyncGenerator[str, None]:
        yield "id,value\n"
        for i in range(1000):
            yield f"{i},{i * i}\n"

    return StreamingResponse(
        csv_generator(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=data.csv"},
    )

8.4 数据库游标流式处理

"""数据库游标流式处理。Python 3.12+。"""
from __future__ import annotations

from collections.abc import Generator
from contextlib import contextmanager

import psycopg2
from psycopg2.extras import DictCursor


@contextmanager
def stream_query(
    dsn: str, query: str, params: tuple = ()
) -> Generator[dict, None, None]:
    """流式查询数据库。

    Args:
        dsn: 数据库连接字符串。
        query: SQL 查询。
        params: 查询参数。

    Yields:
        每行记录作为字典。
    """
    conn = psycopg2.connect(dsn, cursor_factory=DictCursor)
    cursor = conn.cursor(name="stream_cursor")  # 服务端游标
    cursor.itersize = 1000  # 每次从服务器获取 1000 行

    try:
        cursor.execute(query, params)
        for row in cursor:
            yield dict(row)
    finally:
        cursor.close()
        conn.close()


# 使用示例
def process_large_table():
    """处理大表数据。"""
    dsn = "postgresql://user:pass@localhost/db"
    with stream_query(dsn, "SELECT * FROM events WHERE created_at > %s", ("2026-01-01",)) as stream:
        for row in stream:
            process_event(row)

8.5 生成器性能优化

"""生成器性能优化技巧。Python 3.12+。"""
from __future__ import annotations

import time
from collections.abc import Generator


def benchmark_generator_vs_list():
    """生成器 vs 列表性能对比。"""
    n = 10_000_000

    # 列表：占用大量内存
    start = time.perf_counter()
    squares_list = [x * x for x in range(n)]
    list_time = time.perf_counter() - start
    list_size = squares_list.__sizeof__() / 1024 / 1024
    print(f"列表: {list_time:.3f}s, {list_size:.1f} MB")

    # 生成器：内存占用小
    start = time.perf_counter()
    squares_gen = (x * x for x in range(n))
    # 仅在迭代时计算
    total = sum(squares_gen)
    gen_time = time.perf_counter() - start
    gen_size = squares_gen.__sizeof__() / 1024
    print(f"生成器: {gen_time:.3f}s, {gen_size:.1f} KB")


def optimized_chain():
    """优化的链式生成器。"""
    # 使用 yield from 比手动迭代快约 20%
    def chain_yield_from(*iterables):
        for it in iterables:
            yield from it

    def chain_manual(*iterables):
        for it in iterables:
            for item in it:
                yield item

    # 性能对比
    data = [list(range(1000))] * 100
    list(chain_yield_from(*data))  # 更快
    list(chain_manual(*data))       # 较慢


if __name__ == "__main__":
    benchmark_generator_vs_list()

8.6 测试生成器

"""生成器测试。Python 3.12+。"""
from __future__ import annotations

import pytest
from collections.abc import Generator


def fib_gen(n: int) -> Generator[int, None, None]:
    a, b = 0, 1
    for _ in range(n):
        yield a
        a, b = b, a + b


class TestFibGenerator:
    """生成器测试类。"""

    def test_basic_values(self):
        """测试基本值。"""
        result = list(fib_gen(10))
        assert result == [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

    def test_zero(self):
        """测试空序列。"""
        assert list(fib_gen(0)) == []

    def test_is_generator(self):
        """测试是否为生成器。"""
        gen = fib_gen(5)
        assert hasattr(gen, "__next__")
        assert hasattr(gen, "__iter__")

    def test_exhausted_raises_stopiteration(self):
        """测试耗尽后抛出 StopIteration。"""
        gen = fib_gen(2)
        next(gen)
        next(gen)
        with pytest.raises(StopIteration):
            next(gen)

    def test_send_protocol(self):
        """测试 send 协议。"""
        def accumulator():
            total = 0
            while True:
                value = yield total
                if value is None:
                    break
                total += value

        gen = accumulator()
        next(gen)  # 启动
        assert gen.send(10) == 10
        assert gen.send(5) == 15

    @pytest.mark.parametrize("n,expected", [
        (1, [0]),
        (5, [0, 1, 1, 2, 3]),
        (10, [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]),
    ])
    def test_parametrized(self, n, expected):
        """参数化测试。"""
        assert list(fib_gen(n)) == expected


# 异步生成器测试
import asyncio

async def async_range(n):
    for i in range(n):
        await asyncio.sleep(0)
        yield i


@pytest.mark.asyncio
async def test_async_generator():
    """异步生成器测试。"""
    result = [item async for item in async_range(5)]
    assert result == [0, 1, 2, 3, 4]

8.7 调试生成器

"""生成器调试技巧。Python 3.12+。"""
from __future__ import annotations

import inspect
import sys
from collections.abc import Generator


def inspect_generator(gen: Generator) -> dict:
    """检查生成器状态。"""
    state = "unknown"
    if gen.gi_running:
        state = "running"
    elif gen.gi_frame is None:
        state = "closed"
    elif gen.gi_frame.f_lasti == -1:
        state = "created"
    else:
        state = "suspended"

    return {
        "state": state,
        "running": gen.gi_running,
        "frame": gen.gi_frame,
        "code": gen.gi_code,
        "name": gen.gi_name,
        "qualname": gen.gi_qualname,
        "line": (
            inspect.getframeinfo(gen.gi_frame).code_context
            if gen.gi_frame
            else None
        ),
    }


def debug_demo():
    """调试演示。"""
    def sample():
        x = 1
        yield x
        y = 2
        yield y
        return "done"

    gen = sample()
    print(f"初始状态: {inspect_generator(gen)['state']}")  # created

    next(gen)
    print(f"第一次 yield 后: {inspect_generator(gen)['state']}")  # suspended

    next(gen)
    print(f"第二次 yield 后: {inspect_generator(gen)['state']}")  # suspended

    try:
        next(gen)
    except StopIteration as e:
        print(f"返回值: {e.value}")  # done

    print(f"结束状态: {inspect_generator(gen)['state']}")  # closed


if __name__ == "__main__":
    debug_demo()

9. 案例研究

9.1 案例：Python 标准库中的应用

os.walk - 目录遍历生成器：

# 简化实现
import os
from collections.abc import Generator

def walk(top: str) -> Generator[tuple, None, None]:
    """递归遍历目录。"""
    dirs, files = [], []
    for entry in os.scandir(top):
        if entry.is_dir():
            dirs.append(entry.name)
        else:
            files.append(entry.name)
    yield top, dirs, files
    for dir_name in dirs:
        yield from walk(os.path.join(top, dir_name))


# 实际使用
for root, dirs, files in walk("."):
    for file in files:
        print(os.path.join(root, file))

csv.reader - 流式 CSV 解析：

import csv

def read_large_csv(filepath):
    """流式读取大 CSV。"""
    with open(filepath, newline="", encoding="utf-8") as f:
        reader = csv.reader(f)
        headers = next(reader)
        for row in reader:
            yield dict(zip(headers, row))

json.JSONDecoder - 流式 JSON：

import json

def stream_json_array(filepath):
    """流式读取 JSON 数组。"""
    decoder = json.JSONDecoder()
    with open(filepath) as f:
        # 假设是数组
        content = f.read(1)  # [
        while True:
            char = f.read(1)
            if not char:
                break
            content += char
            if content.rstrip().endswith("}"):
                try:
                    yield decoder.decode(content)
                    content = ""
                except json.JSONDecodeError:
                    continue

9.2 案例：Django 流式响应

"""Django StreamingHttpResponse。Python 3.12+。"""
from django.http import StreamingHttpResponse
import csv
import io

def stream_csv_view(request):
    """流式返回 CSV。"""
    def generate():
        buffer = io.StringIO()
        writer = csv.writer(buffer)
        writer.writerow(["id", "name", "email"])
        yield buffer.getvalue()
        buffer.seek(0)
        buffer.truncate(0)

        for user in User.objects.iterator():  # 使用 iterator 避免全部加载
            writer.writerow([user.id, user.name, user.email])
            yield buffer.getvalue()
            buffer.seek(0)
            buffer.truncate(0)

    response = StreamingHttpResponse(generate(), content_type="text/csv")
    response["Content-Disposition"] = 'attachment; filename="users.csv"'
    return response

9.3 案例：Flask 流式响应

"""Flask 流式响应。Python 3.12+。"""
from flask import Flask, Response, stream_with_context
import time

app = Flask(__name__)


def generate_log_stream():
    """生成实时日志流。"""
    while True:
        log_line = fetch_log_line()
        if log_line is None:
            break
        yield f"data: {log_line}\n\n"
        time.sleep(0.1)


@app.route("/logs")
def logs():
    """流式日志端点。"""
    return Response(
        stream_with_context(generate_log_stream()),
        content_type="text/event-stream",
    )

9.4 案例：SQLAlchemy 流式查询

"""SQLAlchemy 流式查询。Python 3.12+。"""
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session
from models import User

engine = create_engine("postgresql://user:pass@localhost/db")


def stream_users(batch_size: int = 1000):
    """流式查询用户表。

    使用服务端游标，避免一次性加载所有数据。
    """
    with Session(engine) as session:
        stmt = select(User).execution_options(
            stream_results=True,
            yield_per=batch_size,
        )
        for partition in session.execute(stmt).partitions():
            for row in partition:
                yield row[0]


# 使用
for user in stream_users():
    process(user)

9.5 案例：实时数据处理管道

"""实时数据处理管道：模拟股票行情流。Python 3.12+。"""
from __future__ import annotations

import asyncio
import json
import random
from collections import defaultdict, deque
from collections.abc import AsyncGenerator
from dataclasses import dataclass
from datetime import datetime


@dataclass
class Tick:
    symbol: str
    price: float
    timestamp: datetime


async def tick_stream(symbols: list[str]) -> AsyncGenerator[Tick, None]:
    """模拟股票行情流。"""
    prices = {s: random.uniform(100, 200) for s in symbols}
    while True:
        symbol = random.choice(symbols)
        prices[symbol] *= random.uniform(0.99, 1.01)
        yield Tick(
            symbol=symbol,
            price=prices[symbol],
            timestamp=datetime.now(),
        )
        await asyncio.sleep(random.uniform(0.01, 0.1))


def moving_average(
    ticks: AsyncGenerator[Tick, None], window: int = 5
) -> AsyncGenerator[dict, None]:
    """计算移动平均。"""
    buffer: dict[str, deque] = defaultdict(lambda: deque(maxlen=window))

    async def inner():
        async for tick in ticks:
            buffer[tick.symbol].append(tick.price)
            if len(buffer[tick.symbol]) == window:
                avg = sum(buffer[tick.symbol]) / window
                yield {
                    "symbol": tick.symbol,
                    "price": tick.price,
                    "avg": avg,
                    "timestamp": tick.timestamp.isoformat(),
                }

    return inner()


async def threshold_alert(
    ticks: AsyncGenerator[dict, None], threshold: float = 0.02
) -> AsyncGenerator[dict, None]:
    """价格异常告警。"""
    prev: dict[str, float] = {}
    async for tick in ticks:
        symbol = tick["symbol"]
        price = tick["price"]
        if symbol in prev:
            change = (price - prev[symbol]) / prev[symbol]
            if abs(change) > threshold:
                yield {
                    **tick,
                    "alert": "price_spike",
                    "change_pct": change * 100,
                }
        prev[symbol] = price


async def main():
    """主流程：组合管道。"""
    symbols = ["AAPL", "GOOG", "MSFT", "TSLA"]
    stream = tick_stream(symbols)

    # 管道：原始 tick -> 移动平均 -> 告警
    avg_stream = moving_average(stream, window=5)
    alert_stream = threshold_alert(avg_stream, threshold=0.015)

    async for alert in alert_stream:
        print(json.dumps(alert, indent=2))


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

10. 练习题

10.1 选择题

1. 下列关于 Python 生成器的描述，正确的是？

A. 生成器函数调用后立即执行函数体 B. 生成器对象可以多次迭代 C. yield 表达式的值是 send() 传入的值 D. 生成器在 return 后仍可继续 yield

答案：C

解析：yield expr 是一个表达式，其值由 send(value) 传入；若通过 next() 调用，则 yield 表达式值为 None。A 错误（调用生成器函数返回生成器对象，不执行函数体）；B 错误（生成器一次性，迭代完毕后再次迭代为空）；D 错误（return 后生成器结束，抛出 StopIteration）。

2. 执行以下代码的输出是？

def gen():
    x = yield 1
    y = yield x + 10
    return x + y

g = gen()
print(next(g))
print(g.send(5))
try:
    g.send(3)
except StopIteration as e:
    print(e.value)

A. 1, 15, 8 B. 1, 15, StopIteration: 8 C. 1, 15, 8 D. 1, 10, 8

答案：C

解析：

• next(g)：启动生成器，执行到 x = yield 1，返回 1；
• g.send(5)：x = 5，执行到 y = yield x + 10 = 15，返回 15；
• g.send(3)：y = 3，return x + y = 8，抛出 StopIteration(8)，e.value = 8。

10.2 填空题

1. 以下代码输出是？

def gen():
    yield from [1, 2, 3]
    yield from "abc"
    return "done"

g = gen()
result = list(g)
print(result)

答案：[1, 2, 3, 'a', 'b', 'c']

解析：yield from 展开迭代器，list() 收集所有 yield 的值，return 的值通过 StopIteration.value 获取，不会被 list() 收集。