【Python】迭代器

迭代器

迭代器概念

可迭代对象（Iterable）

• 定义：
  只要实现了 __iter__() 方法，返回一个迭代器对象的类实例，就称为 可迭代对象。
• 典型代表：
  • list、tuple、str、dict、set、range
  • 文件对象
• 检测方法：

from collections.abc import Iterableprint(isinstance([1,2,3], Iterable))  # True
print(isinstance(123, Iterable))      # False

特点：可迭代对象本身不能直接用 next() 取值，需要先通过 iter() 生成迭代器。

迭代器（Iterator）

• 定义：
  实现了 迭代协议 的对象，即：
  • __iter__()：返回迭代器对象本身。
  • __next__()：返回下一个元素；没有元素时抛出 StopIteration。
• 检测方法：

from collections.abc import Iteratorlst = [1, 2, 3]
it = iter(lst)
print(isinstance(it, Iterator))   # True

迭代器工作原理

以这个例子为基础：

my_list = [1, 2, 3, 4, 5]# 获取迭代器
iterator = iter(my_list)  # 等价于 my_list.__iter__()# 手动迭代
print(next(iterator))  # 1
print(next(iterator))  # 2
print(next(iterator))  # 3
print(next(iterator))  # 4
print(next(iterator))  # 5
print(next(iterator))  # StopIteration 异常

可迭代对象与迭代器的生成

1. my_list = [1, 2, 3, 4, 5]
   • my_list 是一个 可迭代对象（Iterable）。
   • 它内部实现了 __iter__() 方法。
2. iterator = iter(my_list)
   • Python 调用 my_list.__iter__()，返回一个 迭代器对象。
   • 这个迭代器对象内部维护了一个 “游标（current index）”，指向下一个要返回的元素。
   • 这里的游标初始值是 0。

底层概念：

my_list -> iterable -> __iter__() -> iterator (游标=0)

调用 next(iterator) 的机制

每次调用 next(iterator) 时，Python 会执行以下步骤：

1. 检查迭代器内部游标是否到达序列末尾：
   • 如果没有到达，就返回当前元素，并把游标 +1。
   • 如果到达末尾，就抛出 StopIteration 异常。
2. 返回值：
   • 第一次调用 next() → 返回 my_list[0] → 1
   • 第二次调用 next() → 返回 my_list[1] → 2
   • …
   • 第五次调用 next() → 返回 my_list[4] → 5
3. 第六次调用 next() → 游标已经超出范围 → 抛出 StopIteration 异常

底层逻辑示意：

class ListIterator:def __init__(self, data):self.data = dataself.index = 0   # 游标def __iter__(self):return selfdef __next__(self):if self.index < len(self.data):val = self.data[self.index]self.index += 1return valelse:raise StopIteration

• 实际上 Python 的列表迭代器就是这个逻辑，只是 C 语言实现，效率更高。

游标（内部指针）作用

• 迭代器内部维护的游标是 关键机制：
  • 它让迭代器 记住上一次取到哪，下一次直接从上次位置继续。
  • 这就是为什么迭代器 一次性使用，不能回退，除非重新创建迭代器。

it = iter([1,2,3])
print(next(it))  # 1
print(next(it))  # 2# 遍历完成后，如果想重新从头遍历：
it = iter([1,2,3])  # 必须重新生成迭代器

for 循环与迭代过程的原理

for x in [1,2,3,4,5]:print(x)

内部执行流程等价于：

it = iter([1,2,3,4,5])
while True:try:x = next(it)print(x)except StopIteration:break

解释：当执行 for x in obj: 时，Python 内部流程是：

1. 调用 obj.__iter__() → 得到迭代器。
2. 调用 迭代器.__next__() → 依次取值。
3. 捕获到 StopIteration → 结束循环。

所以 for 循环其实就是 迭代器协议的封装，不必手动捕获 StopIteration。

内存与效率

• 列表 vs 迭代器：
  • 列表一次性存储所有元素，占用内存 O(n)。
  • 迭代器每次只生成一个元素，惰性计算（lazy evaluation），适合大数据或流式数据。

import syslst = list(range(10**6))
it = iter(lst)print(sys.getsizeof(lst))  # 占用内存大
print(sys.getsizeof(it))   # 占用内存很小

自定义迭代器

为什么需要自定义迭代器？

Python 内置的迭代器（如 list_iterator, tuple_iterator）可以遍历标准容器。但有时候我们需要：

1. 迭代非标准序列：例如数学序列（斐波那契、素数等）
2. 懒加载数据：按需生成，不一次性加载所有元素
3. 自定义遍历规则：倒计时、步长不规则、过滤条件

这时就需要自己定义迭代器类。

要成为一个迭代器对象，需要实现迭代器协议：

1. __iter__(self)
   • 返回迭代器自身
   • 让迭代器对象可以被 for 循环使用
2. __next__(self)
   • 返回下一个元素
   • 如果没有更多元素，抛出 StopIteration

基础示例：倒计时迭代器

class CountDown:def __init__(self, start):self.current = start  # 当前数字def __iter__(self):# 返回迭代器自身return selfdef __next__(self):# 判断是否已经结束if self.current <= 0:raise StopIteration# 先返回当前数字，再递减value = self.currentself.current -= 1return value# 使用自定义迭代器
count_down = CountDown(5)
for num in count_down:print(num)

输出：

5
4
3
2
1

解析：

1. count_down = CountDown(5)
   • 创建实例，self.current = 5
2. for num in count_down:
   • 内部调用 count_down.__iter__() → 返回迭代器本身
   • 每次循环调用 count_down.__next__() → 返回下一个元素
3. 当 self.current <= 0 时，抛出 StopIteration → for 循环自动结束

注意事项

1. 一次性使用

   it = CountDown(3)
   print(list(it))  # [3, 2, 1]
   print(list(it))  # []，迭代器耗尽
   

2. 返回值顺序
   • __next__() 可以先返回值再修改状态，也可以先修改再返回
   • 例如：

     def __next__(self):if self.current <= 0:raise StopIterationself.current -= 1return self.current + 1
     

3. 不要忘记抛出 StopIteration
   • 这是告诉 Python 迭代结束的信号
   • 没有它会导致无限循环

示例 1：步长可自定义的倒计时

class StepCountDown:def __init__(self, start, step=1):self.current = startself.step = stepdef __iter__(self):return selfdef __next__(self):if self.current <= 0:raise StopIterationvalue = self.currentself.current -= self.stepreturn value# 步长为2
for num in StepCountDown(10, step=2):print(num)

输出：

10
8
6
4
2

示例 2：斐波那契数列迭代器

class Fibonacci:def __init__(self, n):self.n = nself.count = 0self.a, self.b = 0, 1def __iter__(self):return selfdef __next__(self):if self.count >= self.n:raise StopIterationvalue = self.aself.a, self.b = self.b, self.a + self.bself.count += 1return value# 前10个斐波那契数
for num in Fibonacci(10):print(num)

输出：

0
1
1
2
3
5
8
13
21
34

特点：

• 内部维护状态 (self.a, self.b)
• 每次调用 __next__() 都计算下一个值
• 避免一次性生成整个序列 → 节省内存

生成器

生成器本质上是 迭代器的高级简化版，它让我们可以用函数的方式按需生成序列，而不必自己手写迭代器类。

生成器函数基本概念

• 定义：函数内部使用 yield 关键字代替 return，函数就会返回一个 生成器对象（Generator）。
• 生成器对象特点：
  1. 自动实现迭代器协议（__iter__() + __next__()）
  2. 惰性计算：每次 next() 时才计算下一个值
  3. 只保存当前状态，不存储整个序列 → 节省内存

示例：倒计时生成器

def count_down(start):while start > 0:yield startstart -= 1# 使用生成器
for num in count_down(5):print(num)

输出：

5
4
3
2
1

解析：

1. count_down(5) 并没有执行函数体，而是返回一个 生成器对象。
2. for 循环内部调用生成器对象的 __iter__() → 返回自身
3. 每次循环调用 __next__()：
   • 执行到 yield 语句
   • 返回 yield 后的值
   • 暂停函数状态，保留局部变量（如 start）
4. 当 while start > 0 条件不成立 → 抛出 StopIteration → 循环结束

特点：

• 不必写类来管理游标和状态
• 支持复杂逻辑，只需要 yield 中断函数执行并返回值
• 内存占用小，适合大数据流和无限序列

手动控制生成器

生成器也可以用 next() 手动迭代：

gen = count_down(3)
print(next(gen))  # 3
print(next(gen))  # 2
print(next(gen))  # 1
print(next(gen))  # StopIteration

说明：

• 每次 next(gen) 都会从上次 yield 暂停的位置继续执行
• 状态（局部变量）自动保存，无需自己管理

生成器表达式

生成器表达式是 列表推导式的惰性版本，用括号 () 包裹表达式即可。

1. 基本用法

   gen = (x*x for x in range(5))print(next(gen))  # 0
   print(next(gen))  # 1
   print(list(gen))  # [4, 9, 16]
   

   解析：

   1. (x*x for x in range(5)) 生成一个 生成器对象
   2. 每次 next() 执行一次计算
   3. list(gen) 会把剩余的所有元素计算出来

2. 与列表推导式对比

   特性	列表推导式 [x*x for x in range(5)]	生成器表达式 (x*x for x in range(5))
   内存占用	一次性生成整个列表，占用内存大	惰性计算，每次生成一个元素，占用内存小
   适合数据量	小数据量	大数据量 / 无限序列
   用法	list、for 循环	next()、for 循环、函数迭代

生成器的高级特性

1. send() 方法

   • 生成器对象可以接收外部传入的值，用 send(value)
   • 第一次调用必须用 next() 或 send(None) 启动生成器

   def echo():while True:received = (yield)print("Received:", received)gen = echo()
   next(gen)           # 启动生成器
   gen.send("Hello")   # 输出: Received: Hello
   gen.send("World")   # 输出: Received: World
   

2. throw() 方法

   • 向生成器抛出异常，在生成器内部捕获处理

   def gen():try:yield 1except ValueError:print("Caught ValueError")g = gen()
   print(next(g))    # 1
   g.throw(ValueError)  # 输出: Caught ValueError
   

3. close() 方法

   • 关闭生成器，抛出 GeneratorExit 异常

   g = count_down(3)
   print(next(g))  # 3
   g.close()       # 关闭生成器
   # 再调用 next(g) 会抛出 StopIteration
   

生成器 vs 自定义迭代器

特性	自定义迭代器类	生成器
定义方式	类 + __iter__ + __next__	函数 + yield
状态管理	手动管理（游标、计数器）	自动保存局部变量状态
内存	适合大数据	更省内存，惰性计算
语法简洁	较繁琐	简单、清晰

迭代器工具（itertools 模块）

itertools.chain() — 连接多个迭代器

功能：将多个可迭代对象 按顺序连接成一个迭代器，返回的迭代器会依次遍历每个输入迭代器的元素。

示例

import itertools# 连接两个列表
result = itertools.chain([1, 2], [3, 4])
print(list(result))  # [1, 2, 3, 4]# 连接字符串和元组
result2 = itertools.chain("AB", (5, 6))
print(list(result2))  # ['A', 'B', 5, 6]

原理

• chain 不会生成一个新的列表，而是 按需遍历每个输入迭代器，节省内存。
• 相当于：

def chain(*iterables):for it in iterables:for element in it:yield element

使用场景

• 合并多个迭代器而不占用额外内存
• 流式数据处理，如日志合并、多数据源合并

itertools.cycle() — 无限循环迭代器

功能

• 对一个可迭代对象的元素 无限循环返回
• 适合需要重复序列的场景

示例

import itertoolscy = itertools.cycle("AB")  # 无限循环 A、B
print([next(cy) for _ in range(5)])  # ['A', 'B', 'A', 'B', 'A']

原理

• 内部保存原序列
• 游标达到末尾时重新从头开始
• 注意：无限循环，如果不手动停止会永远运行

使用场景

• 轮询任务、循环队列、重复模式生成
• 无限序列计算（例如轮询资源）

itertools.islice() — 可迭代对象切片

功能

• 对迭代器进行 惰性切片，类似 list[start:stop:step]
• 不需要一次性生成完整序列，节省内存

参数

itertools.islice(iterable, start, stop[, step])

• iterable：可迭代对象
• start：起始索引（默认 0）
• stop：结束索引（不包含 stop）
• step：步长（可选）

示例

import itertools# 从 0~9 的序列，取索引 2 到 6，每隔 2 个取一个
result = itertools.islice(range(10), 2, 7, 2)
print(list(result))  # [2, 4, 6]

原理

• 内部迭代迭代器
• 只返回符合索引条件的元素
• 不生成中间序列，支持无限序列

使用场景

• 从大数据或无限序列中取部分元素
• 替代切片操作节省内存

itertools.product() — 笛卡尔积

功能

• 返回输入可迭代对象的 笛卡尔积
• 每个组合都以元组形式返回

参数

itertools.product(*iterables, repeat=1)

• iterables：可迭代对象列表
• repeat：重复次数

示例

import itertools# 两个迭代器的笛卡尔积
result = itertools.product("AB", [1, 2])
print(list(result))  
# [('A', 1), ('A', 2), ('B', 1), ('B', 2)]# 重复两次，相当于自己与自己做笛卡尔积
result2 = itertools.product([0,1], repeat=2)
print(list(result2))
# [(0, 0), (0, 1), (1, 0), (1, 1)]

原理

• 内部生成器按顺序计算每一种组合
• 支持任意长度迭代器
• 惰性生成组合，节省内存

使用场景

• 生成所有可能组合
• 穷举搜索问题（如排列、密码穷举、棋盘格坐标）

itertools.count() — 无限计数器

功能

• 返回一个从 start 开始的 无限整数迭代器，可指定步长
• 非常适合生成序号、时间戳等

示例

import itertoolscnt = itertools.count(10, 2)  # 从 10 开始，每次 +2
for i in range(5):print(next(cnt))  
# 10 12 14 16 18

原理

• 内部维护一个计数器
• 每次 next() 返回计数器当前值，并更新计数器

使用场景

• 无限序列生成
• 需要自动编号或时间戳

itertools.repeat() — 重复元素

功能

• 重复某个元素 n 次（或无限次）

示例

import itertoolsrep = itertools.repeat("A", 3)
print(list(rep))  # ['A', 'A', 'A']

使用场景

• 填充默认值
• 测试数据生成

itertools 模块主要提供以下优势：

函数	功能
chain	连接多个迭代器
cycle	无限循环迭代
islice	迭代器切片
product	笛卡尔积
count	无限计数器
repeat	重复元素
combinations	元素组合（C(n, k)）
permutations	元素排列（P(n, k)）
accumulate	累积和或其他二元操作

异步迭代器

异步迭代器协议

要成为异步迭代器，需要实现 异步迭代器协议：

1. __aiter__(self)
   • 返回异步迭代器对象自身
   • 类似普通迭代器的 __iter__()
2. __anext__(self)
   • 返回下一个元素
   • 必须是 协程（async 函数）
   • 元素生成完成前可以 await 异步操作
   • 没有更多元素时抛出 StopAsyncIteration

注意：__anext__() 必须是 async def，否则无法使用 await。

异步 for 循环

• 异步迭代器只能在 async for 中使用
• async for 会自动调用 __aiter__() 和 __anext__()
• 遇到 StopAsyncIteration 时结束循环

async for item in async_iter:# 异步迭代内容

异步迭代器示例：生成奇数

import asyncioclass AsyncOdd:def __init__(self):self.num = -1def __aiter__(self):# 返回异步迭代器对象自身return selfasync def __anext__(self):# 迭代条件if self.num >= 9:raise StopAsyncIterationself.num += 2await asyncio.sleep(1)  # 模拟异步等待return self.num# 主协程
async def main():async for x in AsyncOdd():print(x)# 执行异步循环
asyncio.run(main())

输出（每隔1秒打印一个数字）：

1
3
5
7
9

工作原理解析

1. async for x in AsyncOdd()
   • Python 调用 AsyncOdd().__aiter__() → 返回异步迭代器对象
2. 循环中每次迭代调用 await __anext__()
   • 执行 __anext__() 内部逻辑
   • 遇到 await asyncio.sleep(1) 时暂停当前协程，让出控制权给事件循环
3. 当 self.num >= 9 时，抛出 StopAsyncIteration
   • async for 循环捕获异常 → 循环结束

普通迭代器 vs 异步迭代器

特性	普通迭代器	异步迭代器
协程支持	不支持	支持 await
用法	for	async for
协议方法	__iter__() + __next__()	__aiter__() + __anext__()
异步 IO	阻塞	非阻塞
使用场景	内存数据迭代	网络请求、文件异步读取、定时任务等

异步生成器（异步迭代器的简化）

Python 3.6+ 支持 异步生成器：

import asyncioasync def async_odd_gen():num = 1while num < 10:await asyncio.sleep(1)yield numnum += 2async def main():async for x in async_odd_gen():print(x)asyncio.run(main())

• 不用写类，直接用 async def + yield
• 每次迭代可以 await 异步操作
• 更简洁、清晰

多异步迭代器组合

async def gen1():for i in range(3):await asyncio.sleep(1)yield iasync def gen2():for i in range(3,6):await asyncio.sleep(1)yield iasync def main():async for x in gen1():print(x)async for y in gen2():print(y)asyncio.run(main())

异步迭代器异常处理

class AsyncFail:def __init__(self):self.n = 0def __aiter__(self):return selfasync def __anext__(self):self.n += 1if self.n > 3:raise StopAsyncIterationif self.n == 2:raise ValueError("模拟异常")return self.nasync def main():async for x in AsyncFail():try:print(x)except ValueError as e:print("捕获异常:", e)asyncio.run(main())