wordflow
返回博客

19. 生成器与迭代器：惰性计算的力量

19. 生成器与迭代器：惰性计算的力量

第十九篇：生成器与迭代器——惰性计算的力量

前言

**生成器（generator）和迭代器（iterator）**是 Python 中实现高效数据处理的核心机制。它们通过惰性求值节省内存，配合 yield 关键字可以优雅地处理海量数据流。本文全面讲解两者的原理、用法及常见应用场景。

生成器基础

什么是生成器？

生成器是一种特殊的函数，使用 yield 关键字逐次返回结果，每次调用时保存执行状态。

def simple_generator():
    yield 1
    yield 2
    yield 3

gen = simple_generator()
print(next(gen))  # 1
print(next(gen))  # 2
print(next(gen))  # 3
# print(next(gen))  # StopIteration — 没有更多元素

# ⚠️ 生成器只能遍历一次！
for item in gen:  # 空 — 已经耗尽
    print(item)

生成器表达式

# 列表生成式 vs 生成器表达式
squares_list = [x ** 2 for x in range(1_000_000)]  # 立即生成
squares_gen = (x ** 2 for x in range(1_000_000))    # 惰性延迟

import sys
print(sys.getsizeof(squares_list))  # ~8 MB
print(sys.getsizeof(squares_gen))   # ~112 bytes — 极小！

# 逐个消费
count = 0
for sq in squares_gen:
    if count >= 10: break
    print(sq, end=" ")
    count += 1
# 输出：0 1 4 9 16 25 36 49 64 81

带 yield 的生成器函数

基本用法

def countdown(n):
    while n > 0:
        yield n       # 暂停并返回值
        n -= 1

for i in countdown(5):
    print(i, end=" ")
# 输出：5 4 3 2 1

生产-消费者模型

def producer(data_stream):
    """生产者：从数据源产出"""
    for data in data_stream:
        cleaned = data.strip().lower()
        if cleaned:
            yield cleaned

def consumer(items):
    """消费者：消费清洗后的数据"""
    count = 0
    for item in items:
        print(f"处理：{item}")
        count += 1
        if count >= 5:
            break

raw_data = ["  Hello  ", "  World  ", "", "  Python  ", " "]
result = producer(raw_data)
consumer(result)
# 输出：
# 处理：hello
# 处理：world
# 处理：python

异常处理与关闭

def safe_generator(n):
    try:
        for i in range(n):
            yield i
    except GeneratorExit:
        print("生成器被提前终止")

gen = safe_generator(10)
for x in gen:
    print(x, end=" ")
    if x >= 3:
        gen.close()      # 主动关闭

send() 方法 — 双向通信

基本用法

def accumulator():
    """接收外部输入并累加"""
    total = 0
    value = None
    
    while True:
        value = yield total  # yield 同时返回值并接受外部传入的值
        if value is not None:
            total += value

acc = accumulator()
next(acc)              # 启动生成器
print(acc.send(10))    # 10 — 发送 10，返回当前总和
print(acc.send(20))    # 30 — 发送 20，返回 10+20
print(acc.send(5))     # 35 — 发送 5，返回 10+20+5

实用示例：实时数据聚合

def moving_average(window_size):
    """移动平均线计算"""
    window = []
    total = 0
    
    while True:
        value = yield total / len(window) if window else 0
        
        window.append(value)
        total += value
        
        if len(window) > window_size:
            removed = window.pop(0)
            total -= removed

avg = moving_average(3)  # 窗口大小为 3
next(avg)

values = [10, 20, 30, 40, 50]
for v in values:
    result = avg.send(v)
    print(f"数据={v}, 移动平均={result:.1f}")
# 输出：
# 数据=10, 移动平均=10.0
# 数据=20, 移动平均=15.0
# 数据=30, 移动平均=20.0
# 数据=40, 移动平均=30.0  ← (20+30+40)/3

装饰器增强生成器

import functools

def generator_logger(func):
    """为生成器添加日志记录"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        gen = func(*args, **kwargs)
        counter = {"calls": 0}
        
        def logging_gen():
            while True:
                try:
                    value = next(gen)
                    counter["calls"] += 1
                    print(f"[{counter['calls']}] 产出了 {value}")
                    yield value
                except StopIteration:
                    print(f"生成器结束，共输出 {counter['calls']} 个值")
                    return
        
        return logging_gen()
    
    return wrapper

@generator_logger
def count_to_five():
    for i in range(1, 6):
        yield i * 10

for val in count_to_five():
    pass
# 输出：
# [1] 产出了 10
# [2] 产出了 20
# ...
# 生成器结束，共输出 5 个值

iter 与 next — 迭代器协议

自定义可迭代对象

class FibonacciIterable:
    """斐波那契数列迭代器"""
    
    def __init__(self, max_count):
        self.max_count = max_count
        self.count = 0
        self.a, self.b = 0, 1
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.count >= self.max_count:
            raise StopIteration
        
        result = self.a
        self.a, self.b = self.b, self.a + self.b
        self.count += 1
        return result

fib = FibonacciIterable(10)
for num in fib:
    print(num, end=" ")
# 输出：0 1 1 2 3 5 8 13 21 34

ReverseIterator 示例

class ReverseIterator:
    def __init__(self, data):
        self.data = data
        self.index = len(data)
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.index <= 0:
            raise StopIteration
        
        self.index -= 1
        return self.data[self.index]

words = ["Python", "is", "awesome"]
for word in ReverseIterator(words):
    print(word, end=" ")
# 输出：awesome is Python

RangeIterable 示例

class RangeIterable:
    """类似 range() 的可迭代对象"""
    
    def __init__(self, start, stop=None, step=1):
        if stop is None:
            self.start, self.stop = 0, start
        else:
            self.start, self.stop = start, stop
        self.step = step
    
    def __iter__(self):
        return RangeIterator(self.start, self.stop, self.step)

class RangeIterator:
    def __init__(self, start, stop, step):
        self.current = start
        self.stop = stop
        self.step = step
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.step > 0 and self.current >= self.stop:
            raise StopIteration
        if self.step < 0 and self.current <= self.stop:
            raise StopIteration
        
        value = self.current
        self.current += self.step
        return value

# 使用
for i in RangeIterable(10, 0, -2):
    print(i, end=" ")
# 输出：10 8 6 4 2

内置工具：enumerate & zip & map/filter

enumerate 和 zip

names = ["Alice", "Bob", "Charlie"]
scores = [95, 88, 92]

# enumerate — 获取索引
for i, name in enumerate(names, start=1):
    print(f"{i}. {name}")

# zip — 并行遍历
for name, score in zip(names, scores):
    print(f"{name}: {score}")

# zip_longest — 补齐最短长度
from itertools import zip_longest
for a, b in zip_longest([1, 2], ["a"], fillvalue="?"):
    print(a, b)
# 输出：1 a / 2 ?

map 和 filter

# map — 转换
nums = [1, 2, 3, 4, 5]
doubled = list(map(lambda x: x * 2, nums))
print(doubled)  # [2, 4, 6, 8, 10]

# filter — 过滤
evens = list(filter(lambda x: x % 2 == 0, nums))
print(evens)  # [2, 4]

# 等同于列表推导式
squares_lc = [x ** 2 for x in nums]
odds_lc = [x for x in nums if x % 2 != 0]

chain 与 from_iterable

from itertools import chain

# chain — 连接多个可迭代对象
combined = chain([1, 2], [3, 4], [5, 6])
print(list(combined))  # [1, 2, 3, 4, 5, 6]

# from_iterable — 直接接收一个可迭代的迭代器
nested = [[1, 2], [3, 4], [5, 6]]
flat = chain.from_iterable(nested)
print(list(flat))  # [1, 2, 3, 4, 5, 6]

实战示例

大文件按块读取

def read_large_file(filepath, chunk_size=8192):
    """生成器方式逐块读取超大文件"""
    with open(filepath, "r", encoding="utf-8") as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

for line_number, chunk in enumerate(read_large_file("huge_file.txt")):
    print(f"块 {line_number + 1} ({len(chunk)} 字节)")
    # 内存中始终只保留一个 chunk 的数据

递归目录扫描

import os

def scan_directory(path):
    """生成器扫描目录下所有文件"""
    for root, dirs, files in os.walk(path):
        for filename in files:
            filepath = os.path.join(root, filename)
            size = os.path.getsize(filepath)
            if size > 1024:  # 只显示大于 1KB 的文件
                yield filepath, size

for filepath, size in scan_directory("."):
    print(f"{filepath}: {size} 字节")

管道式数据处理

def read_lines(filepath):
    """逐行读取"""
    with open(filepath, "r") as f:
        for line in f:
            yield line.strip()

def filter_empty(lines):
    """过滤空行"""
    for line in lines:
        if line:
            yield line

def parse_data(lines):
    """解析 CSV 格式"""
    for line in lines:
        fields = line.split(",")
        if len(fields) >= 3:
            yield {
                "name": fields[0].strip(),
                "age": int(fields[1].strip()),
                "city": fields[2].strip()
            }

# 管道组合
with open("data.csv") as f:
    data = parse_data(
        filter_empty(
            read_lines("data.csv")
        )
    )
    
    for record in data:
        print(record)

总结

本文讲解了生成器与迭代器的核心知识：

  1. 生成器基础 — yield 语法、惰性求值、内存效率对比
  2. send() 双向通信 — 接受外部输入，实现协程式交互
  3. 装饰器增强 — 日志记录、性能监控、错误处理
  4. 迭代器协议 — __iter__ 和 __next__ 自定义迭代器
  5. 内置工具 — enumerate/zip/map/filter/chained iterators
  6. 实际场景 — 大文件分块、目录递归扫描、管道数据处理
  7. 性能优势 — 按需生成，内存占用极小

下一篇文章将深入讲解高阶函数：map、reduce、filter 的高级用法。

本文是 Python 教程系列的第十九篇。