python 迭代器和生成器

参考资料: https://www.runoob.com/python3/python3-iterator-generator.html

Python 的迭代器（Iterator) 和 生成器（Generator) 是实现惰性求值、节省内存的重要机制，二者密切相关，但有明确区别。下面分点说明：

---

一、迭代器（Iterator）

定义：

迭代器是一个实现了迭代器协议的对象，即同时具备以下两个方法：

• __iter__()：返回迭代器自身（必须）；
• __next__()：返回序列中的下一个元素；若无更多元素，抛出 StopIteration 异常。

特点：

• 一次性遍历（遍历完即“耗尽”）；
• 惰性求值：元素在需要时才生成（适用于大数据流或无限序列）；
• 可通过 iter() 函数从可迭代对象（如 list、str）获取其迭代器。

自定义迭代器（手动实现协议）

通过类实现 __iter__() 和 __next__() 方法：

class CountIterator: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  # 无元素时抛出异常，终止迭代res = self.currentself.current += 1return res# 使用迭代器
it = CountIterator(1, 3)
print(next(it))  # 1（手动调用 next()）
print(next(it))  # 2
print(next(it))  # 3
print(next(it))  # 抛出 StopIteration

示例：

it = iter([1, 2, 3])		# 将可迭代对象转化未迭代器对象
print(next(it))  # 1
print(next(it))  # 2
print(next(it))  # 3
# next(it) → StopIteration

✅ 所有生成器都是迭代器，但并非所有迭代器都是生成器。

---

二、生成器（Generator）常用 yield

参考资料: https://www.runoob.com/python3/python3-iterator-generator.html

生成器是一种特殊的迭代器，由生成器函数（含 yield 表达式的函数）或生成器表达式创建。

****	同步生成器（普通生成器）	异步生成器
定义语法	def + yield（普通函数）	async def + yield（异步函数）
迭代协议	实现「同步迭代协议」：- __iter__()返回自身- __next__()普通方法（阻塞）	实现「异步迭代协议」：- __aiter__()回自身- __anext__()异步方法（async def，非阻塞）
迭代方式	用 for循环或 next() 迭代	必须用 async for或 await anext() 迭代
内部支持操作	只能调用同步函数（如 time.sleep 、普通 IO），调用异步函数会报错	可调用异步函数（如 await asyncio.sleep 、异步网络请求），也可混用同步操作
执行模型	运行在当前线程，yield 暂停时阻塞线程（但切换成本低）	运行在事件循环，await 时释放事件循环，可并发处理其他任务（非阻塞）
暂停 / 恢复机制	基于 Python 解释器的「生成器暂停机制」（保存局部变量、指令指针）	结合「生成器暂停机制」+「异步事件循环调度」（await 时触发调度）
异常类型	迭代结束抛 StopIteration	迭代结束抛 StopAsyncIteration
资源释放	支持 close()（同步）、throw()（同步抛异常）	支持 aclose()（异步，需 await）、athrow()（异步抛异常，需 await）
内存特性	惰性生成，节省内存（和异步生成器一致）	惰性生成，节省内存（核心共性）

2.1 同步生成器

2.1.1 single_yield

生成器函数中只有一个yield (常见)

样例1：

def single_yield_generator(n):      # 单个yield生成器函数i = 1while i <= n:yield ii += 1def test_yiled_for():for num in single_yield_generator(5):print(num)  # 输出：1 2 3 4 5# 单个yield生成器进行遍历
if __name__ == '__main__': # 使用test_yiled_for()

样例2：

单个yield生成器进行遍历，注意yield的返回时机， 每次yield就会返回，下次从yield的位置开始继续执行

def single_yield_generator(n):      # 单个yield生成器函数i = 1while i <= n:print("yield ........ before")yield iprint("yield ........ after")i += 1print("single_yield_generator 结束")def test_yiled_for():for num in single_yield_generator(5):# print("--------------------------begin")print(num)  # 输出：1 2 3 4 5print("--------------------------end")# 单个yield生成器进行遍历，注意yield的返回时机， 每次yield就会返回，下次从yield的位置开始继续执行
if __name__ == '__main__': # 使用test_yiled_for()

2.1.2 multi_yield

生成器函数包含多个yield (常见)

from loguru import loggerdef multi_yield_demo(a, b):caculate_sum = a + bprint(f"caculate_sum: {caculate_sum}")yield caculate_sum  # 第一次迭代返回，暂停caculate_sub = a - bprint(f"caculate_sub: {caculate_sub}")yield caculate_sub  # 第二次迭代返回，暂停caculate_multi = a * bprint(f"caculate_multi: {caculate_multi}")yield caculate_multi  # 第二次迭代返回，暂停# 如果一个函数中有多个yield, 那么每次迭代，yield 后面的表达式的值会返回给迭代器，并暂停，等待下一次迭代。 
# 下一次yield迭代会在上次暂停的地方继续执行，并返回值给迭代器。
if __name__ == '__main__': # 测试生成器gen = multi_yield_demo(5, 2)        # 获取得到一个生成器对象 class: generator# 方式1：用 next() 手动驱动print("next迭代：................................................")print(f"next迭代返回：{next(gen)}")  # 输出：执行到第一个 yield 前 → 第一个值print(f"next迭代返回：{next(gen)}")  # 输出：从第一个 yield 恢复，执行到第二个 yield 前 → 第二个值print(f"next迭代返回：{next(gen)}")  # 输出：从第二个 yield 恢复，执行到第三个 yield 前 → 第三个值try: # 无更多 yield，抛出 StopIterationZprint(f"next迭代返回：{next(gen)}")  # 输出：从第三个 yield 恢复，执行到第四个  except Exception as e: print(f"迭代异常：{type(e)}")# 方式2：用 for 循环自动迭代（推荐，无需处理 StopIteration）print("\nfor 循环迭代：................................................")for value in multi_yield_demo(5, 2):print(f"next迭代返回：{value}")

2.1.3 生成器 send用法

参考资料: https://www.runoob.com/python3/python3-iterator-generator.html

类似generator_obj.send(7)会先赋值到上一个yield退出的时候， 例如res = 7, 详细案例如下

def foo():print("starting...")while True:res = yield 4print("res:",res)# https://blog.csdn.net/mieleizhi0522/article/details/82142856
if __name__ == '__main__': generator_obj = foo()             # 这里返回一个生成器对象 不会调用函数  print(next(generator_obj))        # 返回第一个yield  start --> return 4  print("*" * 20)         # 分割.  print(generator_obj.send(7))      # 返回第二个yield  res = g.send(7), res = 7 在while循环中， 返回start就不会调用了 , 相当于先传值 然后next print("-" * 20 )print(next(generator_obj))        # 返回第三个yield  res, None --> return 4# 如果没有.send(7), 那么res， 默认就是None

2.1.4 next() 迭代生成器

1. 生成器函数

def count_up_to(n):i = 1while i <= n:yield ii += 1gen = count_up_to(3)
print(next(gen))  # 1
print(next(gen))  # 2
print(next(gen))  # 3
# next(gen) → StopIteration

• 函数遇到 yield 时暂停执行，保存状态（局部变量、指令位置等）；
• 下次调用 next() 时从 yield 后继续执行；
• 自动实现 __iter__() 和 __next__()，因此是迭代器。

2. 生成器表达式（类似列表推导式，但用 ()）

gen = (x**2 for x in range(3))
list(gen)  # [0, 1, 4]# 列表是可迭代对象，不是迭代器
def square_gen():for x in range(3):yield x**2  # 计算 x 的平方并返回# 等价于square_gen()
gen = (x**2 for x in range(3))      # 得到一个生成器对象for i in gen:print(i)square_g = square_gen()
print(list(square_g) )  # 第一遍遍历
print(list(square_g) )  # 第二遍为空

• 比列表推导式更省内存（不一次性生成所有元素）。
• 这行代码的目的是 演示生成器表达式的创建、惰性迭代特性，以及生成器 “一次性迭代” 的核心行为—— 第一次迭代获取所有结果，第二次迭代因生成器耗尽返回空列表。

---

2.2. 异步迭代器async for

2.2.1 single_yield

示例代码

import asyncio# 异步生成器（自动实现异步迭代器协议）
async def async_generator_single(start: int, end: int, delay: float = 0.2):current = startwhile current < end:await asyncio.sleep(delay)  # 模拟异步耗时操作yield current  # 异步返回值current += 1# 测试异步生成器
async def main():print(f"异步生成器迭代：")result_list = []async for num in async_generator_single(start=1, end=4):print(f"生成值：{num}")result_list.append(num)print(f"异步生成器迭代完成： {result_list}")# 生成器中包含多个yield
if __name__ == "__main__":asyncio.run(main())

2.2.2 multi_yield

示例代码

import asyncio# 异步生成器（自动实现异步迭代器协议）
async def async_generator_multi(start: int, end: int, delay: float = 0.2):current = startwhile current < end:await asyncio.sleep(delay)  # 模拟异步耗时操作yield current  # 异步返回值await asyncio.sleep(delay)  # 模拟异步耗时操作yield current + 1  # 异步返回值current += 1# 测试异步生成器
async def main():print(f"异步生成器迭代：")result_list = []async for num in async_generator_multi(start=1, end=4):print(f"生成值：{num}")result_list.append(num)print(f"异步生成器迭代完成： {result_list}")# 生成器中包含多个yield
if __name__ == "__main__":asyncio.run(main())

2.2.3 并行使用样例

1. 这里是将异步生成器和恰如任务并行

import asyncio
import random
from datetime import datetime# 异步生成器（同上）
async def async_random_generator(count: int, delay: float = 1.0):print(f"[{datetime.now().strftime('%H:%M:%S')}] 异步生成器启动：生成 {count} 个随机数")for i in range(count):await asyncio.sleep(delay)random_num = random.uniform(0, 100)yield f"[{datetime.now().strftime('%H:%M:%S')}] 第 {i+1} 个随机数：{random_num:.2f}"# 另一个异步任务：每秒打印当前时间
async def task_print_time():"""每秒打印一次当前时间，共打印 6 次"""for i in range(5):print(f"[{datetime.now().strftime('%H:%M:%S')}] 当前时间（并发任务1）.............task_print_time: {i}")await asyncio.sleep(1.0)async def consume_random_generator():async for res in async_random_generator(count=5):print(res)
# 主函数：并发运行两个异步任务
async def main():# 方式1：用 asyncio.gather 并发运行两个任务task1 = asyncio.create_task(task_print_time())                           # 并发任务1：打印时间task2 = asyncio.create_task(consume_random_generator())                  # 并发任务2：迭代异步生成器     # 在这之前task1, task2 这两个任务已经启动起来了， await只是确保等待他们完成await task1  # 等待任务1完成 确保任务完成await task2  # 等待任务2完成print("所有任务已完成")if __name__ == "__main__":asyncio.run(main())

2. 在异步生成器内部进行并行

import asyncio
import timeasync def async_task(item):await asyncio.sleep(1)return f"处理结果：{item}"# 异步生成器：批量并行处理任务
async def parallel_async_generator(items, batch_size=2):# 按批次拆分任务for i in range(0, len(items), batch_size):batch = items[i:i+batch_size]# 并行执行批次内的任务（关键：asyncio.gather）batch_results = await asyncio.gather(*[async_task(item) for item in batch])for res in batch_results:yield resasync def test_main():start_time = time.time()items = [1, 2, 3, 4]  # 4个任务，批次大小2async for res in parallel_async_generator(items):print(res)print(f"总耗时：{time.time() - start_time:.2f}秒")if __name__ == '__main__': asyncio.run(test_main())

三、关系总结

特性	迭代器	生成器
是否是对象	✅ 是	✅ 是（生成器是迭代器的子类）
创建方式	实现 __iter__ 和 __next__；或 iter()	用 yield 的函数 或 (… for …) 表达式
状态保存	需手动维护状态	自动保存函数执行上下文（栈帧）
编写复杂度	较高（需定义类）	极低（类似写普通函数）
是否可重复使用	❌ 一次性	❌ 一次性（除非重新调用生成器函数）

✅ 生成器是更轻量、更易用的迭代器实现方式。
🎯 实际开发中，优先使用生成器（除非需要精细控制或实现复杂迭代逻辑）。

---

四、实用场景（结合你的兴趣）

• WebSocket 数据流处理：可用 async for 遍历异步生成器（async def + yield），实现对实时消息流的惰性消费；
• 大数据处理：读取大文件时用生成器逐行返回，避免内存爆炸；
• 加密/解密流水线：构建生成器管道（如 decrypt → verify → parse），实现内存高效、可组合的数据处理。

4.1 读取文件

1. 如果文件很大可以一行一行读取不会爆内存

def read_txt_line_by_line(file_path: str, encoding: str = "utf-8"):"""同步生成器：一行一行读取 TXT 文件:param file_path: TXT 文件路径（相对/绝对路径）:param encoding: 文件编码（默认 utf-8，需根据实际文件调整，如 gbk）:yield: 文件中的每一行（保留原始换行符 \n）"""# with 语句自动处理文件打开/关闭，避免资源泄露with open(file_path, "r", encoding=encoding) as f:# 循环读取每一行（文件对象本身是可迭代对象，逐行返回）for line in f:yield line  # 暂停并返回当前行，下次迭代从下一行继续if __name__ == '__main__': # 读取文件（替换为你的 TXT 文件路径）, 确保你这个路径存在file_path = "test.txt"# 方式1：for 循环迭代（推荐，自动处理迭代终止）for line_num, line in enumerate(read_txt_line_by_line(file_path), start=1):print(f"第 {line_num} 行：{line}", end="")  # end="" 避免重复打印换行符