Python 定义一个迭代器类（手把手讲解）

迭代器类的基本概念

在 Python 编程中，迭代器是一个重要的概念。它允许我们按需逐个访问序列中的元素，而无需一次性将所有数据加载到内存中。理解迭代器类的定义方式，可以帮助我们更好地处理大数据集、实现自定义数据流控制等场景。迭代器类的核心在于实现两个特殊方法：__iter__ 和 __next__。

迭代器类与可迭代对象（如列表、元组）存在本质区别。可迭代对象是容器，迭代器则是从容器中逐个获取元素的工具。我们可以把迭代器想象成一个播放器，它从歌曲库（可迭代对象）中依次读取每首歌曲（元素），而不是一次性将所有歌曲都加载到内存中。

创建迭代器类的必要条件

要定义一个迭代器类，需要同时实现两个魔法方法：

class MyIterator:
    def __iter__(self):
        # 初始化方法
        return self
    
    def __next__(self):
        # 获取下一个元素的逻辑
        pass

iter 方法的作用

iter 方法返回迭代器对象本身。它的功能相当于初始化迭代过程，就像给播放器装入磁带。这个方法确保迭代器在每次使用时都能从初始状态开始。

next 方法的职责

next 方法负责返回下一个元素。当没有更多元素时，它需要抛出 StopIteration 异常。这个方法就像播放器的下一个按钮，每次点击播放下一段内容。

迭代器类的实现示例

基本的数字迭代器

class NumberIterator:
    def __init__(self, max_value):
        # 初始化最大值和当前值
        self.max_value = max_value
        self.current = 0
    
    def __iter__(self):
        # 必须返回迭代器对象
        return self
    
    def __next__(self):
        # 判断是否超出范围
        if self.current >= self.max_value:
            # 如果超出范围，抛出停止异常
            raise StopIteration
        # 获取当前值并递增
        result = self.current
        self.current += 1
        return result

for num in NumberIterator(5):
    print(num)  # 输出 0 1 2 3 4

斐波那契数列迭代器

class Fibonacci:
    def __init__(self, count):
        # 初始化参数
        self.count = count
        self.a, self.b = 0, 1
        self.current = 0
    
    def __iter__(self):
        # 返回迭代器对象
        return self
    
    def __next__(self):
        # 当已生成数量超过限制时停止
        if self.current >= self.count:
            raise StopIteration
        # 保存当前值
        result = self.a
        # 计算下一个数
        self.a, self.b = self.b, self.a + self.b
        self.current += 1
        return result

for num in Fibonacci(10):
    print(num)  # 输出前10个斐波那契数

迭代器类的高级用法

添加异常处理机制

class SafeIterator:
    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
        result = self.data[self.index]
        self.index += 1
        return result

data = [1, 2, 3]
try:
    for item in SafeIterator(data):
        print(item)
    # 强制访问超出范围
    print(next(SafeIterator(data)))
except StopIteration:
    print("已到达迭代器末尾")

实现双向迭代

class BidirectionalIterator:
    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
        result = self.data[self.index]
        self.index += 1
        return result
    
    def previous(self):
        # 自定义前移方法
        if self.index <= 0:
            raise StopIteration
        self.index -= 1
        return self.data[self.index]

bi = BidirectionalIterator([10, 20, 30])
for item in bi:
    print(item)  # 正向迭代输出 10 20 30

print(bi.previous())  # 输出 30
print(bi.previous())  # 输出 20

迭代器类的适用场景

处理无限序列

class InfiniteCounter:
    def __init__(self, start=0):
        self.value = start
    
    def __iter__(self):
        return self
    
    def __next__(self):
        current = self.value
        self.value += 1
        return current

counter = InfiniteCounter()
for _ in range(5):
    print(next(counter))  # 输出 0 1 2 3 4

文件行读取优化

class FileLineIterator:
    def __init__(self, file_path):
        self.file = open(file_path, 'r')
        self.eof = False
    
    def __iter__(self):
        return self
    
    def __next__(self):
        line = self.file.readline()
        if not line:
            self.file.close()
            raise StopIteration
        return line.strip()

with open('example.txt', 'w') as f:
    f.write("第一行\n第二行\n第三行")

with open('example.txt', 'r') as f:
    for line in FileLineIterator(f.name):
        print(line)  # 逐行读取文件内容

迭代器类的性能优化

内存效率对比

迭代器类在处理大型数据集时，相比列表等容器具有显著优势。它不会一次性将所有元素存储在内存中，而是按需生成，就像流水线上的产品，只在需要时进行生产。

提前释放资源

class ResourceIterator:
    def __init__(self, resource):
        self.resource = resource
        self.index = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.index >= len(self.resource):
            raise StopIteration
        result = self.resource[self.index]
        self.index += 1
        return result
    
    def __del__(self):
        # 资源清理方法
        print("迭代器资源已释放")

ri = ResourceIterator([1, 2, 3])
for item in ri:
    print(item)  # 输出 1 2 3

迭代器类的扩展技巧

与生成器的协同

class GeneratorWrapper:
    def __init__(self, generator_func, *args, **kwargs):
        self.generator = generator_func(*args, **kwargs)
    
    def __iter__(self):
        return self
    
    def __next__(self):
        return next(self.generator)

def my_generator(max_value):
    for i in range(max_value):
        yield i

wrapped = GeneratorWrapper(my_generator, 3)
for item in wrapped:
    print(item)  # 输出 0 1 2

迭代器组合应用

class CombinedIterator:
    def __init__(self, *iterators):
        self.iterators = [iter(it) for it in iterators]
        self.current_index = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.current_index >= len(self.iterators):
            raise StopIteration
        current_iter = self.iterators[self.current_index]
        try:
            return next(current_iter)
        except StopIteration:
            self.current_index += 1
            return next(self)

combined = CombinedIterator([1, 2], 'ab', (3,4))
for item in combined:
    print(item)  # 输出 1 a 3

常见问题与解决方案

迭代器类无法重复使用

class ReusableIterator:
    def __init__(self, data):
        self.data = data
    
    def __iter__(self):
        # 每次迭代创建新的状态
        self.index = 0
        return self
    
    def __next__(self):
        if self.index >= len(self.data):
            raise StopIteration
        result = self.data[self.index]
        self.index += 1
        return result

ri = ReusableIterator([5, 6, 7])
for item in ri:
    print(item)  # 输出 5 6 7

for item in ri:
    print(item)  # 再次迭代同样输出 5 6 7

迭代器类的异常处理

class ErrorHandledIterator:
    def __init__(self, data):
        self.data = data
        self.index = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        try:
            result = self.data[self.index]
            self.index += 1
            return result
        except IndexError:
            raise StopIteration
        except Exception as e:
            print(f"迭代器错误: {e}")
            self.index += 1
            return self.__next__()

ehi = ErrorHandledIterator([1, 'a', 3, 4])
for item in ehi:
    print(item)  # 会跳过错误数据继续执行

迭代器类的进阶应用

实现自定义数据结构

class MatrixIterator:
    def __init__(self, matrix):
        self.matrix = matrix
        self.row = 0
        self.col = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        # 判断是否超出矩阵范围
        if self.row >= len(self.matrix):
            raise StopIteration
        result = self.matrix[self.row][self.col]
        self.col += 1
        if self.col >= len(self.matrix[0]):
            self.row += 1
            self.col = 0
        return result

matrix = [[1, 2], [3, 4]]
for item in MatrixIterator(matrix):
    print(item)  # 依次访问矩阵元素

与装饰器的结合使用

def log_decorator(iterator_class):
    def wrapper(*args, **kwargs):
        instance = iterator_class(*args, **kwargs)
        def wrapped_next():
            try:
                print(f"准备获取下一个元素")
                return instance.__next__()
            except StopIteration:
                print("迭代结束")
                raise
        instance.__next__ = wrapped_next
        return instance
    return wrapper

@log_decorator
class LoggedIterator:
    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
        result = self.data[self.index]
        self.index += 1
        return result

li = LoggedIterator([100, 200])
for item in li:
    print(item)  # 每次迭代都会打印日志

迭代器类的调试技巧

添加调试信息输出

class DebugIterator:
    def __init__(self, data):
        self.data = data
        self.index = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        print(f"Debug: 当前索引 {self.index}")
        if self.index >= len(self.data):
            raise StopIteration
        result = self.data[self.index]
        self.index += 1
        return result

di = DebugIterator([1, 2, 3])
for item in di:
    print(item)

使用断言进行验证

class ValidationIterator:
    def __init__(self, data):
        assert isinstance(data, list), "输入必须是列表"
        self.data = data
        self.index = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        assert self.index <= len(self.data), "索引越界"
        if self.index >= len(self.data):
            raise StopIteration
        result = self.data[self.index]
        self.index += 1
        return result

vi = ValidationIterator([5, 6, 7])
for item in vi:
    print(item)  # 运行时验证迭代过程

迭代器类的兼容性处理

适配不同数据源

class DataSourceAdapter:
    def __init__(self, source):
        self.source = source
        self.index = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.index >= len(self.source):
            raise StopIteration
        # 自动适配不同数据结构
        if isinstance(self.source, dict):
            result = list(self.source.items())[self.index]
        elif isinstance(self.source, set):
            result = list(self.source)[self.index]
        else:
            result = self.source[self.index]
        self.index += 1
        return result

adapter = DataSourceAdapter({'a': 1, 'b': 2})
for item in adapter:
    print(item)  # 输出字典的键值对

迭代器类的错误排查

使用迭代器类的正确姿势

class ProperUsageIterator:
    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
        result = self.data[self.index]
        self.index += 1
        return result

for item in ProperUsageIterator([1, 2, 3]):
    print(item)

pui = ProperUsageIterator(['a', 'b', 'c'])
print(next(pui))  # 输出 a
print(pui.__next__())  # 输出 b

迭代器类的性能考量

内存消耗对比

import sys

class MemoryEfficient:
    def __init__(self, max_value):
        self.max_value = max_value
        self.current = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.current >= self.max_value:
            raise StopIteration
        self.current += 1
        return self.current - 1

list_memory = sys.getsizeof(list(range(1000000)))
iterator_memory = sys.getsizeof(MemoryEfficient(1000000))
print(f"列表占用内存: {list_memory/1024:.2f} KB")
print(f"迭代器占用内存: {iterator_memory/1024:.2f} KB")

速度优化技巧

class SpeedOptimizedIterator:
    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
        result = self.data[self.index]
        self.index += 1
        return result

import timeit

def test_list():
    return [i for i in range(1000)]

def test_iterator():
    return list(SpeedOptimizedIterator(range(1000)))

print("列表生成时间:", timeit.timeit(test_list, number=10000))
print("迭代器生成时间:", timeit.timeit(test_iterator, number=10000))

迭代器类的设计模式

状态机模式实现

class StateMachineIterator:
    def __init__(self, states):
        self.states = states
        self.state_index = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.state_index >= len(self.states):
            raise StopIteration
        state = self.states[self.state_index]
        self.state_index += 1
        return state

machine = StateMachineIterator(['start', 'process', 'end'])
for phase in machine:
    print(f"当前状态: {phase}")

迭代器组合模式

class IteratorComposition:
    def __init__(self, *iterators):
        self.iterators = [iter(it) for it in iterators]
        self.current = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.current >= len(self.iterators):
            raise StopIteration
        try:
            return next(self.iterators[self.current])
        except StopIteration:
            self.current += 1
            return self.__next__()

composite = IteratorComposition([1, 2], 'xy', {3,4})
for item in composite:
    print(item)  # 输出 1 x 3 y 4

迭代器类的异常处理

多种异常场景覆盖

class RobustIterator:
    def __init__(self, data):
        self.data = data
        self.index = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        try:
            if self.index >= len(self.data):
                raise StopIteration
            result = self.data[self.index]
            self.index += 1
            return result
        except TypeError as e:
            print(f"类型错误: {e}")
            self.index += 1
            return self.__next__()
        except Exception as e:
            print(f"未知错误: {e}")
            raise

ri = RobustIterator([1, 'a', 3])
for item in ri:
    print(item)  # 会捕获并处理类型错误

迭代器类的测试方法

单元测试示例

import unittest

class TestMyIterator(unittest.TestCase):
    def test_basic_function(self):
        iterator = NumberIterator(3)
        self.assertEqual(list(iterator), [0, 1, 2])
    
    def test_stop_iteration(self):
        iterator = NumberIterator(2)
        self.assertEqual(next(iterator), 0)
        self.assertEqual(next(iterator), 1)
        with self.assertRaises(StopIteration):
            next(iterator)

if __name__ == '__main__':
    unittest.main()

性能测试对比

import time

class PerformanceTest:
    @staticmethod
    def test_list(n):
        start = time.time()
        for i in range(n):
            pass
        return time.time() - start
    
    @staticmethod
    def test_iterator(n):
        start = time.time()
        for i in NumberIterator(n):
            pass
        return time.time() - start

print("range性能:", PerformanceTest.test_list(1000000))
print("迭代器性能:", PerformanceTest.test_iterator(1000000))

迭代器类的使用建议

何时选择迭代器类

1. 当处理的数据量非常庞大时
2. 需要延迟计算或按需生成数据时
3. 需要自定义迭代逻辑时
4. 需要与生成器、协程等异步机制配合时

迭代器类的最佳实践

• 始终实现 iter 和 next 方法
• 在 next 中使用 try/except 捕获异常
• 为迭代器添加适当的文档字符串
• 实现 del 方法清理外部资源
• 考虑使用组合模式支持多个数据源

迭代器类的扩展功能

实现跳步迭代

class StepIterator:
    def __init__(self, start, stop, step):
        self.start = start
        self.stop = stop
        self.step = step
        self.current = start
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if (self.step > 0 and self.current >= self.stop) or \
           (self.step < 0 and self.current <= self.stop):
            raise StopIteration
        result = self.current
        self.current += self.step
        return result

for num in StepIterator(0, 10, 2):
    print(num)  # 输出 0 2 4 6 8

支持反向迭代

class ReverseIterator:
    def __init__(self, data):
        self.data = data
        self.index = len(data) - 1
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.index < 0:
            raise StopIteration
        result = self.data[self.index]
        self.index -= 1
        return result

rev = ReverseIterator([10, 20, 30])
for item in rev:
    print(item)  # 输出 30 20 10

迭代器类的装饰器模式

添加日志功能

class LoggingIterator:
    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
        print(f"迭代器访问索引 {self.index}")
        result = self.data[self.index]
        self.index += 1
        return result

li = LoggingIterator(['a', 'b', 'c'])
for item in li:
    print(item)  # 每次迭代都会记录访问日志

添加缓存机制

class CachingIterator:
    def __init__(self, data):
        self.data = data
        self.index = 0
        self.cache = []
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.index >= len(self.data):
            raise StopIteration
        # 如果有缓存则直接返回
        if self.index < len(self.cache):
            result = self.cache[self.index]
        else:
            # 模拟耗时计算
            result = self.data[self.index] * 2
            self.cache.append(result)
        self.index += 1
        return result

ci = CachingIterator([1, 2, 3])
for item in ci:
    print(item)  # 输出 2 4 6

迭代器类的进阶应用场景

处理二进制数据流

class BinaryReader:
    def __init__(self, file_path):
        self.file = open(file_path, 'rb')
        self.read_bytes = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        chunk = self.file.read(1024)
        if not chunk:
            self.file.close()
            raise StopIteration
        self.read_bytes += len(chunk)
        return chunk

with open('binary_data.bin', 'wb') as f:
    f.write(b'Hello World' * 1000)

with open('binary_data.bin', 'rb') as f:
    for chunk in BinaryReader(f.name):
        print(f"读取 {len(chunk)} 字节")

实现管道式处理

class PipelineIterator:
    def __init__(self, data, *transforms):
        self.data = data
        self.transforms = transforms
        self.index = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.index >= len(self.data):
            raise StopIteration
        result = self.data[self.index]
        # 应用所有转换函数
        for func in self.transforms:
            result = func(result)
        self.index += 1
        return result

def square(x):
    return x ** 2

def add_one(x):
    return x + 1

pipe = PipelineIterator([1, 2, 3], square, add_one)
for item in pipe:
    print(item)  # 输出 2 5 10

总结与展望

掌握 Python 定义一个迭代器类的技巧，能让开发者更灵活地处理各种数据流。通过实现 iter 和 next 两个核心方法，我们能够创建出具有自定义迭代逻辑的对象。迭代器类在大数据处理、资源管理、算法实现等场景中都发挥着重要作用。

随着 Python 3.10 的发布，迭代器类的实现方式有了更多优化空间。建议读者在实际开发中：

1. 对于小型数据优先使用列表等内置类型
2. 处理大型数据时考虑使用迭代器
3. 在需要自定义迭代逻辑时实现迭代器类
4. 结合生成器使用实现更简洁的代码
5. 使用装饰器模式添加日志、缓存等功能

通过反复实践和优化迭代器类的实现方式，开发者可以编写出更高效、更优雅的 Python 代码。在面对复杂的数据处理需求时，自定义迭代器往往能提供最佳解决方案。记住，每个优秀的迭代器类都应该像一个可靠的生产流水线，稳定地输出需要的元素。