第一讲 迭代器协议与生成器范式

大家好，这里是程序员晚枫，正在all in AI编程实战，全网同名。

（3 小时直播脚本 + 课后练习）

──────────────────
0. 开场（5 min）
• 一句话定位：今天之后，你将能“用 Python 的方式”优雅地遍历大文件、流式处理数据，并写出可复用的惰性计算库。
• 课程节奏：45 min 概念 → 30 min 代码实操 → 15 min 调试技巧 → 60 min 综合案例 → 20 min 总结 & 作业 → 10 min Q&A。

──────────────────

温故：for-loop 背后到底发生了什么？（10 min）
代码对比：

lst = [1, 2, 3]
for x in lst:        # 语法糖
    print(x)

it = iter(lst)       # 等价展开
while True:
    try:
        x = next(it)
        print(x)
    except StopIteration:
        break

提问：为什么文件对象 open() 可以 for line in f:？
→ 引出“迭代器协议”。

──────────────────
2. 迭代器协议解剖（15 min）
2.1 协议定义

可迭代对象：__iter__ 返回迭代器
迭代器对象：__iter__ 返回自身 + __next__ 抛 StopIteration

2.2 手写一个 Fibonacci 迭代器（现场编码）

class Fib:
    def __init__(self, n):
        self.n, self.a, self.b, self.i = n, 0, 1, 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.i >= self.n:
            raise StopIteration
        self.a, self.b = self.b, self.a + self.b
        self.i += 1
        return self.a

运行验证：

1	list(Fib(10))

──────────────────
3. 生成器函数：更轻量的迭代器（15 min）
3.1 语法糖

def fib(n):
    a, b = 0, 1
    for _ in range(n):
        a, b = b, a + b
        yield a

对比：代码量 ↓ 80%，状态机由 Python 代管。

3.2 双向通信：send/throw/close
现场演示：

def gen():
    received = yield 'ready'
    yield f'got: {received}'

g = gen()
print(next(g))     # ready
print(g.send(42))  # got: 42

3.3 生成器表达式

1	squares = (x * x for x in range(10**8)) # 惰性

──────────────────
4. 代码实操：重构“逐行读大文件”脚本（30 min）
需求：10 GB 日志文件，提取包含 ERROR 的行并写入新文件。
4.1 原始写法（内存爆炸）：

1 2	with open('big.log') as f: errors = [line for line in f if 'ERROR' in line]

4.2 惰性迭代器版：

def error_lines(path):
    with open(path) as f:
        for line in f:
            if 'ERROR' in line:
                yield line

with open('errors.log', 'w') as out:
    for line in error_lines('big.log'):
        out.write(line)

4.3 进阶：用 itertools.islice 截断前 100 条

1 2	from itertools import islice first_100 = islice(error_lines('big.log'), 100)

──────────────────
5. 调试技巧：生成器断点 & 可视化（15 min）

PyCharm 断点：在 yield 行可单步查看局部变量。
reprlib.repr() 打印大对象不卡死。

sys.getsizeof 对比列表 vs 生成器内存占用：

1
2
3

import sys
sys.getsizeof([x for x in range(1000000)])   # 约 8 MB
sys.getsizeof((x for x in range(1000000)))   # 约 120 B

──────────────────
6. 综合案例：实时日志监控器（60 min）
需求：对持续写入的日志文件，实时输出 ERROR 行并统计数量。
技术栈：生成器 + tail -f 模拟 + asyncio（预告下一章）。
6.1 实现 tail 生成器

import time
def tail(path):
    with open(path) as f:
        f.seek(0, 2)  # 跳到末尾
        while True:
            line = f.readline()
            if not line:
                time.sleep(0.1)
                continue
            yield line

6.2 聚合统计

from collections import Counter
counts = Counter()
for line in tail('app.log'):
    if 'ERROR' in line:
        counts['error'] += 1
        print(line.strip(), counts['error'])