Python · 生成器与迭代协议

场景 1 · 10GB 日志流式 ETL, 512MB 容器跑通

统计每个渠道的错误码分布。全量读必 OOM, 生成器管道全程只驻留一行:

def read_lines(path):
    with open(path, 'rb') as f:
        for line in f:                    # 文件对象本身就是惰性迭代器
            yield line

bad = (l for l in read_lines('app.log') if b'ERROR' in l)
parsed = (parse(l) for l in bad)     # 生成器链: 元素逐个流过
for (code, n) in Counter(p.channel for p in parsed).most_common(10):
    print(code, n)   # 峰值内存 = 一行 + 一个计数器

场景 2 · 数据库游标分页, 接口"无限滚动"

把分页查询封装成生成器, 调用方像遍历普通序列一样用, 内部自动翻页:

def iter_page(db, size=500):
    last_id = 0
    while True:
        rows = db.query("SELECT * FROM t WHERE id > %s ORDER BY id LIMIT %s",
                        (last_id, size))
        if not rows: return
        yield from rows                   # 委托透传
        last_id = rows[-1].id

for row in iter_page(db):    # 消费方无感: 找到目标即 break, 后面页不查 (早停)
    ...

对比 fetchall() 一次拉百万行: 生成器版本数据库压力与内存都恒定; 配合早停, 未见过的数据根本不会查询。

场景 3 · islice 实现生成器的"切片"与限流

from itertools import islice
head = islice(huge_stream(), 100)          # 惰性取前 100 个, 不会拉全量
sample = islice(stream, 0, None, 100)     # 每 100 个抽 1 个做采样监控

场景 4 · 海量数据流式去重

10 亿 URL 放不进 set: 分片哈希让"同一条必落同一片", 每片独立去重, 内存恒定:

def shard_urls(urls, n=64):
    files = [open(f"tmp/shard_{i}.txt", "w") for i in range(n)]
    try:
        for u in urls:                       # urls 是读源文件的生成器
            files[hash(u) % n].write(u + "\n")  # 同一 URL 恒定落同一片
    finally:
        for f in files: f.close()
# 之后逐片 set() 去重 — 单片集合大小 = 总量/64, 可控

场景 5 · 实时日志 tail 监控

自实现 tail -f: 生成器内轮询增量行, 消费方拿到的就是"持续到来的新行":

import time
def follow(path):
    with open(path, "rb") as f:
        f.seek(0, 2)                        # 跳到文件末尾 (断点续读改 seek(上次偏移))
        while True:
            line = f.readline()
            if not line:
                time.sleep(0.5); continue   # 没新内容, 稍后再看
            yield line
for line in follow("/var/log/app.log"):
    if b"ERROR" in line: alert(line)

场景 6 · 可测试的 ETL 管道工厂

每阶段都是"生成器进、生成器出"的纯函数, 单测喂 iter([...]) 三行数据即可全链路验证:

def parse(lines):      yield from (json.loads(l) for l in lines)
def keep(evt):         yield from (e for e in evt if e["ok"])
def enrich(evt):       yield from ({**e, "region": region(e["ip"])} for e in evt)

pipeline = lambda src: enrich(keep(parse(src)))   # 组合即代码
# 单测: assert list(pipeline(iter([line1, line2]))) == [...]

场景 7 · 消息流按批提交

逐条写库太慢、全量攒怕丢: 生成器把消息流切成固定批量, 吞吐与安全的平衡点:

from itertools import islice
def batches(msgs, size=500):
    while batch := list(islice(msgs, size)):   # 取满 500 或耗尽
        yield batch
for batch in batches(consumer):
    db.executemany(INSERT_SQL, batch)            # 500 行一次落库
    consumer.ack(batch)                          # 批内全部成功才 ack

场景 8 · 令牌桶限速器

限速逻辑封进生成器, 业务循环拿不到令牌就等在生成器里, 与业务代码解耦:

import time
def token_bucket(rate=100):                  # 100 QPS
    tokens, last = 0.0, time.monotonic()
    while True:
        now = time.monotonic()
        tokens = min(rate, tokens + (now - last) * rate)   # 按时间补充
        last = now
        if tokens >= 1:
            tokens -= 1; yield                     # 发放一枚令牌
        else:
            time.sleep((1 - tokens) / rate)       # 差多少睡多少

场景 9 · 千万元数据分批导入

游标生成器逐行读源表 + islice 分批写目标表, 源不爆、事务日志不膨胀:

def iter_rows(cur):
    while rows := cur.fetchmany(5000):   # 服务端游标, 常驻内存一行
        yield from rows
src = iter_rows(src_conn.cursor())
while chunk := list(islice(src, 5000)):
    dst.executemany(INSERT, chunk)              # 5000 行/事务, 可断点续跑
    mark_progress(len(chunk))                   # 进度可观测, 失败从断点重来

场景 10 · 惰性测试夹具

fixture 返回生成器, 样本生成开销按需支付, 参数化取之不尽:

import itertools, pytest

@pytest.fixture
def users():
    return (make_user(f"u{i}") for i in itertools.count())  # 无限样本流

def test_topk(users):
    top10 = list(islice(users, 10))       # 只造 10 个, 不是一万个
    assert rank(top10) == sorted(top10, key=score, reverse=True)