Python生成器与迭代器深度解析

Python生成器与迭代器深度解析

一、迭代器协议

迭代器是实现了__iter__()和__next__()方法的对象。

1.1 自定义迭代器

class Counter:
def __init__(self, start, end):
self.current = start
self.end = end

def __iter__(self):
return self

def __next__(self):
if self.current >= self.end:
raise StopIteration
self.current += 1
return self.current - 1

counter = Counter(0, 5)
for num in counter:
print(num) # 输出: 0 1 2 3 4

二、生成器基础

生成器是一种特殊的迭代器，使用yield关键字定义。

2.1 生成器函数

def fibonacci(n):
a, b = 0, 1
count = 0
while count < n:
yield a
a, b = b, a + b
count += 1

for num in fibonacci(10):
print(num, end=' ')
# 输出: 0 1 1 2 3 5 8 13 21 34

2.2 生成器表达式

# 列表推导式 - 立即创建整个列表
squares_list = [x**2 for x in range(1000000)]

# 生成器表达式 - 惰性求值
squares_gen = (x**2 for x in range(1000000))

三、生成器的优势

3.1 内存效率

# 低效：占用大量内存
def read_large_file_bad(file_path):
with open(file_path) as f:
return f.readlines()

# 高效：使用生成器
def read_large_file_good(file_path):
with open(file_path) as f:
for line in f:
yield line.strip()

3.2 惰性求值

def infinite_sequence():
num = 0
while True:
yield num
num += 1

# 可以处理无限序列
gen = infinite_sequence()
print(next(gen)) # 0
print(next(gen)) # 1
print(next(gen)) # 2

四、生成器的高级特性

4.1 send()方法

def echo_generator():
while True:
received = yield
print(f"收到: {received}")

gen = echo_generator()
next(gen) # 启动生成器
gen.send("Hello") # 输出: 收到: Hello
gen.send("World") # 输出: 收到: World

4.2 双向通信

def running_average():
total = 0
count = 0
average = None
while True:
value = yield average
total += value
count += 1
average = total / count

avg = running_average()
next(avg) # 启动生成器
print(avg.send(10)) # 10.0
print(avg.send(20)) # 15.0
print(avg.send(30)) # 20.0

4.3 throw()方法

def generator_with_exception():
try:
while True:
value = yield
print(f"处理: {value}")
except ValueError as e:
print(f"捕获异常: {e}")

gen = generator_with_exception()
next(gen)
gen.send(10)
gen.throw(ValueError, "这是一个错误")

4.4 close()方法

def closable_generator():
try:
while True:
value = yield
print(f"处理: {value}")
finally:
print("生成器关闭，执行清理")

gen = closable_generator()
next(gen)
gen.send(1)
gen.close() # 输出: 生成器关闭，执行清理

五、yield from语法

yield from允许生成器委托给另一个生成器。

def generator1():
yield 1
yield 2

def generator2():
yield 3
yield 4

def combined():
yield from generator1()
yield from generator2()

for num in combined():
print(num) # 输出: 1 2 3 4

六、实战案例：数据管道

def read_csv(file_path):
"""读取CSV文件"""
with open(file_path) as f:
next(f) # 跳过标题行
for line in f:
yield line.strip().split(',')

def filter_by_age(rows, min_age):
"""过滤年龄"""
for row in rows:
if int(row[1]) >= min_age:
yield row

def extract_names(rows):
"""提取姓名"""
for row in rows:
yield row[0]

# 构建数据管道
pipeline = extract_names(
filter_by_age(
read_csv('users.csv'),
min_age=18
)
)

for name in pipeline:
print(name)

七、实战案例：分批处理

def batch_processor(iterable, batch_size):
"""将可迭代对象分批处理"""
batch = []
for item in iterable:
batch.append(item)
if len(batch) == batch_size:
yield batch
batch = []
if batch: # 处理剩余项
yield batch

data = range(25)
for batch in batch_processor(data, batch_size=10):
print(f"处理批次: {batch}")

八、实战案例：树遍历

class TreeNode:
def __init__(self, value, left=None, right=None):
self.value = value
self.left = left
self.right = right

def inorder_traversal(node):
"""中序遍历"""
if node:
yield from inorder_traversal(node.left)
yield node.value
yield from inorder_traversal(node.right)

# 构建树
root = TreeNode(1,
TreeNode(2, TreeNode(4), TreeNode(5)),
TreeNode(3)
)

for value in inorder_traversal(root):
print(value, end=' ') # 输出: 4 2 5 1 3

九、实战案例：滑动窗口

def sliding_window(iterable, window_size):
"""生成滑动窗口"""
from collections import deque
window = deque(maxlen=window_size)

for item in iterable:
window.append(item)
if len(window) == window_size:
yield list(window)

data = [1, 2, 3, 4, 5, 6, 7, 8]
for window in sliding_window(data, 3):
print(window)
# 输出:
# [1, 2, 3]
# [2, 3, 4]
# [3, 4, 5]
# ...

十、迭代器工具：itertools模块

10.1 无限迭代器

from itertools import count, cycle, repeat

# count: 无限计数
for i in count(10, 2):
if i > 20:
break
print(i) # 10, 12, 14, 16, 18, 20

# cycle: 循环迭代
counter = 0
for item in cycle(['A', 'B', 'C']):
if counter >= 5:
break
print(item)
counter += 1

# repeat: 重复元素
for item in repeat('Hello', 3):
print(item)

10.2 组合迭代器

from itertools import chain, zip_longest, product

# chain: 连接多个迭代器
for item in chain([1, 2], [3, 4], [5, 6]):
print(item)

# zip_longest: 最长的迭代器
for item in zip_longest([1, 2], ['a', 'b', 'c'], fillvalue=0):
print(item) # (1, 'a'), (2, 'b'), (0, 'c')

# product: 笛卡尔积
for item in product([1, 2], ['a', 'b']):
print(item) # (1, 'a'), (1, 'b'), (2, 'a'), (2, 'b')

十一、性能对比

import sys

# 列表推导式
list_comp = [x**2 for x in range(10000)]
print(f"列表大小: {sys.getsizeof(list_comp)} 字节")

# 生成器表达式
gen_exp = (x**2 for x in range(10000))
print(f"生成器大小: {sys.getsizeof(gen_exp)} 字节")

十二、最佳实践

1. 处理大数据集时优先使用生成器
2. 使用生成器表达式替代简单的列表推导式
3. 利用yield from简化生成器委托
4. 合理使用itertools模块提供的工具
5. 注意生成器只能迭代一次的特性

十三、总结

生成器和迭代器是Python中处理序列数据的强大工具。生成器通过惰性求值提供了内存高效的解决方案，特别适合处理大数据集和无限序列。掌握生成器的高级特性如send()、throw()和yield from，可以构建更加灵活和强大的数据处理管道。