迭代器协议与自定义迭代器

Python进阶编程专题 · 创建可迭代对象和自定义迭代器

专题：Python进阶编程系统学习

关键词：Python, 迭代器, 迭代器协议, __iter__, __next__, Iterable, Iterator, 惰性求值

一、迭代器协议概述

迭代器协议（Iterator Protocol）是Python中用于遍历数据容器的核心机制，它定义了一套标准的接口规范，使得任何遵循此协议的对象都可以被迭代。在Python中，迭代器协议由两个特殊方法构成：__iter__ 和 __next__。理解这两个方法是掌握Python迭代体系的关键前提。

迭代器协议的思想精髓在于"惰性求值"——数据不是一次性全部加载到内存中，而是按需逐个生成。这种设计哲学使得Python能够高效处理大规模甚至无限的数据序列，而不会因为内存耗尽而崩溃。从语言设计的角度来看，迭代器协议将"遍历"这一行为进行了抽象，统一了列表、字典、文件、数据库游标等不同数据源的遍历方式。

在Python编程实践中，我们几乎每天都在使用迭代器，即使没有意识到这一点。当你编写 for x in my_list 时，底层就在调用迭代器协议。当你使用列表推导式、生成器表达式、map()、filter() 等函数式工具时，迭代器协议同样在背后默默工作。

核心定义：迭代器协议要求对象同时实现 __iter__() 和 __next__() 两个方法。__iter__() 返回迭代器自身，__next__() 返回序列中的下一个元素，当没有更多元素时抛出 StopIteration 异常。

协议的双方法定义

迭代器协议的完整定义包含两个方法，每个方法承担不同的职责：

__iter__(self)：返回一个迭代器对象。对于容器类，这通常意味着返回一个新的迭代器实例；对于迭代器类，通常返回 self。这个方法使得对象可以在 for 循环中被使用，也可以传递给 iter() 内置函数。
__next__(self)：返回序列中的下一个可用元素。如果没有更多元素可供返回，必须抛出 StopIteration 异常来通知调用方迭代已经结束。这个方法可以通过 next() 内置函数手动调用。

下面的代码演示了迭代器协议最基础的使用方式，通过手动调用 iter() 和 next() 来模拟迭代过程：

# 基础迭代器协议演示
numbers = [10, 20, 30]

# 获取迭代器对象
iterator = iter(numbers)  # 等价于 numbers.__iter__()

# 手动驱动迭代过程
print(next(iterator))     # 输出: 10,  等价于 iterator.__next__()
print(next(iterator))     # 输出: 20
print(next(iterator))     # 输出: 30

# 第四次调用将抛出 StopIteration
try:
    next(iterator)
except StopIteration:
    print("迭代已经结束！")  # 输出: 迭代已经结束！

关键洞察：iter() 和 next() 是Python提供的两个内置函数，它们分别作为 __iter__() 和 __next__() 的"语法糖"封装。使用内置函数更为规范，因为它们在调用前会进行一些额外的类型检查和处理。

二、Iterable vs Iterator 深入辨析

在Python的迭代生态中，可迭代对象（Iterable）和迭代器（Iterator）是两个容易混淆但本质不同的概念。理解二者的区别是掌握迭代器协议的重要前提。

可迭代对象（Iterable）

可迭代对象是指实现了 __iter__() 方法的对象，或者实现了 __getitem__() 方法且能够从索引0开始依次返回元素的对象。可迭代对象是"可以被迭代"的对象，但它本身不维护迭代状态。常见的可迭代对象包括：列表、元组、字典、集合、字符串、文件对象等。

可迭代对象的核心特征是：它可以被传递给 iter() 函数并返回一个迭代器。同一个可迭代对象可以被多次迭代，每次都会创建一个新的迭代器，从第一个元素重新开始。这是因为可迭代对象本身不保存迭代的进度状态。

# 可迭代对象可以反复迭代
my_list = [1, 2, 3, 4, 5]

for item in my_list:
    print(item, end=' ')  # 输出: 1 2 3 4 5
print()

# 第二次迭代，从头开始
for item in my_list:
    print(item, end=' ')  # 输出: 1 2 3 4 5
print()

# 判断是否为可迭代对象
from collections.abc import Iterable
print(isinstance([1, 2, 3], Iterable))   # True
print(isinstance("abc", Iterable))       # True
print(isinstance(42, Iterable))          # False

迭代器（Iterator）

迭代器是同时实现了 __iter__() 和 __next__() 方法的对象。迭代器是一个"有状态"的对象——它会记录当前迭代的位置，每次调用 __next__() 都会前进一位并返回该位置的元素。当迭代器被消耗完毕后，再次调用 __next__() 将抛出 StopIteration 异常。

迭代器的核心特征是"一次性消费"：一旦迭代器产生了所有元素，它就处于"耗尽"状态，无法重置。如果需要重新遍历，必须重新创建一个新的迭代器。这在处理网络流、传感器数据、文件句柄等不可回退的数据源时尤为明显。

# 迭代器一次性消费的演示
my_list = [1, 2, 3]
iterator = iter(my_list)

# 正常迭代
print(list(iterator))    # 输出: [1, 2, 3]

# 迭代器已经耗尽，再次转换为列表是空的
print(list(iterator))    # 输出: []

# 需要重新创建迭代器
iterator = iter(my_list)
print(list(iterator))    # 输出: [1, 2, 3]

# 判断是否为迭代器
from collections.abc import Iterator
print(isinstance(iter([1, 2, 3]), Iterator))  # True
print(isinstance([1, 2, 3], Iterator))       # False
print(isinstance(open('example.txt'), Iterator))  # True (文件是迭代器)

核心差异对比

对比维度	Iterable（可迭代对象）	Iterator（迭代器）
实现方法	`__iter__()` 或 `__getitem__()`	`__iter__()` + `__next__()`
状态管理	无状态，不记录迭代位置	有状态，记录当前位置
消费特征	可重复使用，每次创建新的迭代器	一次性使用，耗尽后不可重置
内存使用	通常存储全部数据	惰性求值，按需计算
`iter()` 调用结果	返回一个新的迭代器	返回自身（`self`）
典型例子	list, dict, tuple, str, set	generator, file object, `map()` 返回值

记忆技巧：每次 for 循环开始时，Python都会调用 iter() 来获取迭代器。如果传入的是可迭代对象，就得到一个"新鲜"的迭代器；如果传入的已经是迭代器，就得到它自身。这就是为什么迭代器可以在for循环中使用，但只能遍历一次。

检查协议的鸭子类型方式

Python推崇"鸭子类型"（Duck Typing）——"如果它走起路来像鸭子，叫起来像鸭子，那它就是鸭子"。在迭代器上下文中，我们不应依赖类型检查，而应通过尝试使用来判断：

# 鸭子类型方式检查迭代能力
def safe_iterate(obj):
    """安全地获取对象的迭代器"""
    try:
        iterator = iter(obj)
        # 尝试获取第一个元素验证迭代器有效
        first = next(iterator)
        return True, first
    except TypeError:
        return False, None
    except StopIteration:
        return True, None  # 空迭代器也是有效的

# 测试不同对象
print(safe_iterate([1, 2, 3]))    # (True, 1)
print(safe_iterate("abc"))        # (True, 'a')
print(safe_iterate(42))           # (False, None)
print(safe_iterate([]))           # (True, None) 空列表

三、for循环的底层机制

Python的 for 循环是迭代器协议最广泛的应用场景。理解 for 循环的底层实现机制，有助于我们写出更高效的代码，并且能在需要时用更底层的迭代控制来替代 for 循环。

很多人可能认为 for item in iterable 是一种简单的遍历语法，但实际上Python解释器会将它展开为一个包含 try/except StopIteration 的 while 循环。下面是用 while 循环模拟 for 循环内部机制的完整代码：

# 模拟 for 循环的底层执行机制
def for_loop_simulation(iterable):
    """
    展示 Python for 循环的内部工作流程
    适用于任何可迭代对象
    """
    # 步骤1: 调用 iter() 获取迭代器
    try:
        iterator = iter(iterable)
    except TypeError:
        raise TypeError(f"'{type(iterable).__name__}' object is not iterable")

    # 步骤2: 循环调用 next() 直到 StopIteration
    while True:
        try:
            item = next(iterator)
        except StopIteration:
            break  # 迭代正常结束
        else:
            # 执行循环体（这里用打印模拟）
            print(f"处理元素: {item}")

# 这个模拟对于自定义的可迭代对象也适用
for_loop_simulation([1, 2, 3])

for 循环背后的完整工作流

获取迭代器：Python首先调用 iter(iterable) 获取一个迭代器对象。如果 iterable 实现了 __iter__() 方法，则调用该方法；如果实现了 __getitem__() 方法，Python会创建一个旧的风格迭代器，从索引0开始依次访问。
循环调用 next：在每次迭代中，Python调用 next(iterator) 获取下一个元素。这个调用被隐藏在 for 循环的机制中，用户看不到。
捕获 StopIteration：当迭代器没有更多元素时，__next__() 抛出 StopIteration 异常。for 循环静默地捕获这个异常并退出循环。这正是为什么 for 循环不会因为迭代耗尽而崩溃的原因。
else 分支：如果 for 循环正常结束（没有被 break 中断），可选的 else 子句会被执行。

# 利用 else 分支的 for 循环完整示例
def find_first_even(numbers):
    """查找列表中的第一个偶数"""
    for num in numbers:
        if num % 2 == 0:
            print(f"找到偶数: {num}")
            break
    else:
        print("列表中没有偶数！")

find_first_even([1, 3, 5, 7])    # 输出: 列表中没有偶数！
find_first_even([1, 2, 5, 7])    # 输出: 找到偶数: 2

可迭代对象的回退机制：getitem

Python提供了一种向后兼容的旧式迭代协议：如果一个对象没有实现 __iter__() 方法，但实现了 __getitem__() 方法，Python会创建一个旧的迭代器，从索引0开始依次调用 __getitem__(0)、__getitem__(1)……直到抛出 IndexError 为止。

# 利用 __getitem__ 实现旧式可迭代对象
class OldStyleIterable:
    """仅实现 __getitem__，不实现 __iter__ 的可迭代对象"""
    def __init__(self, data):
        self.data = data

    def __getitem__(self, index):
        """
        支持通过索引访问元素。
        Python 的 iter() 会从索引0开始递增调用此方法，
        直到 IndexError 被抛出，迭代结束。
        """
        if index < len(self.data):
            return self.data[index]
        raise IndexError("超出范围")

# 测试旧式迭代
obj = OldStyleIterable(["A", "B", "C"])
for item in obj:
    print(item, end=' ')   # 输出: A B C
print()

# iter() 和 next() 也能正常工作
it = iter(obj)
print(next(it))  # A
print(next(it))  # B
print(next(it))  # C

注意事项：__getitem__ 基于索引的回退机制是Python 2时代的产物，主要用于向后兼容。在现代Python编程中，应优先实现 __iter__() 方法，因为它更明确、性能更好，且能够更好地与其他迭代工具（如 itertools 模块）协同工作。

四、自定义迭代器完整实现

掌握了迭代器协议的理论基础后，我们来学习如何从头实现自定义迭代器。自定义迭代器在实际项目中有着广泛的应用场景，比如遍历树结构、生成特定序列、读取分页API、流式处理大文件等。

基础自定义迭代器

下面是一个基础范围迭代器的实现，它模仿了内置 range() 的行为，但使用迭代器协议手动实现：

class MyRange:
    """
    自定义范围迭代器，模拟内置 range() 的迭代行为。
    演示了迭代器协议的标准实现模式。
    """
    def __init__(self, start, stop=None, step=1):
        # 处理不同参数调用方式
        if stop is None:
            self.start = 0
            self.stop = start
        else:
            self.start = start
            self.stop = stop

        self.step = step
        self.current = self.start

    def __iter__(self):
        """返回迭代器自身（符合迭代器协议）"""
        return self

    def __next__(self):
        """返回下一个元素，或抛出 StopIteration"""
        # 检查是否超出范围
        if (self.step > 0 and self.current >= self.stop) or \
           (self.step < 0 and self.current <= self.stop):
            raise StopIteration

        value = self.current
        self.current += self.step
        return value

# 使用自定义迭代器
for i in MyRange(5):
    print(i, end=' ')          # 输出: 0 1 2 3 4
print()

for i in MyRange(2, 10, 2):
    print(i, end=' ')          # 输出: 2 4 6 8
print()

# 手动驱动迭代
my_range = MyRange(3)
print(next(my_range))          # 0
print(next(my_range))          # 1
print(next(my_range))          # 2
# print(next(my_range))        # StopIteration!

可迭代与迭代器分离模式

高级的做法是将"可迭代对象"和"迭代器"分开为两个独立的类。这样做的好处是：可迭代对象可以反复生成新的迭代器，每次迭代从初始状态开始。这是Python内置容器（如列表、元组）采用的设计模式。

class Playlist:
    """
    可迭代对象 — 代表一个播放列表。
    每次迭代都创建一个新的 PlaylistIterator，
    因此可以反复遍历。
    """
    def __init__(self, songs):
        self.songs = songs

    def __iter__(self):
        """每次调用 iter() 都返回一个全新的迭代器"""
        return PlaylistIterator(self.songs)


class PlaylistIterator:
    """
    迭代器 — 负责遍历播放列表。
    在一次迭代过程中维护当前播放位置的状态。
    """
    def __init__(self, songs):
        self.songs = songs
        self.index = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.index >= len(self.songs):
            raise StopIteration

        song = self.songs[self.index]
        self.index += 1
        return f"正在播放: {song}"


# 可重复遍历演示
playlist = Playlist(["青花瓷", "七里香", "简单爱"])

print("=== 第一轮播放 ===")
for song in playlist:
    print(song)

print("\n=== 第二轮播放（从头开始）===")
for song in playlist:
    print(song)

设计模式最佳实践：将可迭代对象和迭代器分离是实现"可重复迭代"的关键。可迭代对象（如 Playlist）负责管理数据但不维护迭代状态；迭代器（如 PlaylistIterator）负责维护迭代状态，但设计为一次性使用。这种"关注点分离"的设计模式使得代码更清晰、更健壮。

树结构遍历迭代器

迭代器协议的一个典型应用场景是遍历树形数据结构。下面我们实现一个二叉树的前序遍历迭代器，展示迭代器在复杂数据结构中的应用：

class TreeNode:
    """二叉树节点"""
    def __init__(self, value, left=None, right=None):
        self.value = value
        self.left = left
        self.right = right


class BinaryTreeIterator:
    """
    二叉树前序遍历迭代器。
    使用栈来模拟递归遍历，避免递归深度限制。
    """
    def __init__(self, root):
        self.stack = []
        self._push_left(root)

    def _push_left(self, node):
        """将节点及其所有左子节点入栈"""
        current = node
        while current is not None:
            self.stack.append(current)
            current = current.left

    def __iter__(self):
        return self

    def __next__(self):
        if not self.stack:
            raise StopIteration

        node = self.stack.pop()
        # 当前节点的右子节点入栈
        self._push_left(node.right)
        return node.value


# 构建二叉树
#       1
#      / \
#     2   3
#    / \
#   4   5
root = TreeNode(1)
root.left = TreeNode(2, TreeNode(4), TreeNode(5))
root.right = TreeNode(3)

# 遍历二叉树
for value in BinaryTreeIterator(root):
    print(value, end=' ')  # 输出: 4 2 5 1 3  (中序遍历顺序)
print()

五、反向迭代器与 reversed

除了正向迭代，Python还提供了反向迭代的能力，通过 __reversed__() 方法实现。当对象定义了 __reversed__() 方法时，内置函数 reversed() 会调用它，返回一个反向迭代器。

如果对象没有实现 __reversed__() 但实现了 __len__() 和 __getitem__()，Python可以自动推导出反向迭代的方式。但对于自定义迭代器，显式实现 __reversed__() 可以显著提升效率和灵活性。

class CountDown:
    """倒计时迭代器：既支持正向也支持反向迭代"""
    def __init__(self, start, end=0):
        self.start = start
        self.end = end

    def __iter__(self):
        """正向迭代：从 start 到 end"""
        current = self.start
        while current >= self.end:
            yield current
            current -= 1

    def __reversed__(self):
        """反向迭代：从 end 到 start"""
        current = self.end
        while current <= self.start:
            yield current
            current += 1


# 正向迭代
cd = CountDown(5, 2)
print("正向:", list(cd))      # 输出: 正向: [5, 4, 3, 2]

# 反向迭代
cd2 = CountDown(5, 2)
print("反向:", list(reversed(cd2)))  # 输出: 反向: [2, 3, 4, 5]

# 注意：每次迭代需要创建新实例，因为生成器是一次性的

实现建议：如果你的类有明确的"正向"和"反向"遍历语义，同时实现 __iter__() 和 __reversed__() 是最佳实践。这样不仅支持 reversed() 内置函数，还让类的迭代行为更加完整和符合直觉。

自定义序列的反向迭代

对于实现了 __len__() 和 __getitem__() 的类，Python自动支持 reversed()，无需显式定义 __reversed__()：

class PaginatedData:
    """模拟分页数据的自定义序列"""
    def __init__(self, items, page_size=10):
        self.items = items
        self.page_size = page_size

    def __len__(self):
        return len(self.items)

    def __getitem__(self, index):
        """支持索引访问，使reversed()可以自动工作"""
        return self.items[index]

    def __iter__(self):
        """逐页返回数据"""
        for i in range(0, len(self.items), self.page_size):
            page = self.items[i:i + self.page_size]
            yield f"第{i//self.page_size + 1}页: {page}"


# 由于实现了 __len__ 和 __getitem__，reversed() 自动生效
data = PaginatedData(list(range(1, 26)), 5)

print("正向遍历:")
for item in data:
    print(item)

print("\n反向遍历 (自动支持):")
for item in reversed(data):
    print(item)

性能提示：自动推导的反向迭代（通过 __getitem__ 和 __len__）通常效率较低，因为它需要从最后一个索引开始依次向前访问。如果数据量较大，手动实现 __reversed__() 可以提供更高效的反向迭代方式。

六、惰性求值与无穷迭代器

迭代器协议最强大的特性之一就是惰性求值（Lazy Evaluation）——元素只有在被请求时才产生，而不是提前全部计算并存储在内存中。惰性求值使得我们可以处理概念上无限的数据集，以及实现按需加载的数据管道。

无穷迭代器

无穷迭代器永远不会抛出 StopIteration 异常，它可以无限的生成元素。在实际使用中，必须通过 break 条件或 itertools.islice() 来限制迭代次数，否则会无限循环。

class FibonacciInfinite:
    """无穷斐波那契数列迭代器 — 惰性求值的典型代表"""
    def __init__(self):
        self.a = 0
        self.b = 1

    def __iter__(self):
        return self

    def __next__(self):
        # 计算当前值并更新状态
        value = self.a
        self.a, self.b = self.b, self.a + self.b
        return value


# 使用 islice 限制迭代次数
from itertools import islice

fib = FibonacciInfinite()
print("前10个斐波那契数:", list(islice(fib, 10)))
# 输出: [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

# 继续使用同一个迭代器（状态持续）
print("接下来5个:", list(islice(fib, 5)))
# 输出: [55, 89, 144, 233, 377]

# !!! 无限循环警告：不要直接迭代无穷迭代器
# for x in FibonacciInfinite():
#     print(x)  # 会一直运行下去！

按需加载迭代器

惰性求值的另一大应用是按需加载（Lazy Loading）。当数据源较大时（如数据库查询结果、远程API分页数据、大文件），使用惰性迭代器可以避免一次性将所有数据加载到内存中：

class LazyFileReader:
    """
    惰性文件读取器 — 按需逐行读取，不一次性加载整个文件。
    即使文件有几十GB，内存消耗也始终保持极低水平。
    """
    def __init__(self, filename, chunk_size=1024):
        self.filename = filename
        self.chunk_size = chunk_size
        self.file = None

    def __iter__(self):
        return self

    def __next__(self):
        if self.file is None:
            self.file = open(self.filename, 'r', encoding='utf-8')

        line = self.file.readline()
        if not line:
            self.file.close()
            self.file = None
            raise StopIteration

        return line.rstrip('\n')

    def __del__(self):
        """确保文件在迭代器被销毁时关闭"""
        if self.file is not None:
            self.file.close()

    def close(self):
        """显式关闭文件"""
        if self.file is not None:
            self.file.close()
            self.file = None


# 使用上下文管理器模式增强安全性
class LazyFileReaderSafe(LazyFileReader):
    """支持上下文管理的版本"""
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()
        return False  # 不抑制异常

链式惰性管道

惰性迭代器的真正威力在于可以链式组合，形成数据处理管道。每个环节只在需要时才处理数据，这类似于Unix管道命令的工作方式：

class TransformIterator:
    """对上游迭代器的每个元素应用转换函数"""
    def __init__(self, iterator, transform_func):
        self.iterator = iterator
        self.transform = transform_func

    def __iter__(self):
        return self

    def __next__(self):
        return self.transform(next(self.iterator))


class FilterIterator:
    """过滤上游迭代器中符合条件的元素"""
    def __init__(self, iterator, predicate):
        self.iterator = iterator
        self.predicate = predicate

    def __iter__(self):
        return self

    def __next__(self):
        while True:
            item = next(self.iterator)
            if self.predicate(item):
                return item


# 构建数据处理管道
def number_generator():
    """基础数据源"""
    i = 0
    while True:
        yield i
        i += 1

# 管道: 生成数字 → 过滤出偶数 → 平方
base = number_generator()
even_filter = FilterIterator(base, lambda x: x % 2 == 0)
squares = TransformIterator(even_filter, lambda x: x ** 2)

# 惰性执行 — 只计算需要的值
from itertools import islice
result = list(islice(squares, 8))
print(f"前8个偶数的平方: {result}")
# 输出: [0, 4, 16, 36, 64, 100, 144, 196]

# 注意: 即使在管道末尾，数据也只有在被消费时才计算
print("惰性管道已就绪，但尚未执行任何计算")

惰性求值的内存优势：传统的"先全部计算再存储"方式可能消耗大量内存。如上例中的偶数平方计算，如果使用列表推导 [x**2 for x in range(100000000) if x % 2 == 0] 会立即耗尽内存。而惰性迭代器方案无论数据多大，内存占用始终是常数级别。

七、迭代器的状态管理与生命周期

迭代器的核心是状态管理——它必须精确记录当前迭代位置，并在正确的时机抛出 StopIteration。不当的状态管理是自定义迭代器最常见的bug来源。本节深入讨论迭代器状态管理的各个方面。

状态重置与一次性消费

迭代器的"一次性消费"特性是由其设计决定的。一旦迭代器抛出 StopIteration，它就应该一直处于"耗尽"状态。但有时我们需要提供重置能力，这需要通过重新创建迭代器来实现，而不是修改已耗尽迭代器的内部状态。

class ResettableIterable:
    """
    可重置迭代的容器类。
    不直接重置迭代器，而是通过重新创建迭代器来实现重置。
    """
    def __init__(self, data):
        self.data = data

    def __iter__(self):
        """每次都返回一个新的迭代器"""
        return self._Iterator(self.data)

    class _Iterator:
        """私有的迭代器实现类"""
        def __init__(self, data):
            self.data = data
            self.index = 0

        def __iter__(self):
            return self

        def __next__(self):
            if self.index >= len(self.data):
                raise StopIteration
            value = self.data[self.index]
            self.index += 1
            return value


# 验证可重复迭代
container = ResettableIterable([10, 20, 30])

# 第一轮
for x in container:
    print(x, end=' ')    # 10 20 30
print()

# 第二轮 — 完全从头开始
for x in container:
    print(x, end=' ')    # 10 20 30
print()

迭代器的"中毒"问题

一个常见的陷阱是迭代器在迭代过程中被部分消费后又被传入其他代码，导致意想不到的行为。这种状态被称为"迭代器中毒"：

# 迭代器中毒演示
def process_first_two(iterator):
    """处理前两个元素（但不消费更多）"""
    return [next(iterator), next(iterator)]

def process_remaining(iterator):
    """处理剩余元素"""
    return list(iterator)

numbers = [1, 2, 3, 4, 5, 6]
it = iter(numbers)

first_two = process_first_two(it)
print(f"前两个: {first_two}")   # [1, 2]

# 迭代器已经前进到第3个位置
remaining = process_remaining(it)
print(f"剩余: {remaining}")      # [3, 4, 5, 6]

# ======= 意外的陷阱 =======
data = [1, 2, 3, 4, 5, 6]
# 错误：将列表推导式的迭代器泄露给外部
it = iter(x for x in data)
first = next(it)                  # 消费了第一个元素
# ... 更多操作 ...
result = list(it)                 # 只得到了 [2, 3, 4, 5, 6]
print(f"泄露后的结果: {result}")

# 应对策略：使用 list() 立即固化
data_snapshot = list(data)        # 安全快照，可反复使用

多线程环境下的状态安全

在多线程环境中，共享迭代器的状态访问可能导致数据竞争和不可预期的结果。迭代器的 __next__() 调用不是原子操作，多个线程同时调用可能导致元素重复、丢失或索引越界。

import threading

class SafeIterator:
    """
    线程安全的迭代器 — 使用锁保护状态更新。
    适用于多线程环境下的共享迭代。
    """
    def __init__(self, data):
        self.data = data
        self.index = 0
        self.lock = threading.Lock()

    def __iter__(self):
        return self

    def __next__(self):
        with self.lock:
            if self.index >= len(self.data):
                raise StopIteration
            value = self.data[self.index]
            self.index += 1
        return value


# 线程安全测试
def worker(iterator, results, thread_id):
    """工作线程：从共享迭代器中获取元素"""
    try:
        while True:
            item = next(iterator)
            results.append((thread_id, item))
    except StopIteration:
        pass

safe_it = SafeIterator(list(range(100)))
results = []
threads = [
    threading.Thread(target=worker, args=(safe_it, results, i))
    for i in range(4)
]

for t in threads:
    t.start()
for t in threads:
    t.join()

# 验证：所有100个元素都被处理，且没有重复或遗漏
print(f"总处理元素数: {len(results)}")        # 100
print(f"无重复: {len(set(v for _, v in results)) == 100}")
# 注意: 由于线程调度，元素分配顺序可能不定

重要提醒：大多数情况下，不要在多个线程之间共享同一个迭代器。更安全的设计是"每个线程获取自己的独立迭代器"。如果确实需要共享，务必使用 threading.Lock 或 queue.Queue 等同步机制保护状态变更。

八、迭代器 vs 生成器深度对比

在Python中，生成器（Generator）通常被描述为"简化版的迭代器"，但这句话背后有着更丰富的内涵。生成器函数和生成器表达式是创建迭代器的语法糖，但两者之间存在重要的差异和各自的适用场景。

生成器是迭代器的语法糖

任何生成器都是迭代器（反之不成立）。生成器自动实现了 __iter__() 和 __next__() 方法，并自动管理 StopIteration 异常。最核心的区别在于：手写迭代器需要显式维护状态（如索引变量），而生成器利用函数的执行暂停和恢复来隐式管理状态。

# 对比：手写迭代器 vs 生成器

# 方式一：手写迭代器（显式状态管理）
class HandmadeCounter:
    """手写的计数器迭代器"""
    def __init__(self, limit):
        self.limit = limit
        self.current = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.current >= self.limit:
            raise StopIteration
        value = self.current
        self.current += 1
        return value

# 方式二：生成器函数（隐式状态管理）
def generator_counter(limit):
    """生成器版本的计数器"""
    current = 0
    while current < limit:
        yield current
        current += 1

# 两者行为完全一致
hc = HandmadeCounter(5)
gc = generator_counter(5)

print("手写迭代器:", list(hc))      # [0, 1, 2, 3, 4]
print("生成器:", list(gc))          # [0, 1, 2, 3, 4]

# 验证：生成器即是迭代器
from collections.abc import Iterator
print(isinstance(generator_counter(3), Iterator))  # True

核心差异对比表

对比维度	手写迭代器（class）	生成器（yield）
代码量	较多（需实现协议方法）	较少（yield自动处理协议）
可读性	中等（协议样板代码干扰）	高（专注于迭代逻辑）
状态管理	显式（自己维护实例变量）	隐式（函数帧自动保存）
StopIteration	手动抛出	函数返回时自动抛出
额外方法	可添加任意方法	具有 send(), throw(), close()
适用场景	复杂状态/需额外API	简单迭代/数据转换
可重置性	可设计为可重置	不可重置（需重新调用函数）

何时选择手写迭代器

尽管生成器语法更简洁，但在以下场景中手写迭代器是更好的选择：

迭代逻辑涉及复杂的内部状态（如树遍历需要维护栈结构）
迭代器需要提供额外的属性和方法（如 .close()、.progress()）
需要精确控制生命周期（如确保文件句柄的正确释放）
实现双向迭代或可重置迭代（生成器一旦耗尽不可重置）
需要线程安全的迭代器实现

# 生成器无法胜任的复杂场景
class TreeWalker:
    """
    目录树遍历器 — 需要额外方法，适合用手写迭代器实现。
    生成器难以同时提供遍历和额外API。
    """
    def __init__(self, root_path):
        self.root_path = root_path
        self.stats = {"dirs": 0, "files": 0, "errors": 0}

    def __iter__(self):
        return self._walk(self.root_path)

    def _walk(self, path):
        """实际的遍历逻辑"""
        import os
        try:
            entries = os.scandir(path)
        except PermissionError:
            self.stats["errors"] += 1
            return

        for entry in entries:
            if entry.is_dir():
                self.stats["dirs"] += 1
                yield entry.path
                yield from self._walk(entry.path)
            else:
                self.stats["files"] += 1
                yield entry.path

    def get_statistics(self):
        """返回遍历统计信息（生成器迭代器无法提供此方法）"""
        return dict(self.stats)

    def reset_statistics(self):
        """重置统计信息"""
        self.stats = {"dirs": 0, "files": 0, "errors": 0}

# 使用示例
# walker = TreeWalker("/some/path")
# for path in walker:
#     print(path, walker.get_statistics())

选择指南：对于80%的迭代场景，生成器是最简洁、最正确的选择。只有当你的迭代器需要"身份"（即需要 isinstance 检查）、需要额外方法、或者迭代逻辑确实过于复杂时，才考虑手写迭代器类。

九、实战案例

理论的学习最终要回归实践。本节将通过两个完整的实战案例，展示迭代器协议在实际项目中的应用，包括如何处理真实世界中的挑战。

案例一：分页API迭代器

在实际开发中，经常需要从远程API获取分页数据。使用迭代器封装分页逻辑，可以屏蔽分页细节，让调用方专注于数据处理：

import time
from typing import Any, Optional, Callable

class PaginatedAPIWrapper:
    """
    通用分页API迭代器。
    自动处理翻页，透明地提供所有页面的数据。
    """
    def __init__(
        self,
        fetch_page_func: Callable,
        page_size: int = 100,
        max_pages: Optional[int] = None,
        delay_seconds: float = 0.5,
    ):
        """
        Args:
            fetch_page_func: 接收 (page_number, page_size) 参数，
                             返回 (items_list, has_more) 的函数
            page_size: 每页大小
            max_pages: 最大页数限制，None表示不限
            delay_seconds: 请求间延迟，避免触发限流
        """
        self.fetch_page = fetch_page_func
        self.page_size = page_size
        self.max_pages = max_pages
        self.delay = delay_seconds

        # 统计信息
        self.pages_fetched = 0
        self.total_items = 0
        self.elapsed_time = 0.0

    def __iter__(self):
        return self._iterator()

    def _iterator(self):
        """内部迭代器 — 使用生成器实现"""
        page = 1
        start_time = time.time()

        while self.max_pages is None or page <= self.max_pages:
            # 请求当前页
            items, has_more = self.fetch_page(page, self.page_size)
            self.pages_fetched += 1

            if not items:
                break

            # 逐个 yield 当前页的元素
            for item in items:
                self.total_items += 1
                yield item

            # 如果没有更多页面，退出循环
            if not has_more:
                break

            # 避免请求过于频繁
            if self.delay > 0:
                time.sleep(self.delay)
            page += 1

        self.elapsed_time = time.time() - start_time

    def get_stats(self):
        """返回请求统计信息"""
        return {
            "pages_fetched": self.pages_fetched,
            "total_items": self.total_items,
            "elapsed_seconds": round(self.elapsed_time, 2),
        }

# 模拟API调用（实际使用时替换为真实API）
def mock_fetch_page(page, page_size):
    """模拟的分页API"""
    start = (page - 1) * page_size
    items = [f"数据项_{i}" for i in range(start, start + page_size)]
    has_more = page < 5  # 模拟5页数据
    return items, has_more

# 使用分页迭代器
wrapper = PaginatedAPIWrapper(
    fetch_page_func=mock_fetch_page,
    page_size=20,
    delay_seconds=0.1
)

# 透明地遍历所有页面数据
results = []
for item in wrapper:
    results.append(item)

print(f"共获取 {len(results)} 条数据")
print(f"统计信息: {wrapper.get_stats()}")
# 输出: 共获取 100 条数据
# 统计信息: {'pages_fetched': 5, 'total_items': 100, 'elapsed_seconds': 0.51}

# 新迭代自动从头开始
print("第二次迭代将重新从第1页开始")

案例二：数据流管道系统

构建一个可组合的流式数据处理管道，每个阶段都是独立的迭代器，可以灵活地插入、移除和组合：

from itertools import islice

class PipelineStage:
    """管道阶段的基类"""
    def __init__(self, source=None):
        self.source = source

    def __iter__(self):
        return self.process(self.source)

    def process(self, data):
        """子类重写此方法实现具体处理逻辑"""
        raise NotImplementedError


class DataSource(PipelineStage):
    """数据源 — 管道的起点"""
    def __init__(self, data):
        super().__init__(iter(data))

    def process(self, data):
        for item in data:
            yield item


class Filter(PipelineStage):
    """过滤阶段"""
    def __init__(self, source, predicate):
        super().__init__(source)
        self.predicate = predicate

    def process(self, data):
        for item in data:
            if self.predicate(item):
                yield item


class Map(PipelineStage):
    """转换阶段"""
    def __init__(self, source, transform):
        super().__init__(source)
        self.transform = transform

    def process(self, data):
        for item in data:
            yield self.transform(item)


class Batch(PipelineStage):
    """分批阶段 — 将流式数据聚合成批次"""
    def __init__(self, source, batch_size):
        super().__init__(source)
        self.batch_size = batch_size

    def process(self, data):
        batch = []
        for item in data:
            batch.append(item)
            if len(batch) >= self.batch_size:
                yield batch
                batch = []
        if batch:  # 处理最后剩余的数据
            yield batch


class Take(PipelineStage):
    """限制阶段 — 只取前N个元素"""
    def __init__(self, source, n):
        super().__init__(source)
        self.n = n

    def process(self, data):
        count = 0
        for item in data:
            if count >= self.n:
                break
            yield item
            count += 1


# ========== 管道组合使用 ==========

# 原始数据：模拟传感器读数
sensor_data = [
    {"id": i, "value": i * 1.5, "valid": i % 3 != 0}
    for i in range(1, 31)
]

# 构建管道: 数据源 → 过滤无效 → 提取value → 取前10个
pipeline = Take(
    Map(
        Filter(
            DataSource(sensor_data),
            lambda x: x["valid"]
        ),
        lambda x: x["value"]
    ),
    10
)

print("=== 管道执行结果 ===")
for value in pipeline:
    print(f"  处理值: {value:.1f}")

# 管道可以复用 — 创建新实例而不是复用已消耗的实例
print(f"有效数据点总数: {sum(1 for _ in Filter(DataSource(sensor_data), lambda x: x['valid']))}")

# 更复杂的管道：分批处理
batched_pipeline = Take(
    Batch(
        Filter(DataSource(sensor_data), lambda x: x["valid"]),
        batch_size=4
    ),
    2
)

print("\n=== 分批处理结果 ===")
for batch in batched_pipeline:
    print(f"  批次: {[item['id'] for item in batch]}")

管道模式优势：管道中的每个阶段都是独立的迭代器，整个管道采用惰性求值——数据只有在最终消费时才逐流过每个阶段。这意味着：第一，内存占用始终是O(1)；第二，可以提前终止（如 Take 阶段截断数据后，上游阶段立即停止处理）；第三，每个阶段都可以独立测试和复用。

十、最佳实践与常见陷阱

在多年实践中，Python开发者总结出了关于迭代器使用的一系列最佳实践和需要避免的陷阱。掌握这些经验可以避免很多难以调试的问题。

最佳实践清单

优先使用生成器：对于简单的迭代需求，生成器函数（yield）比手写迭代器类更简洁、更不易出错。
可迭代对象与迭代器分离：如果需要支持重复遍历，务必采用"可迭代对象返回新迭代器"的模式，而不是让迭代器对象自身可重置。
使用 collections.abc 进行类型检查：用 isinstance(x, Iterable) 和 isinstance(x, Iterator) 进行协议检查，比检查具体类型更符合Python哲学。
注意一次性消费：迭代器只能遍历一次。如果需要反复使用，用 list() 将结果固化为列表。
使用 itertools 工具库：标准库 itertools 提供了大量高效的迭代器工具，如 islice、chain、cycle、zip_longest 等，在实现迭代逻辑时优先考虑。
大文件逐行读取：for line in open('file.txt'): 利用了文件对象的迭代器特性，逐行读取而非一次性将整个文件加载到内存。

常见陷阱（附解决方案）

陷阱1：迭代器被意外消耗

# 错误示例
numbers = [1, 2, 3, 4, 5]
squared = (x**2 for x in numbers)  # 生成器表达式

if 25 in squared:       # 第一次遍历: 消耗了迭代器
    print("25 在序列中")
print(list(squared))    # 第二次遍历: 已经空了！输出 []

# 正确做法：需要多次遍历时提前固化
squared_list = [x**2 for x in numbers]  # 列表推导式
if 25 in squared_list:
    print("25 在序列中")
print(squared_list)     # 可以反复使用

陷阱2：在迭代过程中修改容器

# 错误示例：边遍历边修改列表
my_list = [1, 2, 3, 4, 5]
for i, item in enumerate(my_list):
    if item % 2 == 0:
        my_list.pop(i)   # 删除元素导致索引错位！
# 结果不可预期

# 正确做法：创建副本或使用列表推导
my_list = [1, 2, 3, 4, 5]
my_list = [x for x in my_list if x % 2 != 0]  # 重新创建列表
print(my_list)  # [1, 3, 5]

# 或者遍历副本
my_list = [1, 2, 3, 4, 5]
for item in my_list[:]:  # 遍历切片副本
    if item % 2 == 0:
        my_list.remove(item)

陷阱3：StopIteration 被生成器内部意外捕获

# 错误示例
def broken_generator():
    items = [1, 2, 3]
    iterator = iter(items)
    try:
        while True:
            yield next(iterator)
    except StopIteration:
        # Python 3.7+ 中，生成器内部的 StopIteration
        # 会被自动转换为 RuntimeError！
        pass  # 这种行为在 Python 3.7+ 中被弃用

# 正确做法：在生成器中使用 return 而非捕获 StopIteration
def correct_generator():
    items = [1, 2, 3]
    for item in items:  # for 循环自动处理 StopIteration
        yield item

# 或者使用 yield from
def correct_generator2():
    items = [1, 2, 3]
    yield from items  # yield from 正确处理迭代协议

itertools 实用工具速览

标准库 itertools 模块提供了一组高效的迭代器工具函数，建议所有Python开发者熟练掌握：

# itertools 常用工具演示
from itertools import (
    count, cycle, repeat,
    chain, compress, dropwhile, takewhile,
    islice, zip_longest, product, permutations,
    combinations, groupby, accumulate
)

# 1. count(start, step) — 无穷计数器
for i in islice(count(10, 2), 5):
    print(i, end=' ')  # 10 12 14 16 18
print()

# 2. cycle(iterable) — 无限循环
for i, item in enumerate(cycle(['A', 'B', 'C'])):
    if i >= 7:
        break
    print(item, end=' ')  # A B C A B C A
print()

# 3. chain — 串联多个可迭代对象
combined = chain([1, 2], ['a', 'b'], (10, 20))
print(list(combined))  # [1, 2, 'a', 'b', 10, 20]

# 4. islice — 切片迭代器
result = islice(range(100), 10, 20, 2)
print(list(result))    # [10, 12, 14, 16, 18]

# 5. accumulate — 累积计算
result = accumulate([1, 2, 3, 4, 5])
print(list(result))    # [1, 3, 6, 10, 15] (默认累加)

# 6. product — 笛卡尔积
result = product('AB', [1, 2])
print(list(result))    # [('A', 1), ('A', 2), ('B', 1), ('B', 2)]

# 7. groupby — 分组（需要已排序数据）
data = [('A', 1), ('A', 2), ('B', 3), ('B', 4)]
for key, group in groupby(data, key=lambda x: x[0]):
    print(f"键 {key}: {list(group)}")
    # 输出:
    # 键 A: [('A', 1), ('A', 2)]
    # 键 B: [('B', 3), ('B', 4)]

总结与思维导图

迭代器协议是Python语言中最为核心和优雅的设计之一。通过 __iter__ 和 __next__ 两个方法，Python定义了一套统一的遍历接口，使得列表、文件、数据库查询、传感器数据流等截然不同的数据源可以用完全相同的方式处理。

核心概念回顾：

迭代器协议：__iter__() 返回迭代器，__next__() 返回下一个元素，StopIteration 标记结束。
Iterable vs Iterator：可迭代对象可重复创建迭代器（有 __iter__），迭代器一次性消费（有 __iter__ + __next__）。
for循环的本质：先 iter() 获取迭代器，然后无限 next() 直到 StopIteration。
惰性求值：只在需要时才计算下一个元素，内存效率极高，可处理无限序列。
生成器：基于 yield 的自动迭代器创建方式，更简洁，适用于80%的场景。
最佳实践：优先用生成器，可迭代对象与迭代器分离，注意一次性消费陷阱。

进一步学习的建议：

深入学习 itertools 模块，掌握 chain、islice、groupby、product 等高效迭代工具。
研究 asyncio 中的异步迭代器协议（__aiter__ 和 __anext__）。
了解 yield from 语法在委派生成器（delegating generator）中的应用。
阅读 CPython 源码中迭代器协议的具体实现，加深对底层机制的理解。
尝试将数据管道模式应用到实际项目中，体验惰性求值在大数据处理中的威力。

一句话总结：迭代器协议定义了Python遍历数据的方式，它让代码更简洁、内存更高效、表达力更强。掌握了迭代器协议，你就掌握了Python数据流处理的精髓。