4. 生成器表达式实战：处理大文件、流式数据处理、无限序列生成、管道模式设计
说实话，生成器表达式这东西，光看语法会觉得挺简单——不就是把列表推导式的方括号换成圆括号嘛。但真正把它用在实际项目里，你会发现它解决的是 Python 编程中一个非常核心的痛点：数据太大，内存装不下。
我个人习惯把生成器表达式看作「懒加载的迭代器」。它不一次性生成所有数据，而是等你来要的时候，才吐出一个。这种特性让它在大文件处理、流式数据、无限序列这些场景下，几乎是唯一优雅的解法。
4.1 大文件处理：逐行读取，绝不爆炸
先看一个最常见的场景——处理一个 10GB 的日志文件。如果你用 .readlines()，程序直接崩给你看。但用生成器表达式，你可以一行一行地处理，内存占用几乎为零。
# 传统方式：一次性读入内存，10GB 文件直接 OOM
with open('huge_log.txt', 'r') as f:
lines = f.readlines() # 别这么干
# 生成器方式：逐行处理，内存友好
with open('huge_log.txt', 'r') as f:
# 过滤掉空行和注释行
valid_lines = (line.strip() for line in f if line.strip() and not line.startswith('#'))
for line in valid_lines:
process(line) # 每次只处理一行
我在项目中遇到过一个问题：一个运维同事每天要分析 50GB 的 Nginx 访问日志，之前用列表推导式写了个脚本，跑一次内存就飙到 30GB，服务器直接报警。后来改成生成器表达式，内存稳定在 200MB 以内，跑完整个文件只需要多花 10% 的时间。
itertools.islice 做分块处理。比如每 10000 行做一次批量写入，既保持内存低占用，又能利用批量 I/O 的性能优势。
4.2 流式数据处理：数据来了就处理，不等
流式数据，说白了就是数据源源不断地来，你不知道什么时候结束。比如传感器数据、股票行情、用户点击流。这时候生成器表达式就特别适合做「实时过滤和转换」。
import random
import time
# 模拟一个传感器数据流
def sensor_stream():
while True:
yield random.uniform(20.0, 30.0) # 温度数据
time.sleep(0.1)
# 用生成器表达式做实时过滤：只保留异常温度
stream = sensor_stream()
alerts = (temp for temp in stream if temp > 28.0 or temp < 22.0)
# 实时监控
for alert in alerts:
print(f"⚠️ 异常温度: {alert:.2f}°C")
你想想看，如果这里用列表推导式，程序会永远等下去——因为数据流是无限的，列表推导式根本不会返回结果。而生成器表达式是懒加载的，来一个数据就判断一个，完全没问题。
map() 配合生成器，或者把计算逻辑拆成独立的生成器函数，保持管道清晰。
4.3 无限序列生成：数学上的「无穷」在代码里也能用
无限序列听起来很玄乎，但用生成器表达式实现起来其实很自然。比如生成所有自然数的平方：
# 无限自然数序列
def natural_numbers():
n = 0
while True:
yield n
n += 1
# 生成器表达式：所有自然数的平方（懒加载）
squares = (x * x for x in natural_numbers())
# 取前 10 个
from itertools import islice
first_10 = list(islice(squares, 10))
print(first_10) # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
嗯，这里要注意：itertools.islice 是处理无限序列的标配工具。没有它，你没法从无限序列里「切」出一段来用。
我曾经在做一个数学建模项目时，需要生成斐波那契数列的前 N 项。如果用递归，N=50 就卡死了；用列表推导式，你得先知道 N 是多少。但用生成器表达式配合 islice，代码既简洁又高效：
def fibonacci():
a, b = 0, 1
while True:
yield a
a, b = b, a + b
# 取前 20 个斐波那契数
fib_gen = fibonacci()
first_20 = list(islice(fib_gen, 20))
print(first_20)
4.4 管道模式设计：像搭积木一样组合数据处理
管道模式，说白了就是把多个生成器表达式串起来，每个环节只做一件事。数据像流水线一样，经过一个个工位，最终产出结果。
我个人特别喜欢这种设计——它让代码变得极其清晰，每个步骤都可以独立测试和复用。
# 原始数据：一堆数字
data = range(1000)
# 管道：过滤 → 转换 → 聚合
pipeline = (
x * 2 # 步骤3：翻倍
for x in (
x + 1 # 步骤2：加1
for x in data
if x % 2 == 0 # 步骤1：只保留偶数
)
)
# 执行管道
result = list(pipeline)
print(result[:10]) # [2, 6, 10, 14, 18, 22, 26, 30, 34, 38]
你看，这个管道做了三件事：过滤偶数 → 每个数加1 → 翻倍。每个步骤都是一个独立的生成器表达式，数据在它们之间「流」过去，不会产生中间列表。
下面这张图展示了管道模式的数据流向：
4.5 实战对比：列表推导式 vs 生成器表达式
为了让你更直观地感受两者的差异，我做了个对比测试：
| 场景 | 列表推导式 | 生成器表达式 | 我的建议 |
|---|---|---|---|
| 处理 1000 万条日志 | 内存爆炸（~800MB） | 内存稳定（~1MB） | 必须用生成器 |
| 无限数据流 | 无法使用 | 完美支持 | 只能用生成器 |
| 小数据集（<1000 条） | 速度快，代码直观 | 速度略慢，但可接受 | 两者皆可，我倾向列表 |
| 需要多次遍历 | 支持 | 不支持（一次消费） | 用列表或 list() 转存 |
| 管道式多步处理 | 产生大量中间列表 | 零中间开销 | 强烈推荐生成器 |
list() 转成列表，要么重新创建生成器。
4.6 综合案例：实时日志分析管道
最后，我把今天讲的内容整合成一个完整的实战案例——实时分析 Web 服务器日志，提取出所有 5xx 错误，统计每个错误码出现的次数：
import re
from collections import Counter
from itertools import islice
# 模拟日志流（实际场景中可能是 tail -f 或 Kafka 消费者）
def log_stream():
logs = [
"192.168.1.1 - - [10/Oct/2023:13:55:36] \"GET /api/users HTTP/1.1\" 200 2326",
"192.168.1.2 - - [10/Oct/2023:13:55:37] \"POST /api/order HTTP/1.1\" 500 12",
"192.168.1.3 - - [10/Oct/2023:13:55:38] \"GET /api/products HTTP/1.1\" 503 0",
"192.168.1.4 - - [10/Oct/2023:13:55:39] \"GET /api/health HTTP/1.1\" 200 2",
"192.168.1.5 - - [10/Oct/2023:13:55:40] \"POST /api/payment HTTP/1.1\" 502 45",
]
while True:
for log in logs:
yield log
# 管道构建
stream = log_stream()
# 步骤1：提取状态码
status_codes = (
int(re.search(r'\" (\d{3}) ', log).group(1))
for log in stream
)
# 步骤2：只保留 5xx 错误
errors = (code for code in status_codes if code >= 500)
# 步骤3：统计前 100 条日志中的错误分布
error_counts = Counter(islice(errors, 100))
print("5xx 错误分布：")
for code, count in error_counts.most_common():
print(f" {code}: {count} 次")
这个管道里，每一步都是独立的生成器表达式。你可以随时在中间插入新的处理环节，比如过滤特定 IP、提取请求路径、计算响应时间等。每个环节只关心自己的事，互不干扰。
itertools.tee() 把管道分叉，同时做多个维度的分析。但要注意 tee() 会占用额外内存，需要根据实际情况权衡。
好了，这一章的内容就到这里。生成器表达式和管道模式，说白了就是「用空间换时间」的反面——用一点点时间开销，换来巨大的内存节省。在数据量越来越大的今天，这个 trade-off 往往非常划算。