Python案例如何用生成器优化

wen python案例 2026-06-05 56

Python案例：如何用生成器优化内存与性能——实战指南

目录导读

生成器基础：从迭代器到惰性求值
处理超大日志文件避免内存溢出
斐波那契数列的无限生成
数据管道中的流式处理
生成器 vs 列表推导式：性能对比实测
常见问题与问答（FAQ）
最佳实践与注意事项

生成器基础：从迭代器到惰性求值

Python生成器是一种特殊的迭代器，通过yield关键字实现惰性求值——只在需要时才生成下一个元素，而不是一次性将所有数据载入内存，这种机制特别适合处理大数据流、无限序列或管道式数据处理。

Python案例如何用生成器优化

核心区别：普通函数用return返回完整结果，生成器函数用yield返回一个可迭代对象，每次调用next()才计算下一个值。

# 普通函数：一次性生成所有数据
def square_list(n):
    return [i**2 for i in range(n)]  # 占用O(n)内存
# 生成器函数：逐个生成
def square_gen(n):
    for i in range(n):
        yield i**2  # 仅保存当前状态

案例一：处理超大日志文件避免内存溢出

场景：有一个10GB的服务器日志文件，需要统计每个IP的访问次数，如果用readlines()读取,会导致内存不足。

优化方案：使用生成器逐行读取并处理。

def read_log_lines(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:  # f本身就是生成器
            yield line.strip()
def count_ip(lines):
    from collections import Counter
    counter = Counter()
    for line in lines:
        ip = line.split()[0]  # 假设IP在第一列
        counter[ip] += 1
    return counter
# 使用生成器链
log_lines = read_log_lines("giant_log.txt")
result = count_ip(log_lines)
print(result.most_common(10))

效果：内存占用从10GB降低到数十KB（仅存储计数器和当前行）,且性能损失极小。

案例二：斐波那契数列的无限生成

场景：需要生成斐波那契数列的前N项，但N是动态变化的,且不希望预先生成所有项占用内存。

生成器实现无限序列：

def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b
# 取前100个
fib = fibonacci()
first_100 = [next(fib) for _ in range(100)]
# 或者使用islice
from itertools import islice
first_50 = list(islice(fibonacci(), 50))

优势：无需定义N的上限，生成器始终保持O(1)内存,适用于流式数据场景。

案例三：数据管道中的流式处理

场景：从多个数据源读取记录，进行过滤、转换、聚合，形成一个数据处理流水线，用生成器实现每个阶段,避免中间结果占用内存。

# 阶段1：读取原始数据
def read_data_from_api():
    for page in range(1, 1000):
        yield fetch_page(page)  # 模拟分页获取
# 阶段2：过滤无效记录
def filter_valid(records):
    for rec in records:
        if rec.get("status") == "active":
            yield rec
# 阶段3：转换数据结构
def transform(records):
    for rec in records:
        yield {
            "id": rec["user_id"],
            "score": rec["score"] * 1.2,
            "time": rec["timestamp"]
        }
# 组合生成器管道
pipeline = transform(filter_valid(read_data_from_api()))
for item in pipeline:
    process(item)  # 每处理完一条就释放

关键点：每个生成器只处理当前元素，数据流像工厂流水线一样逐级传递,内存峰值等于单条记录大小。

生成器 vs 列表推导式：性能对比实测

测试代码（模拟处理1000万个整数）：

import time, sys
# 列表推导式：一次性生成
start = time.time()
big_list = [x**2 for x in range(10_000_000)]
print(f"列表占用内存: {sys.getsizeof(big_list) / 1024 / 1024:.2f} MB")
print(f"时间: {time.time() - start:.2f}s")
# 生成器表达式：惰性求值
start = time.time()
big_gen = (x**2 for x in range(10_000_000))
print(f"生成器占用内存: {sys.getsizeof(big_gen)} 字节")
print(f"时间: {time.time() - start:.2f}s")
# 实际遍历时才会计算
sum_val = sum(big_gen)  # 这步才真正迭代

结果：