【一起学 Python】第 67 天:生成器的优势:惰性计算与内存节省

# 列表：立即计算所有值（ eager evaluation）numbers_list = [x ** 2 for x in range(1000000)]  # 立即生成 100 万个元素# 生成器：惰性计算，按需生成（lazy evaluation）numbers_gen = (x ** 2 for x in range(1000000))   # 不会立即计算

import sys# 列表占用大量内存numbers_list = [i for i in range(1000000)]print(f"列表内存：{sys.getsizeof(numbers_list):,} 字节")  # 约 8,448,728 字节 (8MB+)# 生成器几乎不占内存numbers_gen = (i for i in range(1000000))print(f"生成器内存：{sys.getsizeof(numbers_gen):,} 字节")  # 约 200 字节

# range 对象是惰性的r = range(1000000000)  # 10 亿个数print(f"range 内存：{sys.getsizeof(r)} 字节")  # 只有 48 字节# 转为列表会占用大量内存（不要运行，会卡死）# l = list(r)  # 需要约 40GB 内存

def expensive_calculation(n):    print(f"计算 {n}")    return n ** 2# 列表：全部计算（即使只需要前 5 个）result_list = [expensive_calculation(i) for i in range(100)]first_five = result_list[:5]  # 但已经计算了 100 次# 生成器：按需计算def gen_expensive(n):    for i in range(n):        yield expensive_calculation(i)gen = gen_expensive(100)first_five = [next(gen) for _ in range(5)]  # 只计算 5 次

# ❌ 错误做法：一次性读取def process_file_bad(filepath):    with open(filepath) as f:        lines = f.readlines()  # 大文件会耗尽内存    for line in lines:        process(line)# ✅ 正确做法：使用生成器逐行处理def process_file_good(filepath):    with open(filepath, encoding="utf-8") as f:        for line in f:  # 文件对象本身是迭代器            yield line.strip()# 即使文件有 10GB，内存占用也很小for line in process_file_good("huge_log.txt"):    if "ERROR" in line:        print(line)

def read_data(filepath):    """第一步：读取数据"""    with open(filepath) as f:        for line in f:            yield line.strip()def parse_data(lines):    """第二步：解析数据"""    for line in lines:        parts = line.split(",")        yield {"id": parts[0], "value": float(parts[1])}def filter_data(records, threshold):    """第三步：过滤数据"""    for record in records:        if record["value"] > threshold:            yield recorddef transform_data(records):    """第四步：转换数据"""    for record in records:        yield {**record, "doubled": record["value"] * 2}# 构建管道 - 数据流式处理，内存占用极小pipeline = transform_data(    filter_data(        parse_data(            read_data("data.csv")        ),        threshold=100    ))# 遍历时才真正执行for record in pipeline:    print(record)

def batch_process(iterable, batch_size):    """将数据分批处理"""    batch = []    for item in iterable:        batch.append(item)        if len(batch) >= batch_size:            yield batch            batch = []    if batch:  # 处理剩余的        yield batch# 处理 100 万条数据，每批 1000 条data = range(1000000)for batch in batch_process(data, 1000):    # 每次只有 1000 条在内存中    process_batch(batch)

import randomimport timedef sensor_data():    """模拟传感器数据流"""    while True:        yield {            "timestamp": time.time(),            "temperature": random.uniform(20, 30),            "humidity": random.uniform(40, 60)        }def moving_average(data_stream, window_size):    """计算移动平均"""    window = []    for data in data_stream:        window.append(data["temperature"])        if len(window) > window_size:            window.pop(0)        yield sum(window) / len(window)# 实时处理传感器数据sensor = sensor_data()for avg in moving_average(sensor, 10):    print(f"平均温度：{avg:.2f}")    time.sleep(1)

# 错误：先转列表再求和，浪费内存# total = sum(list(x * 2 for x in range(10000000)))# 正确：直接用生成器求和total = sum(x * 2 for x in range(10000000))  # 内存占用极小

from typing import Generator, Iterabledef read_lines(filepath: str) -> Generator[str, None, None]:    with open(filepath) as f:        for line in f:            yield linedef process_data(data: Iterable[int]) -> Generator[int, None, None]:    for item in data:        yield item * 2

# ❌ 调试陷阱gen = (x for x in range(5))print(list(gen))  # [0, 1, 2, 3, 4] - 但生成器已耗尽# ✅ 使用 itertools.tee 创建副本（仅用于调试）from itertools import teegen = (x for x in range(5))gen1, gen2 = tee(gen, 2)print(list(gen1))  # 用于调试# gen2 仍可使用

import tracemalloctracemalloc.start()# 列表方式numbers_list = [x ** 2 for x in range(100000)]snapshot1 = tracemalloc.take_snapshot()# 生成器方式numbers_gen = (x ** 2 for x in range(100000))list(numbers_gen)  # 消费生成器snapshot2 = tracemalloc.take_snapshot()# 打印统计print("列表方式的内存使用:")for stat in snapshot1.statistics('lineno')[:3]:    print(stat)

gen = (x for x in range(3))print(list(gen))  # [0, 1, 2]print(list(gen))  # [] (已耗尽)# 正确：重新创建gen = (x for x in range(3))

def gen_length(gen):    return sum(1 for _ in gen)  # 会耗尽生成器# 如果需要长度和数据，先转列表gen = (x for x in range(100))data = list(gen)print(len(data), data)