参考资料: 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.current
        self.current += 1
        return 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 = 1
    while i <= n:
        yield i
        i += 1

def 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 = 1
    while i <= n:
        print("yield ........ before")
        yield i
        print("yield ........ after")
        i += 1
    print("single_yield_generator 结束")

def test_yiled_for():
    for num in single_yield_generator(5):
        # print("--------------------------begin")
        print(num)  # 输出：1 2 3 4 5
        print("--------------------------end")

# 单个yield生成器进行遍历，注意yield的返回时机， 每次yield就会返回，下次从yield的位置开始继续执行
if __name__ == '__main__': 
    # 使用
    test_yiled_for()

2.1.2 multi_yield

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

from loguru import logger

def multi_yield_demo(a, b):
    caculate_sum = a + b
    print(f"caculate_sum: {caculate_sum}")
    yield caculate_sum  # 第一次迭代返回，暂停
    
    caculate_sub = a - b
    print(f"caculate_sub: {caculate_sub}")
    yield caculate_sub  # 第二次迭代返回，暂停
    
    caculate_multi = a * b
    print(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，抛出 StopIterationZ
        print(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 4
        print("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 = 1
    while i <= n:
        yield i
        i += 1

gen = 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 = start
    while 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 = start
    while 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 time

async 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 res

async def test_main():
    start_time = time.time()
    items = [1, 2, 3, 4]  # 4个任务，批次大小2
    async 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="" 避免重复打印换行符