字数 3827，阅读大约需 20 分钟
在Python编程中，生成器(Generators)是一个既强大又优雅的特性。如果曾经遇到过处理大量数据时内存不足的问题，或者想要创建无限序列，那么生成器就是你的最佳选择。
什么是生成器？
生成器是一种特殊的迭代器(Iterator)，它允许你定义一个迭代行为，但不会在内存中一次性构建整个序列。相反，它会在每次需要时按需生成下一个值。
核心思想：延迟计算(Lazy Evaluation)和按需生成(Generate on demand)。
生成器的实战示例
结合代码开发，下面是基于 FastAPI 和 SQLAlchemy 的Demo。在 FastAPI 中，get_db() 函数是一个依赖注入函数，返回一个生成器（Generator）。自动获取数据库会话和归还数据库会话，且数据库会话的自动归还发生在以下时机：
1. 请求开始时：FastAPI 调用
get_db()，执行到yield db之前，创建并返回数据库会话实例给路由函数使用。2. 请求处理完成后：当路由函数执行完毕后，FastAPI 会继续执行生成器的剩余部分，即
finally块中的db.close()，自动关闭数据库会话并归还连接到连接池。
engine = create_engine(
settings.SQLALCHEMY_DATABASE_URI, # 数据库连接URI，包含主机、端口、数据库名、用户名和密码
pool_pre_ping=True, # 连接池预检查，获取连接前发送测试查询验证连接有效性，防止使用已失效的连接
pool_size=settings.SQLALCHEMY_POOL_SIZE, # 连接池初始大小，保持的最小连接数，减少频繁创建连接的开销
pool_recycle=settings.SQLALCHEMY_POOL_RECYCLE, # 连接回收时间(秒)，超过此时间的连接会被回收重建，防止连接长时间空闲导致失效
pool_timeout=settings.SQLALCHEMY_POOL_TIMEOUT, # 获取连接超时时间(秒)，超过此时间未获取到连接会抛出超时异常
max_overflow=settings.SQLALCHEMY_POOL_OVERFLOW, # 连接池最大溢出连接数，超出pool_size后可额外创建的连接数上限
connect_args=dict(
options=(
"-c idle_in_transaction_session_timeout=" # PostgreSQL会话空闲事务超时设置
f"{settings.POSTGRES_IDLE_IN_TRANSACTION_SESSION_TIMEOUT}" # 空闲事务超时时间(毫秒)，防止长时间占用事务资源
# 添加时区设置（PostgreSQL为例）
f" -c TimeZone=Asia/Shanghai"
)
),
)
adapter = Adapter(engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db() -> Generator:
try:
db = SessionLocal()
yield db
finally:
db.close()为什么使用生成器？
1. 内存效率(Memory Efficiency)
不像列表(list)或元组(tuple)那样一次性将所有数据加载到内存。特别适用于处理大数据文件或庞大的数据集。
2. 性能优化(Performance)
减少了不必要的计算。如果不是所有元素都被使用，那么未被访问的元素就不会被计算，避免了大量数据结构的创建和销毁开销。
3. 代码简洁(Cleaner Code)
生成器表达式通常比等效的列表推导式或循环更简洁，生成器函数通过yield关键字提供了直观的迭代逻辑。
4. 无限序列(Infinite Sequences)
可以轻松生成无限序列（例如斐波那契数列、素数），因为它们不需要预先存储所有元素。
5. 暂停与恢复(Pause and Resume)
生成器函数在yield语句处暂停执行，并返回一个值。当下次请求值时，它会从上次暂停的地方继续执行，并保留局部变量的状态。
创建生成器的两种方式
方式一：生成器函数(Generator Functions)
一个包含yield关键字的函数就成了生成器函数。它不像普通函数那样使用return返回值，而是使用yield。
def count_up_to(n):
i = 1
while i <= n:
print(f"即将 yield {i}")
yield i # 暂停并返回 i
print(f"从 yield {i} 处恢复")
i += 1
print("生成器函数执行完毕")
# 创建生成器对象
my_generator = count_up_to(3)
print("生成器对象已创建:", my_generator)
# 使用 next() 手动迭代
print("第一次 next():", next(my_generator))
print("第二次 next():", next(my_generator))
print("第三次 next():", next(my_generator))
# 尝试再次 next()，将抛出 StopIteration
try:
print("第四次 next():", next(my_generator))
except StopIteration:
print("生成器已耗尽，捕获到 StopIteration 异常。")执行特点：
• 当调用生成器函数时，它不会立即执行函数体，而是返回一个生成器对象
• 每次调用
next()方法时，函数体才会执行到下一个yield语句•
yield返回一个值，并暂停函数的执行，同时保留所有局部变量的状态• 当函数执行完毕时，会抛出
StopIteration异常，表示生成器已耗尽
方式二：生成器表达式(Generator Expressions)
语法上类似列表推导式，但是使用**圆括号()**而不是方括号[]：
# 列表推导式 - 一次性生成所有结果到内存
squares_list = [x * x for x in range(10)]
print("列表:", squares_list)
print("列表类型:", type(squares_list)) # <class 'list'>
# 生成器表达式 - 返回一个生成器对象
squares_generator = (x * x for x in range(10))
print("生成器对象:", squares_generator)
print("生成器类型:", type(squares_generator)) # <class 'generator'>
# 遍历生成器表达式
print("遍历生成器表达式:")
for num in squares_generator:
print(num, end=" ")
# 注意：生成器只能遍历一次！经典应用示例
1. 斐波那契数列生成器
def fibonacci_generator(limit):
"""生成斐波那契数列的前limit个数"""
a, b = 0, 1
count = 0
while count < limit:
yield a
a, b = b, a + b
count += 1
# 使用示例
print("斐波那契数列(前10个):")
for num in fibonacci_generator(10):
print(num, end=" ")
# 输出: 0 1 1 2 3 5 8 13 21 342. 大文件处理
def read_large_file(file_path):
"""逐行读取大文件，避免内存溢出"""
with open(file_path, 'r', encoding='utf-8') as file:
for line in file:
yield line.strip()
# 处理GB级别的文件也不会占用太多内存
def process_large_log(file_path):
error_count = 0
for line in read_large_file(file_path):
if 'ERROR' in line:
error_count += 1
if error_count > 1000: # 只处理前1000个错误
break
return error_count3. 数据管道处理
def read_numbers(filename):
"""读取数字文件"""
with open(filename) as f:
for line in f:
yield int(line.strip())
def square_numbers(numbers):
"""对数字求平方"""
for num in numbers:
yield num ** 2
def filter_even(numbers):
"""过滤偶数"""
for num in numbers:
if num % 2 == 0:
yield num
# 构建处理管道
pipeline = filter_even(square_numbers(read_numbers('numbers.txt')))
for result in pipeline:
print(result)生成器的进阶用法
生成器不仅仅是简单的值序列，它们也可以实现更复杂的交互模式。
1. send()方法：双向通信
def echo_generator():
"""接收并回显消息的生成器"""
while True:
received = yield # 暂停，并等待一个值被发送过来
print(f"收到: {received}")
g = echo_generator()
next(g) # 启动生成器，运行到第一个 yield 处暂停
g.send("Hello") # 发送 "Hello"
g.send("World")
# 输出:
# 收到: Hello
# 收到: World2. throw()方法：异常处理
def error_prone_generator():
"""演示异常处理的生成器"""
try:
yield 1
yield 2
yield 3
except ValueError:
print("生成器内部捕获到 ValueError!")
finally:
print("生成器清理完成。")
g = error_prone_generator()
print(next(g)) # 1
g.throw(ValueError, "Something went wrong!")3. close()方法：优雅关闭
def cleanup_generator():
"""演示清理资源的生成器"""
try:
yield 1
yield 2
finally:
print("生成器被关闭了，执行清理操作。")
g = cleanup_generator()
print(next(g)) # 1
g.close() # 强制关闭生成器实际应用场景
1. 网络爬虫中的URL队列
def url_generator(base_urls):
"""生成待爬取的URL"""
for base_url in base_urls:
for page in range(1, 101): # 假设每个站点有100页
yield f"{base_url}/page/{page}"
# 只有在需要时才生成URL，避免内存占用过大
for url in url_generator(['https://example1.com', 'https://example2.com']):
# 爬取逻辑
crawl_page(url)2. 时间序列数据处理
def moving_average(data, window_size):
"""计算移动平均值"""
window = []
for value in data:
window.append(value)
if len(window) > window_size:
window.pop(0)
if len(window) == window_size:
yield sum(window) / window_size
# 处理实时数据流
sensor_data = get_sensor_data() # 假设这是一个数据流
for avg in moving_average(sensor_data, window_size=5):
print(f"5分钟移动平均: {avg}")3. 树形结构遍历
class TreeNode:
def __init__(self, value):
self.value = value
self.children = []
def dfs_traversal(root):
"""深度优先遍历树结构"""
if root:
yield root.value
for child in root.children:
yield from dfs_traversal(child)
# 使用示例
root = TreeNode('A')
root.children = [TreeNode('B'), TreeNode('C')]
root.children[0].children = [TreeNode('D'), TreeNode('E')]
for node_value in dfs_traversal(root):
print(node_value) # A B D E C性能对比
让我们通过一个实际例子来看看生成器的性能优势：
import sys
import time
# 使用列表推导式
def list_approach(n):
start_time = time.time()
squares = [x*x for x in range(n)]
memory_usage = sys.getsizeof(squares)
# 只使用前10个元素
result = sum(squares[:10])
end_time = time.time()
return result, memory_usage, end_time - start_time
# 使用生成器
def generator_approach(n):
start_time = time.time()
squares = (x*x for x in range(n))
memory_usage = sys.getsizeof(squares)
# 只使用前10个元素
result = sum(next(squares) for _ in range(10))
end_time = time.time()
return result, memory_usage, end_time - start_time
# 测试
n = 1000000
list_result = list_approach(n)
gen_result = generator_approach(n)
print(f"列表方式: 结果={list_result[0]}, 内存={list_result[1]}字节, 时间={list_result[2]:.4f}秒")
print(f"生成器方式: 结果={gen_result[0]}, 内存={gen_result[1]}字节, 时间={gen_result[2]:.4f}秒")最佳实践
1. 选择合适的场景
• 使用生成器：处理大数据集、流式数据、无限序列
• 使用列表：需要随机访问、多次遍历、数据量较小
2. 错误处理
def safe_generator(data):
"""安全的生成器，包含错误处理"""
try:
for item in data:
if item is not None:
yield process_item(item)
except Exception as e:
print(f"生成器处理错误: {e}")
finally:
print("生成器清理完成")3. 生成器组合
def compose_generators(*generators):
"""组合多个生成器"""
for gen in generators:
yield from gen
# 使用示例
gen1 = (x for x in range(5))
gen2 = (x for x in range(5, 10))
combined = compose_generators(gen1, gen2)常见陷阱和注意事项
1. 生成器只能遍历一次
gen = (x for x in range(5))
list1 = list(gen) # [0, 1, 2, 3, 4]
list2 = list(gen) # [] 空列表！2. 延迟计算的副作用
# 错误的做法
funcs = [(lambda: x) for x in range(5)]
# 所有lambda都会返回4（最后一个x的值）
# 正确的做法
funcs = [(lambda x=x: x) for x in range(5)]源代码
# Python生成器学习教程 - 完整用例代码
print("=== Python生成器学习教程 ===\n")
# 1. 基础概念和创建方式
print("--- 1. 基础概念和创建方式 ---")
# 1.1 生成器函数 - 使用yield关键字
def simple_generator():
"""最简单的生成器函数"""
yield 1
yield 2
yield 3
# 创建生成器对象
gen = simple_generator()
print("生成器对象:", gen)
print("类型:", type(gen))
# 使用next()获取值
print("第一次调用next():", next(gen))
print("第二次调用next():", next(gen))
print("第三次调用next():", next(gen))
# 1.2 生成器表达式
print("\n生成器表达式:")
squares = (x*x for x in range(5))
print("生成器表达式创建:", squares)
print("转换为列表:", list(squares))
# 2. 实际应用场景
print("\n--- 2. 实际应用场景 ---")
# 2.1 斐波那契数列生成器
def fibonacci(n):
"""生成前n个斐波那契数"""
a, b = 0, 1
for _ in range(n):
yield a
a, b = b, a + b
print("斐波那契数列前10个:")
fib_gen = fibonacci(10)
print(list(fib_gen))
# 2.2 文件读取生成器（模拟）
def read_lines_simulation():
"""模拟读取文件行的生成器"""
lines = ["第一行数据", "第二行数据", "第三行数据", "第四行数据"]
for line in lines:
yield line
print("\n模拟文件读取:")
file_gen = read_lines_simulation()
for line_num, line in enumerate(file_gen, 1):
print(f"第{line_num}行: {line}")
# 2.3 无限序列生成器
def infinite_counter(start=0):
"""无限计数器生成器"""
count = start
while True:
yield count
count += 1
print("\n无限计数器（显示前5个）:")
counter = infinite_counter()
for _ in range(5):
print(next(counter))
# 3. 高级特性
print("\n--- 3. 高级特性 ---")
# 3.1 生成器委托 - yield from
def subgenerator():
"""子生成器"""
yield 'a'
yield 'b'
return '子生成器结束'
def main_generator():
"""主生成器，使用yield from委托"""
yield '开始'
result = yield from subgenerator()
yield f'主生成器收到: {result}'
yield '结束'
print("生成器委托示例:")
main_gen = main_generator()
for value in main_gen:
print(value)
# 3.2 双向通信 - send()方法
def echo_generator():
"""回声生成器 - 双向通信示例"""
print("生成器启动")
value = yield "准备就绪"
while True:
value = yield f"收到: {value}"
print("\n双向通信示例:")
echo_gen = echo_generator()
print(next(echo_gen)) # 启动生成器
print(echo_gen.send("你好"))
print(echo_gen.send("Python"))
print(echo_gen.send("生成器"))
# 3.3 throw()和close()方法
def safe_generator():
"""安全的生成器，处理异常"""
try:
yield "正常执行1"
yield "正常执行2"
yield "正常执行3"
except GeneratorExit:
print("生成器被关闭")
raise
except ValueError as e:
print(f"收到值错误: {e}")
yield f"错误已处理: {e}"
print("\n异常处理示例:")
safe_gen = safe_generator()
print(next(safe_gen))
print(next(safe_gen))
# 向生成器抛出异常
try:
print(safe_gen.throw(ValueError, "测试异常"))
except StopIteration:
print("生成器结束")
# 4. 性能对比
print("\n--- 4. 性能对比 ---")
import time
import sys
# 列表方式 - 占用内存大
def list_approach(n):
"""列表方式生成平方数"""
return [x*x for x in range(n)]
# 生成器方式 - 占用内存小
def generator_approach(n):
"""生成器方式生成平方数"""
for x in range(n):
yield x*x
# 测试大数据量
print("内存使用对比:")
large_list = list_approach(1000000)
print(f"列表方式内存占用: {sys.getsizeof(large_list)} 字节")
large_gen = generator_approach(1000000)
print(f"生成器方式内存占用: {sys.getsizeof(large_gen)} 字节")
# 5. 实用工具生成器
print("\n--- 5. 实用工具生成器 ---")
# 5.1 批量处理生成器
def batch_processor(items, batch_size):
"""批量处理生成器"""
for i in range(0, len(items), batch_size):
yield items[i:i + batch_size]
data = list(range(20))
print("批量处理示例（每批5个）:")
for batch in batch_processor(data, 5):
print(f"批次: {batch}")
# 5.2 过滤生成器
def filter_generator(items, predicate):
"""过滤生成器"""
for item in items:
if predicate(item):
yield item
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
print("\n过滤偶数:")
even_gen = filter_generator(numbers, lambda x: x % 2 == 0)
print(list(even_gen))
# 5.3 链式生成器
def chain_generators(*generators):
"""链式连接多个生成器"""
for gen in generators:
yield from gen
gen1 = (x for x in range(3))
gen2 = (x for x in range(3, 6))
gen3 = (x for x in range(6, 9))
print("\n链式生成器:")
chained = chain_generators(gen1, gen2, gen3)
print(list(chained))
# 6. 错误处理和调试
print("\n--- 6. 错误处理和调试 ---")
def debug_generator():
"""调试生成器"""
print("[DEBUG] 生成器开始")
value1 = yield 1
print(f"[DEBUG] 收到值: {value1}")
value2 = yield 2
print(f"[DEBUG] 收到值: {value2}")
print("[DEBUG] 生成器结束")
debug_gen = debug_generator()
print("调试生成器运行:")
print(next(debug_gen))
print(debug_gen.send("测试值1"))
# print(debug_gen.send("测试值2")) # 会触发StopIteration
# 7. 生成器模式最佳实践
print("\n--- 7. 最佳实践总结 ---")
# 7.1 使用生成器表达式简化代码
numbers = range(10)
squared = (x**2 for x in numbers if x % 2 == 0)
print("生成器表达式简化:", list(squared))
# 7.2 生成器组合
def pipeline(data):
"""数据处理管道"""
# 步骤1: 过滤偶数
even_numbers = (x for x in data if x % 2 == 0)
# 步骤2: 平方运算
squared = (x**2 for x in even_numbers)
# 步骤3: 限制结果数量
limited = (x for i, x in enumerate(squared) if i < 5)
return limited
print("数据处理管道:")
result = pipeline(range(20))
print(list(result))
# 7.3 生成器 vs 迭代器
print("\n生成器 vs 列表推导式:")
import time
# 列表推导式 - 立即计算
start = time.time()
list_result = [x**2 for x in range(1000000)]
list_time = time.time() - start
print(f"列表推导式时间: {list_time:.4f}秒")
# 生成器表达式 - 惰性计算
start = time.time()
gen_result = (x**2 for x in range(1000000))
gen_time = time.time() - start
print(f"生成器表达式创建时间: {gen_time:.4f}秒")
# 实际使用时才计算
start = time.time()
sum(gen_result)
gen_sum_time = time.time() - start
print(f"生成器求和时间: {gen_sum_time:.4f}秒")
print("\n=== 生成器学习教程结束 ===")
print("关键要点:")
print("1. 使用yield关键字创建生成器")
print("2. 惰性求值，节省内存")
print("3. 支持双向通信和异常处理")
print("4. 适合处理大数据和流式数据")
print("5. 可以组合使用构建复杂的数据处理管道")
评论区