迭代器与生成器

我们在使用 Python 编程的时候经常会用到一个 for 循环 for i in range(10): ，你知道它背后的逻辑是什么样的吗？

Python 的 for 循环不是直接遍历对象，底层是迭代器协议的自动化实现。

可迭代对象 Iterable VS 迭代器 Iterator：

• 可迭代对象 Iterable ：只要对象实现了 __iter__() 方法，就是可迭代对象。

• 常见的可迭代对象有 list、tuple、str、dict、set、生成器等。

• 迭代器 Iterator：同时实现了 __iter__() 和 __next__() 方法的对象：

• __iter__() 方法返回对象自身。
• __next__() 方法返回下一个元素；遍历完毕抛出 StopIteration 异常。

Python 中只要是可迭代对象 Iterable 就可以被 for 循环遍历。

from collections.abc import Iterableisinstance(obj, Iterable)  # 结果为 True 就可以被 for 循环遍历for x in obj:    循环体

在 for i in range(10): 循环遍历语句中，range(10) 本质就是创建一个可迭代对象 Iterable，但它并不是迭代器 Iterator，for 循环底层的执行逻辑如下：

# for x in obj:#     循环体# for 循环底层实现，伪代码：iterator = iter(obj) # iter() 将可迭代对象转换成迭代器whileTrue:try:        item = next(iterator)        循环体except StopIteration:break

所以只要是可迭代对象就可以被 for 循环遍历，但反过来，能被 for 循环遍历的不一定是可迭代对象。

从 Python 的设计思想层面来看，for 循环的核心机制是迭代器协议 Iterable 。但 Python 为了兼容历史旧版本，保留了不是可迭代对象但实现了 __getitem__() 方法并且支持整数从 0 开始索引的对象仍可被 for 循环遍历，Python 会为其自动创建一个迭代器。

classDemo:def__getitem__(self, index):if index >= 10:raise IndexErrorreturn indexdemo = Demo()from collections.abc import Iterableisinstance(demo, Iterable)  # Falsefor i in demo:  # 仍可遍历    print(i)

自定义迭代器

实现一个斐波那契数列迭代器：

classFibonacciIterator:"""一个可以无限生成斐波那契数列的迭代器"""def__init__(self, max_count=None):        self.max_count = max_count        self.count = 0        self.a, self.b = 0, 1def__iter__(self):return selfdef__next__(self):if self.max_count isnotNoneand self.count >= self.max_count:raise StopIteration        value = self.a        self.a, self.b = self.b, self.a + self.b        self.count += 1return value

# 使用fib = FibonacciIterator(10)for num in fib:    print(num, end=" ")  # 0 1 1 2 3 5 8 13 21 34

# 等价写法whileTrue:try:        item = next(fib) # 或者 item = fib.__next__()        print(num, end=" ")except StopIteration:break

生成器 Generator —— 优雅的迭代器工厂

生成器是一种特殊的迭代器，它不需要显式定义 __iter__ 和 __next__ 方法，Python 解释器会自动生成这些方法。

生成器两种形式：

• 生成器函数，使用 yield 关键字 —— 函数的"暂停与恢复"。当函数遇到 yield 时：

• 暂停函数执行
• 保存当前所有局部变量状态
• 返回 yield 后面的值给调用者
• 下次调用 next() 时，从暂停点继续执行

• 生成器表达式：类似列表推导式 [x for x in range(100)]，生成器表达式使用圆括号(x for x in range(100)) 。

• 列表推导式会立刻计算所有元素，并把全部结果存放在内存列表中，占用更多内存；优势是数据常驻内存，可以重复遍历、支持索引和切片。

• 生成器表达式并不会提前计算元素，仅仅创建一个生成器对象，采用惰性求值：只有迭代取值的时候，才按需生成下一个元素，初始内存开销很小。但生成器本质是迭代器，元素只能一次性消费，遍历完成后无法再次从头使用，同时不支持索引、切片操作。

• 当元素数量巨大时，列表推导式内存开销更大；如果元素很少，差距可以忽略。

• 数据量小、需要多次遍历，列表推导式合适；海量数据、不需要重复读取、想节省内存，生成器表达式合适。

  import syslist_comp = [i for i in range(1000000)]gen_exp = (i for i in range(1000000))print(sys.getsizeof(list_comp))  # 约 8MBprint(sys.getsizeof(gen_exp))    # 约 120 字节（固定！）

defsimple_generator():    print("=== 开始执行 ===")yield1    print("恢复执行，获取第二个值")yield2    print("恢复执行，获取第三个值")yield3    print("=== 函数结束，即将抛出 StopIteration ===")print(type(simple_generator))   # <class 'function'>  生成器函数本质只是一个函数from collections.abc import Iteratorgen = simple_generator()    # 当生成器函数被调用的时候，函数体不会立即执行，而是返回一个迭代器对象print(type(gen))# <class 'generator'>  迭代器print(isinstance(gen, Iterator))# Trueg = (x for x in range(100))  # 生成器表达式，本质就是一个迭代器print(isinstance(g, Iterator))# True

defsimple_generator():    print("=== 开始执行 ===")yield1    print("恢复执行，获取第二个值")yield2    print("恢复执行，获取第三个值")yield3    print("=== 函数结束，即将抛出 StopIteration ===")gen = simple_generator()print(next(gen))print(next(gen))print(next(gen))print(next(gen))

运行：

=== 开始执行 ===1恢复执行，获取第二个值2恢复执行，获取第三个值3=== 函数结束，即将抛出 StopIteration ===StopIteration

如果使用迭代器实现上面生成器函数的效果，那么将是这样的：

classGeneratorSimulator:def__init__(self):        self.ip = 0        self.finished = Falsedef__iter__(self):return selfdef__next__(self):if self.finished:raise StopIterationif self.ip == 0:            print("=== 开始执行 ===")            self.ip = 1return1elif self.ip == 1:            print("恢复执行，获取第二个值")            self.ip = 2return2elif self.ip == 2:            print("恢复执行，获取第三个值")            self.ip = 3return3elif self.ip == 3:            print("=== 函数结束，即将抛出 StopIteration ===")            self.finished = Trueraise StopIterationgen = GeneratorSimulator()print(next(gen))print(next(gen))print(next(gen))print(next(gen))

运行：

=== 开始执行 ===1恢复执行，获取第二个值2恢复执行，获取第三个值3=== 函数结束，即将抛出 StopIteration ===StopIteration

虽然两者的最终效果是一样的，但其背后的底层原理完全不同。

从 CPython 视角来看，真正的生成器函数的底层原理是靠一个结构体记录函数的运行状态：

typedefstruct {    PyObject_HEAD    PyFrameObject *gi_frame; // 帧对象：保存局部变量和字节码的位置    PyObject *gi_code;   // 代码对象char gi_running;   // 是否正在运行char gi_suspended;   // 是否暂停在 yield} genobject;

每次调用 next(gen)：

1. 恢复 gi_frame 中保存的字节码位置
2. 执行字节码直到遇到 YIELD_VALUE 指令
3. 暂停执行，保存当前帧状态
4. 返回 yield 的值

生成器实现斐波那契数列：

deffibonacci_generator(max_count=None):"""生成器版本的斐波那契，代码更简洁"""    a, b = 0, 1    count = 0while max_count isNoneor count < max_count:yield a        a, b = b, a + b        count += 1for num in fibonacci_generator(10):    print(num, end=" ")  # 输出 0 1 1 2 3 5 8 13 21 34

和前面那段使用迭代器实现的斐波那契数列相比代代码量要少很多。

生成器高级特性

send() 方法双向通信

defecho_generator():    received = yield"准备接收"whileTrue:        received = yieldf"收到：{received}"gen = echo_generator()print(next(gen))   # 准备接收print(gen.send("hello")) # 收到：helloprint(gen.send("world")) # 收到：world

• 首次启动生成器必须使用 next(gen) 或 gen.send(None)，因为生成器初始时停在函数开头，需要一个 yield 来接收第一个值。
• send(value) 会恢复执行，并将 value 赋值给当前 yield 表达式的结果。

throw() / close() 异常处理与资源释放

defsafe_generator():try:yield"运行中"except ValueError:yield"捕获到 ValueError"finally:        print("资源清理（关闭文件、数据库连接等）")gen = safe_generator()print(next(gen))             # 运行中print(gen.throw(ValueError))   # 捕获到 ValueErrorgen.close()                  # 输出: 资源清理...

生成器函数异常抛出方式：

• try 代码段运行发生异常。
• 外部 throw() 方法主动抛出异常。

yield from 生成器的委托（子生成器）

yield from 用于将迭代任务委托给另一个生成器：

defsub_generator():yield1yield2yield3defmain_generator():yield"开始"yieldfrom sub_generator()  # 完全委托yield"结束"for item in main_generator():    print(item)  # 开始 1 2 3 结束

yield from 处理嵌套列表扁平化处理：

defflatten(nested_list):for sublist in nested_list:if isinstance(sublist, list):yieldfrom flatten(sublist)  # 递归委托else:yield sublistnested = [1, [2, [3, 4], 5], 6]print(list(flatten(nested)))  # [1, 2, 3, 4, 5, 6]

生成器妙用 —— 依赖注入

FastAPI 数据库会话依赖注入：

from fastapi import FastAPI, Dependsfrom sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmaker, Sessionapp = FastAPI()engine = create_engine("Database_url")SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)defget_db() -> Session: db = SessionLocal()try:yield db  # 将数据库会话注入到路由  db.commit()  # 如果一切正常，提交事务except Exception:  db.rollback() # 出错回滚raisefinally:  db.close()  # 数据库事务后关闭连接，释放资源# 路由：依赖注入@app.get("/items/{item_id}")defread_item(item_id: int, db: Session = Depends(get_db)):"""    db 参数从 get_db 注入    当路由函数执行完毕后，get_db 的 finally 会自动执行    """    item = db.query(Item).filter(Item.id == item_id).first()return item

执行流程：

1. 请求进入 → FastAPI 调用 get_db()2. get_db() 执行到 yield db → 暂停，db 被注入到路由3. 路由函数执行（使用 db）4. 路由返回响应 → FastAPI 继续执行 get_db()5. 执行 commit/rollback → finally 中的 close()6. 响应返回给客户端

Redis 依赖注入：

# 创建 Redis 连接池redis_pool = redis.ConnectionPool.from_url("Redis_Url",    decode_responses=True,    socket_connect_timeout=5,    socket_timeout=5,    retry_on_timeout=True,    health_check_interval=30,    max_connections=20)defget_redis():"""依赖注入：从连接池获取 Redis 连接"""    client = redis.Redis(connection_pool=redis_pool)try:yield clientexcept redis.ConnectionError as e:        logger.error(f"Redis connection error: {e}")raisefinally:        client.close()

yield 在这里实现了"资源的获取与释放"的完美配对，这是依赖注入中生命周期管理的最佳实践。