深入 Python 迭代器协议：从 __iter__ 到 async for 的完全指南

引言

Python 中的迭代器（Iterator）是语言最优雅的设计之一。从 for x in iterable 这种看似简单的语法，到生成器表达式、协程、异步迭代器，迭代协议贯穿了整个 Python 生态。本文将深入剖析 Python 迭代器协议的底层机制，从最基础的 __iter__ 和 __next__ 魔法方法开始，逐步深入到生成器函数、生成器表达式、以及 Python 3.6+ 引入的异步迭代器协议。

1. 迭代器协议基础

Python 的迭代协议基于两个核心概念：可迭代对象（Iterable）和 迭代器（Iterator）。

可迭代对象：实现了 __iter__() 方法的对象。该方法返回一个迭代器。常见的可迭代对象包括 list、tuple、dict、set、str 等。

迭代器：同时实现了 __iter__() 和 __next__() 方法的对象。__iter__() 返回迭代器自身，__next__() 返回序列中的下一个元素，在没有更多元素时抛出 StopIteration 异常。

class CountDown:
    """一个简单的倒计时迭代器"""
    
    def __init__(self, start):
        self.current = start
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.current <= 0:
            raise StopIteration
        num = self.current
        self.current -= 1
        return num

# 使用
for n in CountDown(5):
    print(n, end=' ')  # 输出: 5 4 3 2 1

需要注意的是，迭代器是一次性的。一旦耗尽，就无法重置或重新开始：

cd = CountDown(3)
print(list(cd))  # [3, 2, 1]
print(list(cd))  # [] -- 迭代器已耗尽

2. 可迭代对象 vs 迭代器：关键区别

一个常见的误区是混淆可迭代对象和迭代器。实际上，它们是两个相关但不同的概念：

class RangeWrapper:
    """可迭代对象：每次调用 __iter__ 返回新的迭代器"""
    
    def __init__(self, start, end):
        self.start = start
        self.end = end
    
    def __iter__(self):
        # 返回一个独立的迭代器对象
        return RangeIterator(self.start, self.end)


class RangeIterator:
    """迭代器：维护遍历状态"""
    
    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
        value = self.current
        self.current += 1
        return value

# 可迭代对象可以被多次遍历
rw = RangeWrapper(1, 4)
print(list(rw))  # [1, 2, 3]
print(list(rw))  # [1, 2, 3] -- 再次迭代依然有效

这种分离设计的好处是：同一个可迭代对象可以同时被多个独立的迭代器遍历，互不影响。Standard 库中的文件对象就遵循这个模式 —— 可以多次打开同一个文件进行遍历。

3. 迭代工具函数

Python 的 itertools 模块提供了一套强大的迭代工具，是函数式编程风格在 Python 中的精华体现：

import itertools

# chain：将多个可迭代对象串联
for item in itertools.chain([1, 2], [3, 4], [5]):
    print(item, end=' ')  # 1 2 3 4 5

print()

# islice：迭代器切片（支持无限序列）
counter = itertools.count(10, 2)  # 10, 12, 14, ...
for val in itertools.islice(counter, 5):
    print(val, end=' ')  # 10 12 14 16 18

print()

# cycle：无限循环迭代
colors = itertools.cycle(['红', '绿', '蓝'])
for _ in range(7):
    print(next(colors), end=' ')  # 红 绿 蓝 红 绿 蓝 红

print()

# tee：从一个迭代器创建多个独立迭代器
it = iter(range(5))
a, b = itertools.tee(it, 2)
print(list(a))  # [0, 1, 2, 3, 4]
print(list(b))  # [0, 1, 2, 3, 4]

# groupby：相邻分组（需要先排序！）
data = [('A', 1), ('A', 2), ('B', 3), ('B', 4), ('A', 5)]
data.sort(key=lambda x: x[0])
for key, group in itertools.groupby(data, key=lambda x: x[0]):
    print(f'{key}: {list(group)}')
# A: [('A', 1), ('A', 2)]
# B: [('B', 3), ('B', 4)]
# A: [('A', 5)]

itertools 中最精妙的可能是无穷迭代器：count()、cycle()、repeat()。它们与 islice() 配合使用，让你在有限操作中处理无限的序列。

4. 生成器：语法糖下的迭代器工厂

生成器函数是 Python 中最具"Pythonic"特性的语法之一。它们是迭代器的语法糖 —— 任何包含 yield 关键字的函数都自动成为生成器函数。

def fibonacci():
    """斐波那契数列生成器（无限序列）"""
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# 取前 10 个斐波那契数
import itertools
print(list(itertools.islice(fibonacci(), 10)))
# [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

生成器函数的神奇之处在于：调用它时不会立即执行函数体，而是返回一个生成器对象。每次调用 next() 时，函数从上次 yield 的位置恢复执行，直到遇到下一个 yield 或函数结束。

4.1 send() 方法：双向通信

生成器不仅仅是"数据生产者" —— 通过 send() 方法，我们可以向生成器注入数据，实现生产者和消费者之间的双向通信：

def running_average():
    """计算运行平均值的协程式生成器"""
    total = 0.0
    count = 0
    average = None
    while True:
        value = yield average  # 产出平均值，同时接收外部输入
        if value is None:
            break
        total += value
        count += 1
        average = total / count

# 使用
avg = running_average()
next(avg)  # 预激生成器（执行到第一个 yield）
print(avg.send(10))   # 10.0
print(avg.send(20))   # 15.0
print(avg.send(30))   # 20.0
avg.close()

4.2 生成器表达式

生成器表达式是列表推导式的惰性版本，使用圆括号而非方括号：

# 列表推导式：立即计算，占用内存
squares_list = [x*x for x in range(1000000)]

# 生成器表达式：惰性计算，几乎不占用内存
squares_gen = (x*x for x in range(1000000))

# 生成器表达式可以链式组合
result = sum(x*x for x in range(1000000) if x % 2 == 0)

生成器表达式的另一个优势是可以组合：多个生成器表达式串联不会创建中间列表。

5. 生成器的高级模式

5.1 yield from：生成器委托

Python 3.3 引入的 yield from 语法允许一个生成器将部分操作委托给另一个生成器，是构建复杂迭代管道的基石：

def sub_generator():
    yield from range(10, 13)
    yield from "ab"

# 等价于:
# yield 10; yield 11; yield 12; yield 'a'; yield 'b'

print(list(sub_generator()))  # [10, 11, 12, 'a', 'b']

# 实际应用：扁平化嵌套列表
def flatten(nested):
    for item in nested:
        if isinstance(item, (list, tuple)):
            yield from flatten(item)  # 递归展开
        else:
            yield item

nested = [1, [2, [3, 4], 5], 6, [7, 8]]
print(list(flatten(nested)))  # [1, 2, 3, 4, 5, 6, 7, 8]

5.2 生成器作为协程

生成器配合 yield from 可以构建复杂的处理流水线，实现类协程的协作式多任务：

def data_producer(items, target):
    """生产者：生成数据并发送给下游"""
    for item in items:
        target.send(item)

def data_processor(factor, target):
    """中间处理器：处理数据后传递给下一个节点"""
    while True:
        value = (yield)
        result = value * factor
        target.send(result)

def data_consumer():
    """消费者：接收最终结果"""
    total = 0
    while True:
        value = (yield)
        total += value
        print(f"Consumer received: {value}, running total: {total}")

# 构建流水线
consumer = consumer()
next(consumer)  # 消费者预激
processor = processor(10, consumer)
next(processor)  # 处理器预激
data_producer([1, 2, 3, 4, 5], processor)
# Consumer received: 10, running total: 10
# Consumer received: 20, running total: 30
# Consumer received: 30, running total: 60
# Consumer received: 40, running total: 100
# Consumer received: 50, running total: 150

6. 异步迭代器协议

Python 3.6 引入的异步迭代器协议（PEP 525）让 for 循环也能在异步上下文中工作。它与同步迭代器协议类似，但使用不同的魔法方法：

协议	可迭代对象方法	迭代器方法	终止信号
同步	__iter__()	__next__()	StopIteration
异步	__aiter__()	__anext__()	StopAsyncIteration

import asyncio

class AsyncCounter:
    """异步迭代器示例"""
    
    def __init__(self, start, stop, delay=0.1):
        self.current = start
        self.stop = stop
        self.delay = delay
    
    def __aiter__(self):
        self.current = start  # 重置计数器
        return self
    
    async def __anext__(self):
        if self.current >= self.stop:
            raise StopAsyncIteration
        await asyncio.sleep(self.delay)
        value = self.current
        self.current += 1
        return value

async def main():
    async for num in AsyncCounter(5, delay=0.3):
        print(f"异步计数: {num}")

# asyncio.run(main())

6.1 异步生成器

Python 3.6 还引入了异步生成器（PEP 525），在 async def 函数中使用 yield：

import asyncio

async def async_range(start, stop, delay=0.1):
    """异步生成器：异步产生序列值"""
    for i in range(start, stop):
        await asyncio.sleep(delay)
        yield i

async def main():
    async for val in async_range(0, 5, delay=0.2):
        print(f"Received: {val}")

# asyncio.run(main())

6.2 异步生成器的实际应用

异步生成器在处理流式数据（如 WebSocket 消息、数据库游标、API 分页）时非常有用：

async def paginated_fetch(api_url, page_size=100):
    """逐页异步获取 API 数据，使用异步生成器 yield 每条记录"""
    page = 1
    while True:
        batch = await fetch_page(api_url, page, page_size)
        if not batch:
            break
        for record in batch:
            yield record
        page += 1

async def process_user_stream():
    """处理用户事件流"""
    count = 0
    async for event in paginated_fetch("https://api.example.com/users"):
        count += 1
        if count > 1000:  # 限制处理数量
            break
        await handle_event(event)
    print(f"Processed {count} events")

7. 迭代器性能考量

虽然生成器的内存效率无可匹敌，但它并非没有代价：

对比维度	列表	迭代器/生成器
内存占用	O(n) -- 全部存储	O(1) -- 仅当前值
单次访问速度	快（CPU 缓存友好）	稍慢（函数调用开销）
可重复遍历	是	否（一次性）
随机访问	O(1)	不支持
适用场景	小数据集、多次访问	大数据流、流水线处理

import sys
import time

# 内存对比
list_comp = [x for x in range(10000)]
gen_exp = (x for x in range(10000))

print(f"列表大小: {sys.getsizeof(list_comp)} bytes")  # ~85KB
print(f"生成器大小: {sys.getsizeof(gen_exp)} bytes")  # ~112 bytes（恒定！）

# 速度对比
def time_list():
    start = time.perf_counter()
    total = sum([x*x for x in range(100000)])
    return time.perf_counter() - start

def time_gen():
    start = time.perf_counter()
    total = sum(x*x for x in range(100000))
    return time.perf_counter() - start

print(f"列表推导式: {time_list():.4f}s")
print(f"生成器表达式: {time_gen():.4f}s")

在大多数场景下，列表推导式由于 CPU 缓存局部性优势，速度更快。但当数据量巨大或内存是硬约束时，生成器的 O(1) 内存特性无可替代。

8. 实践模式：构建数据处理管道

迭代器的真正威力在于构建声明式的数据处理管道。以下是一个实用的日志分析器示例：

import re
from datetime import datetime
from collections import Counter
from pathlib import Path

def read_log_lines(filepath):
    """逐行读取日志文件（不加载整个文件到内存）"""
    with open(filepath, 'r', encoding='utf-8', errors='replace') as f:
        for line in f:
            line = line.strip()
            if line:
                yield line

def parse_log_format(lines):
    """解析日志行，提取结构化信息"""
    pattern = re.compile(
        r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) \[(\w+)\] (.*)'
    )
    for line in lines:
        match = pattern.match(line)
        if match:
            timestamp, level, message = match.groups()
            try:
                yield {
                    'timestamp': datetime.fromisoformat(timestamp),
                    'level': level,
                    'message': message,
                }
            except ValueError:
                continue

def filter_by_level(records, min_level='WARNING'):
    """按日志级别过滤"""
    levels = {'DEBUG': 0, 'INFO': 1, 'WARNING': 2, 'ERROR': 3, 'CRITICAL': 4}
    min_val = levels.get(min_level, 2)
    for record in records:
        if levels.get(record['level'], 0) >= min_val:
            yield record

def summarize_errors(records):
    """统计错误摘要"""
    return Counter(r['level'] for r in records)

# 使用管道处理日志
pipeline = filter_by_level(
    parse_log_format(read_log_lines('app.log')),
    min_level='ERROR'
)
error_stats = summarize_errors(pipeline)
print("Error summary:", error_stats)

这种模式的优势在于：

• 内存高效：无论日志文件多大，内存使用保持恒定
• 声明式表达：代码清晰描述"做什么"而非"怎么做"
• 可组合：每个阶段独立，可以轻松插入新过滤器
• 惰性求值：只有消费时才读取和处理数据

9. Python 迭代器的内部机制

了解迭代器的内部实现有助于写出更高效的代码。当 Python 执行 for x in iterable 时，实际发生的是：

# 以下代码：
for item in my_list:
    print(item)

# 等价于：
_iterator = iter(my_list)  # 调用 my_list.__iter__()
while True:
    try:
        item = next(_iterator)  # 调用 _iterator.__next__()
        print(item)
    except StopIteration:
        break

Python 的 for 循环本质上是将迭代器协议内建化了。任何实现了迭代协议的对象都可以直接在 for 循环中使用。

对于生成器对象，yield 实际上是使用了 Python 的 帧对象（Frame Object）机制来实现挂起和恢复。生成器被调用时创建帧对象，执行到 yield 时将帧的状态（局部变量、指令指针）保存起来，下次调用 next() 时恢复执行。

def inspect_generator():
    """检查生成器的内部状态"""
    gen = fibonacci()
    print(f"Generator frame: {gen.gi_frame}")
    print(f"Generator running: {gen.gi_running}")
    print(f"Last yielded: {None}")  # 尚未开始
    
    next(gen)
    next(gen)
    print(f"After 2 next() calls:")
    print(f"  Local vars: {gen.gi_frame.f_locals if gen.gi_frame else 'finished'}")
    
    gen.close()
    print(f"After close(), frame: {gen.gi_frame}")  # None（已销毁）

10. 总结与最佳实践

Python 的迭代器协议虽然简单（只有两个魔法方法），但它支撑了整个 Python 的数据处理生态：

• 优先使用生成器处理大数据集，节省内存
• 善用 itertools 构建高效流水线
• 用 yield from 简化嵌套生成器的委托
• 异步生成器配合 async for 处理 IO 密集型流式数据
• 迭代器是一次性的 —— 需要多次遍历时使用可迭代对象而非迭代器
• 生成器表达式比列表推导式更省内存，但速度稍慢
• yield from 不仅是语法糖，还能自动传递 send() 和 throw()
• 在 Python 3.10+ 中，enumerate()、zip() 等内置函数返回的迭代器性能更好

理解迭代器协议不仅仅是掌握一个语法特性，更是理解 Python "惰性求值"哲学的关键 —— 它让你的代码更节省内存、更富有表达力。当你需要处理海量数据流、构建事件驱动系统，或优雅地遍历复杂结构时，迭代器始终是你最强大的工具之一。