python的迭代器

1.基本概念

迭代器是 Python 中用于遍历数据集合的工具，它提供了一种统一、高效的方式来访问各种数据结构中的元素。

1.1.什么是迭代器？

迭代器是一个实现了迭代协议的对象，具有以下特征：

• 迭代协议：必须实现 __iter__() 和 __next__() 方法
  • __iter__()：返回迭代器对象本身，使对象可以被 iter() 调用
  • __next__()：返回下一个元素，如果没有更多元素则抛出 StopIteration 异常
• 惰性求值：按需生成元素，不预计算所有值，节省内存
• 状态保持：记住当前的遍历位置
• 单向遍历：只能向前遍历，不能后退或重复

1.2.迭代器 vs 可迭代对象

理解迭代器和可迭代对象的区别很重要：

• 可迭代对象 (Iterable)：可以被迭代的对象，如列表、元组、字符串等
• 迭代器 (Iterator)：实际执行迭代的对象，实现了 __next__() 方法

1from collections.abc import Iterable, Iterator
2
3# 列表是可迭代对象，但不是迭代器
4numbers = [1, 2, 3]
5print(isinstance(numbers, Iterable))  # True
6print(isinstance(numbers, Iterator))  # False
7
8# 通过 iter() 获取迭代器
9numbers_iterator = iter(numbers)
10print(isinstance(numbers_iterator, Iterator))  # True

重要说明：

• 可迭代对象可以通过 iter() 函数转换为迭代器
• 迭代器只能使用一次，遍历完后需要重新创建
• 大多数内置容器类型（如 list、tuple、dict）都是可迭代对象，但不是迭代器

2.迭代器协议

迭代器协议是 Python 中定义迭代器行为的标准，任何对象只要实现了这个协议，就可以被用于迭代。

2.1.必须实现的方法

要成为一个合格的迭代器，必须实现以下两个方法：

1. __iter__() 方法

   • 返回迭代器对象自身
   • 使对象可以被 for 循环或 iter() 调用
   • 通常实现为 return self

2. __next__() 方法

   • 返回下一个元素
   • 如果没有更多元素，必须抛出 StopIteration 异常
   • 这是迭代器的核心逻辑

1def __iter__(self):
2    return self
3
4def __next__(self):
5    # 实现具体的迭代逻辑
6    # 如果没有更多元素，抛出 StopIteration
7    pass

协议特点：

• 任何实现了这两个方法的对象都可以用于 for 循环
• 支持 next() 函数调用
• 可以与 enumerate()、zip() 等内置函数配合使用

2.2.完整示例

1class MyIterator:
2    """自定义迭代器示例"""
3    def __init__(self, data):
4        self.data = data
5        self.index = 0
6
7    def __iter__(self):
8        """返回迭代器自身"""
9        return self
10
11    def __next__(self):
12        """返回下一个元素"""
13        if self.index >= len(self.data):
14            raise StopIteration
15        value = self.data[self.index]
16        self.index += 1
17        return value
18
19# 使用自定义迭代器
20my_iter = MyIterator([1, 2, 3])
21for item in my_iter:
22    print(item)  # 输出: 1, 2, 3
23
24# 也可以手动调用 next()
25my_iter2 = MyIterator(['a', 'b', 'c'])
26print(next(my_iter2))  # 'a'
27print(next(my_iter2))  # 'b'
28print(next(my_iter2))  # 'c'
29# print(next(my_iter2))  # 抛出 StopIteration 异常

3.创建迭代器的方法

3.1.方法 1：实现迭代器类

通过定义一个类并实现迭代器协议来创建自定义迭代器：

1class Countdown:
2    """倒计时迭代器"""
3    def __init__(self, start):
4        self.current = start
5
6    def __iter__(self):
7        return self
8
9    def __next__(self):
10        if self.current <= 0:
11            raise StopIteration
12        value = self.current
13        self.current -= 1
14        return value
15
16# 使用倒计时迭代器
17countdown = Countdown(5)
18for number in countdown:
19    print(number, end=" ")  # 输出: 5 4 3 2 1

适用场景：

• 需要复杂的状态管理
• 需要实现特殊的迭代逻辑
• 需要封装复杂的迭代行为

3.2.方法 2：使用生成器函数

生成器是一种特殊的迭代器，使用 yield 关键字创建，代码更简洁且内存效率更高：

1def simple_gen():
2    """简单的生成器函数"""
3    print("First yield")
4    yield 1
5    print("Second yield")
6    yield 2
7    print("Done")
8
9# 创建生成器对象
10g = simple_gen()
11print(next(g))  # 输出: First yield \n 1
12print(next(g))  # 输出: Second yield \n 2
13# print(next(g))  # 会引发 StopIteration 异常
14
15# 生成器可以用 for 循环遍历
16for value in simple_gen():
17    print(f"Got: {value}")

生成器的优势：

• 代码更简洁，不需要实现 __iter__() 和 __next__() 方法
• 自动管理状态，无需手动维护索引
• 内存效率高，按需生成值
• 支持复杂的控制流（如异常处理、资源管理）

3.3.方法 3：使用生成器表达式

生成器表达式是创建迭代器最简洁的方式，语法类似列表推导式，但使用圆括号：

1# 基本语法
2(expression for item in iterable [if condition])

示例：

1# 创建平方数生成器
2squares = (x**2 for x in range(5))
3print(list(squares))  # [0, 1, 4, 9, 16]
4
5# 带条件的生成器表达式
6even_squares = (x**2 for x in range(10) if x % 2 == 0)
7print(list(even_squares))  # [0, 4, 16, 36, 64]
8
9# 手动迭代
10gen = (x * 2 for x in range(3))
11print(next(gen))  # 0
12print(next(gen))  # 2
13print(next(gen))  # 4
14# print(next(gen))  # StopIteration

生成器表达式的优势：

• 语法简洁，一行代码创建迭代器
• 内存效率高，惰性求值
• 适合简单的数据转换和过滤
• 可以与其他迭代器函数链式组合

4.内置迭代器工具

Python 提供了许多内置函数来操作迭代器，这些工具让迭代操作更加灵活和高效。

4.1.iter() 和 next() 函数

这两个函数是操作迭代器的基础工具：

• iter(iterable)：将可迭代对象转换为迭代器
• next(iterator)：获取迭代器的下一个元素
• next(iterator, default)：获取下一个元素，如果迭代结束则返回默认值

1# 创建迭代器
2numbers = [1, 2, 3, 4, 5]
3iterator = iter(numbers)
4
5# 手动获取元素
6print(next(iterator))  # 1
7print(next(iterator))  # 2
8print(next(iterator))  # 3
9print(next(iterator))  # 4
10print(next(iterator))  # 5
11# print(next(iterator))  # 抛出 StopIteration 异常
12
13# 使用默认值避免异常
14iterator2 = iter([1, 2])
15print(next(iterator2))  # 1
16print(next(iterator2))  # 2
17print(next(iterator2, "没有更多元素"))  # 没有更多元素

使用场景：

• 需要精确控制迭代过程
• 处理大型数据集时避免一次性加载
• 实现自定义的迭代逻辑

4.2.enumerate() - 带索引的迭代

enumerate() 函数在遍历可迭代对象时同时提供索引和值，是处理需要索引的场景的最佳选择。

语法：

1enumerate(iterable, start=0)

参数：

• iterable：要遍历的可迭代对象
• start：索引起始值，默认为 0

返回值：返回 (索引, 元素) 元组的迭代器

1# 基本用法
2colors = ['red', 'green', 'blue']
3for idx, color in enumerate(colors):
4    print(f"{idx}: {color}")
5# 输出:
6# 0: red
7# 1: green
8# 2: blue
9
10# 自定义起始索引
11for idx, color in enumerate(colors, start=1):
12    print(f"第{idx}个颜色: {color}")
13# 输出:
14# 第1个颜色: red
15# 第2个颜色: green
16# 第3个颜色: blue

常见应用场景：

• 需要同时访问索引和值的循环
• 创建带编号的输出
• 在循环中修改列表元素
• 生成带索引的数据结构

4.3.zip() - 并行迭代

zip() 函数将多个可迭代对象"并行"组合，每次迭代返回一个包含各对象对应元素的元组。

语法：

1zip(iter1, iter2, ..., iterN)

特点：

• 返回迭代器，惰性求值
• 以最短的可迭代对象长度为准
• 支持任意数量的可迭代对象

1# 基本用法
2names = ['Alice', 'Bob', 'Charlie']
3ages = [25, 30, 35]
4cities = ['北京', '上海', '广州']
5
6for name, age, city in zip(names, ages, cities):
7    print(f"{name}, {age}岁, 来自{city}")
8# 输出:
9# Alice, 25岁, 来自北京
10# Bob, 30岁, 来自上海
11# Charlie, 35岁, 来自广州

长度不同的处理：

1# 以最短的为准
2list1 = [1, 2, 3, 4]
3list2 = ['a', 'b']
4for num, letter in zip(list1, list2):
5    print(f"{num}: {letter}")
6# 输出:
7# 1: a
8# 2: b

高级用法：

1. 数据转置：

1# 行转列
2scores = [
3    (85, 92, 78),  # 张三的成绩
4    (76, 88, 95),  # 李四的成绩
5    (90, 85, 92)   # 王五的成绩
6]
7chinese, math, english = zip(*scores)
8print("语文成绩:", chinese)  # (85, 76, 90)
9print("数学成绩:", math)     # (92, 88, 85)
10print("英语成绩:", english)  # (78, 95, 92)

2. 与推导式结合：

1# 计算对应位置元素的和
2list1 = [1, 2, 3]
3list2 = [4, 5, 6]
4result = [x + y for x, y in zip(list1, list2)]
5print(result)  # [5, 7, 9]

3. 处理不同长度的序列：

1from itertools import zip_longest
2
3# 保留所有元素，用默认值填充
4a = [1, 2, 3]
5b = [4, 5]
6for x, y in zip_longest(a, b, fillvalue=0):
7    print(f"{x} + {y} = {x + y}")
8# 输出:
9# 1 + 4 = 5
10# 2 + 5 = 7
11# 3 + 0 = 3

4.4.map() - 映射迭代

map() 函数对可迭代对象的每个元素应用指定函数，返回一个迭代器。

语法：

1map(function, iterable, ...)

参数：

• function：要应用的函数（可以是内置函数、lambda 或自定义函数）
• iterable：一个或多个可迭代对象

特点：

• 返回惰性迭代器，需要显式消费
• 支持多个可迭代对象，以最短的为准
• 适合批量数据转换

1# 基本用法
2numbers = [1, 2, 3, 4]
3squared = map(lambda x: x ** 2, numbers)
4print(list(squared))  # [1, 4, 9, 16]
5
6# 类型转换
7nums = [10, 20, 30]
8str_list = map(str, nums)
9print(list(str_list))  # ['10', '20', '30']
10
11# 多序列处理
12list1 = [1, 2, 3]
13list2 = [4, 5, 6, 7]
14result = map(lambda x, y: x + y, list1, list2)
15print(list(result))  # [5, 7, 9]
16
17# 自定义函数
18def double(x):
19    return x * 2
20
21numbers = [5, 6, 7]
22doubled = map(double, numbers)
23print(list(doubled))  # [10, 12, 14]

优势：

• 代码简洁，函数式编程风格
• 惰性求值，内存效率高
• 易于与其他迭代器函数组合

4.5.filter() - 过滤迭代

filter() 函数根据指定条件过滤可迭代对象中的元素，只保留满足条件的元素。

语法：

1filter(function, iterable)

参数：

• function：判断函数，返回 True/False。如果为 None，则使用元素自身的布尔值
• iterable：要过滤的可迭代对象

特点：

• 返回惰性迭代器，需要显式消费
• 只保留使函数返回 True 的元素
• 适合数据清洗和条件筛选

1# 基本用法
2numbers = [1, 2, 3, 4, 5, 6]
3even_numbers = filter(lambda x: x % 2 == 0, numbers)
4print(list(even_numbers))  # [2, 4, 6]
5
6# 过滤假值
7data = ['hello', '', 'python', None, ' ', 'AI']
8valid_data = filter(None, data)
9print(list(valid_data))  # ['hello', 'python', ' ', 'AI']
10
11# 自定义过滤函数
12def is_positive(x):
13    return x > 0
14
15numbers = [0, -1, 2, -3, 4]
16positive = filter(is_positive, numbers)
17print(list(positive))  # [2, 4]
18
19# 复杂条件
20def is_valid_email(email):
21    return email and '@' in email and '.' in email
22
23emails = ['user@example.com', 'invalid', 'test@domain.org', '']
24valid_emails = filter(is_valid_email, emails)
25print(list(valid_emails))  # ['user@example.com', 'test@domain.org']

应用场景：

• 数据清洗和验证
• 条件筛选
• 去除无效数据
• 与 map() 等函数组合使用

5.自定义迭代器示例

5.1.范围迭代器

实现一个类似内置 range() 的自定义迭代器：

1class RangeIterator:
2    """自定义范围迭代器，类似内置的 range()"""
3
4    def __init__(self, start, stop=None, step=1):
5        # 处理单参数情况：range(5) -> start=0, stop=5
6        if stop is None:
7            self.start = 0
8            self.stop = start
9        else:
10            self.start = start
11            self.stop = stop
12        self.step = step
13        self.current = self.start
14
15    def __iter__(self):
16        return self
17
18    def __next__(self):
19        # 检查终止条件
20        if (self.step > 0 and self.current >= self.stop) or \
21           (self.step < 0 and self.current <= self.stop):
22            raise StopIteration
23
24        value = self.current
25        self.current += self.step
26        return value
27
28# 使用示例
29print("正步长:")
30for i in RangeIterator(1, 5):
31    print(i, end=" ")  # 输出: 1 2 3 4
32
33print("\n负步长:")
34for i in RangeIterator(5, 1, -1):
35    print(i, end=" ")  # 输出: 5 4 3 2

实现要点：

• 支持正负步长
• 处理单参数和双参数情况
• 正确实现终止条件判断

5.2.文件行迭代器

实现一个安全的文件行迭代器，自动管理文件资源：

1class FileLineIterator:
2    """文件行迭代器，自动管理文件资源"""
3
4    def __init__(self, filename):
5        self.filename = filename
6        self.file = None
7
8    def __iter__(self):
9        # 打开文件
10        self.file = open(self.filename, 'r', encoding='utf-8')
11        return self
12
13    def __next__(self):
14        if self.file is None:
15            raise RuntimeError("必须先调用 __iter__ 方法")
16
17        line = self.file.readline()
18        if not line:  # 文件结束
19            self.file.close()
20            raise StopIteration
21
22        return line.strip()
23
24    def __del__(self):
25        """确保文件被关闭"""
26        if self.file and not self.file.closed:
27            self.file.close()
28
29# 使用示例
30def process_file(filename):
31    """处理文件内容"""
32    for line in FileLineIterator(filename):
33        if line:  # 跳过空行
34            print(f"处理: {line}")
35
36# 更安全的使用方式
37def safe_process_file(filename):
38    """更安全的文件处理方式"""
39    try:
40        with open(filename, 'r', encoding='utf-8') as file:
41            for line in file:
42                line = line.strip()
43                if line:
44                    print(f"处理: {line}")
45    except FileNotFoundError:
46        print(f"文件 {filename} 不存在")
47    except Exception as e:
48        print(f"处理文件时出错: {e}")
49
50# 使用
51process_file('example.txt')

改进建议：

• 使用 with 语句更安全
• 添加异常处理
• 考虑使用生成器函数更简洁

5.3.树结构迭代器

实现一个树结构的前序遍历迭代器，用于处理层级数据：

1class TreeNode:
2    """树节点类"""
3    def __init__(self, value):
4        self.value = value
5        self.children = []
6
7    def add_child(self, child):
8        self.children.append(child)
9
10class TreeIterator:
11    """树的前序遍历迭代器"""
12    def __init__(self, root):
13        self.stack = [root] if root else []
14
15    def __iter__(self):
16        return self
17
18    def __next__(self):
19        if not self.stack:
20            raise StopIteration
21
22        # 弹出当前节点
23        node = self.stack.pop()
24
25        # 将子节点逆序加入栈（保证前序遍历顺序）
26        for child in reversed(node.children):
27            self.stack.append(child)
28
29        return node.value
30
31# 构建示例树
32#        A
33#       / \
34#      B   C
35#     / \
36#    D   E
37root = TreeNode('A')
38b = TreeNode('B')
39c = TreeNode('C')
40d = TreeNode('D')
41e = TreeNode('E')
42
43root.add_child(b)
44root.add_child(c)
45b.add_child(d)
46b.add_child(e)
47
48# 遍历树
49print("前序遍历:")
50for value in TreeIterator(root):
51    print(value, end=" ")  # 输出: A B D E C

实现要点：

• 使用栈结构保存遍历状态
• 子节点逆序入栈保证前序遍历
• 支持任意深度的树结构

6.无限迭代器

6.1.itertools 中的无限迭代器

itertools 模块提供了多种无限迭代器，可以持续产生数据：

• count(start=0, step=1)：生成无限数字序列
• cycle(iterable)：无限循环遍历可迭代对象
• repeat(object, times=None)：重复指定对象

1import itertools
2
3# count - 无限计数器
4counter = itertools.count(10, 2)
5print("Count:", [next(counter) for _ in range(5)])  # [10, 12, 14, 16, 18]
6
7# cycle - 循环遍历
8cycler = itertools.cycle(['A', 'B', 'C'])
9print("Cycle:", [next(cycler) for _ in range(6)])  # ['A', 'B', 'C', 'A', 'B', 'C']
10
11# repeat - 重复元素
12repeater = itertools.repeat('hello', 3)
13print("Repeat:", list(repeater))  # ['hello', 'hello', 'hello']
14
15# 无限重复
16infinite = itertools.repeat('hello')
17print("Infinite:", [next(infinite) for _ in range(3)])  # ['hello', 'hello', 'hello']

注意事项：

• 无限迭代器需要配合 islice() 或手动计数使用
• 直接遍历会导致死循环
• 适合生成器模式和惰性计算

6.2.自定义无限迭代器

实现自定义无限迭代器，用于特定场景：

1class InfiniteNumbers:
2    """无限自然数迭代器"""
3    def __init__(self, start=0):
4        self.current = start
5
6    def __iter__(self):
7        return self
8
9    def __next__(self):
10        value = self.current
11        self.current += 1
12        return value
13
14# 使用示例
15numbers = InfiniteNumbers(5)
16print([next(numbers) for _ in range(7)])  # [5, 6, 7, 8, 9, 10, 11]
17
18# 配合 islice 使用
19from itertools import islice
20limited = islice(InfiniteNumbers(1), 10)
21print(list(limited))  # [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

应用场景：

• 生成唯一序列号
• 轮询任务
• 持续数据采集
• 模拟无限数据流

注意事项：

• 必须配合限制条件使用
• 避免直接遍历导致死循环
• 考虑内存使用情况

7.迭代器链式操作

链式操作将多个迭代器函数串联起来，形成数据处理流水线，具有惰性求值优势。

7.1.常见链式操作函数

• filter(func, iterable)：过滤满足条件的元素
• map(func, iterable)：对每个元素应用函数
• itertools.islice(iterable, n)：取前 n 个元素

7.2.链式操作示例

1import itertools
2
3def process_data_pipeline(data):
4    """数据处理流水线"""
5    # 1. 过滤正数
6    filtered = filter(lambda x: x > 0, data)
7    # 2. 计算平方
8    squared = map(lambda x: x ** 2, filtered)
9    # 3. 取前5个
10    limited = itertools.islice(squared, 5)
11    return limited
12
13# 示例数据
14data = [-3, -2, -1, 0, 1, 2, 3, 4, 5, 6]
15result = process_data_pipeline(data)
16print(list(result))  # [1, 4, 9, 16, 25]
17
18# 更简洁的写法
19result = itertools.islice(
20    map(lambda x: x ** 2, filter(lambda x: x > 0, data)), 5
21)
22print(list(result))  # [1, 4, 9, 16, 25]

优势：

• 内存效率：按需处理，不一次性加载所有数据
• 性能优化：只为需要的数据执行运算
• 代码简洁：易于组合复杂的数据处理逻辑
• 可读性强：流水线式的处理流程清晰明了

应用场景：

• 数据清洗和预处理
• 日志文件分析
• 流式数据处理
• 大数据集处理

8.迭代器工具函数

8.1.itertools 常用函数

itertools 模块提供了丰富的迭代器构造工具，支持流式数据处理：

• chain(*iterables)：连接多个可迭代对象
• compress(data, selectors)：根据布尔掩码选择元素
• dropwhile(predicate, iterable)：丢弃开头满足条件的元素
• takewhile(predicate, iterable)：获取开头满足条件的元素
• groupby(iterable, key=None)：按 key 分组（需要先排序）

1import itertools
2
3# chain - 连接多个迭代器
4chain_iter = itertools.chain([1, 2], [3, 4], [5, 6])
5print("Chain:", list(chain_iter))  # [1, 2, 3, 4, 5, 6]
6
7# compress - 条件过滤
8data = ['A', 'B', 'C', 'D']
9selectors = [1, 0, 1, 0]
10compress_iter = itertools.compress(data, selectors)
11print("Compress:", list(compress_iter))  # ['A', 'C']
12
13# dropwhile - 丢弃开头满足条件的元素
14drop_iter = itertools.dropwhile(lambda x: x < 5, [1, 2, 3, 4, 5, 6, 1, 2])
15print("Dropwhile:", list(drop_iter))  # [5, 6, 1, 2]
16
17# takewhile - 获取开头满足条件的元素
18take_iter = itertools.takewhile(lambda x: x < 5, [1, 2, 3, 4, 5, 6, 1, 2])
19print("Takewhile:", list(take_iter))  # [1, 2, 3, 4]
20
21# groupby - 分组（需要先排序）
22data = ['apple', 'animal', 'banana', 'bird', 'cherry', 'cat']
23sorted_data = sorted(data, key=lambda x: x[0])  # 按首字母排序
24grouped = itertools.groupby(sorted_data, key=lambda x: x[0])
25for key, group in grouped:
26    print(f"{key}: {list(group)}")
27# 输出:
28# a: ['animal', 'apple']
29# b: ['banana', 'bird']
30# c: ['cat', 'cherry']

优势：

• 惰性求值，内存效率高
• 函数式编程风格
• 易于组合复杂的数据处理逻辑
• 适合处理大数据集

9.迭代器性能优势

9.1.内存使用比较

1import sys
2
3def using_list_memory(n):
4    """使用列表，占用大量内存"""
5    return [i**2 for i in range(n)]
6
7def using_iterator_memory(n):
8    """使用迭代器，占用很少内存"""
9    return (i**2 for i in range(n))
10
11n = 1000000
12
13# 内存使用比较
14list_result = using_list_memory(n)
15iter_result = using_iterator_memory(n)
16
17print(f"列表内存使用: {sys.getsizeof(list_result)} 字节")
18print(f"迭代器内存使用: {sys.getsizeof(iter_result)} 字节")
19# 输出示例:
20# 列表内存使用: 800984 字节
21# 迭代器内存使用: 200 字节

9.2.性能优势总结

内存效率：

• 列表：一次性创建所有元素，占用大量内存
• 迭代器：按需生成元素，内存占用极小

处理速度：

• 列表：需要预先计算所有元素
• 迭代器：惰性求值，只计算需要的元素

适用场景：

• 大数据集处理
• 无限数据流
• 内存受限环境
• 流式数据处理

10.实际应用场景

10.1.数据库查询结果处理

1class DatabaseQueryIterator:
2    """数据库查询结果迭代器，支持分页处理"""
3
4    def __init__(self, query, chunk_size=1000):
5        self.query = query
6        self.chunk_size = chunk_size
7        self.current_chunk = []
8        self.current_index = 0
9        self.has_more = True
10
11    def __iter__(self):
12        return self
13
14    def __next__(self):
15        if self.current_index >= len(self.current_chunk):
16            if not self.has_more:
17                raise StopIteration
18            self._fetch_next_chunk()
19
20        if not self.current_chunk:
21            raise StopIteration
22
23        value = self.current_chunk[self.current_index]
24        self.current_index += 1
25        return value
26
27    def _fetch_next_chunk(self):
28        """获取下一批数据（模拟实现）"""
29        import random
30        if random.random() < 0.2:  # 20% 概率没有更多数据
31            self.has_more = False
32            self.current_chunk = []
33        else:
34            self.current_chunk = [f"record_{i}" for i in range(self.chunk_size)]
35        self.current_index = 0
36
37# 使用示例
38query_iter = DatabaseQueryIterator("SELECT * FROM large_table")
39for i, record in enumerate(query_iter):
40    if i >= 5000:  # 限制处理数量
41        break
42    if i % 1000 == 0:
43        print(f"处理第 {i} 条记录")

优势：

• 内存效率高，不会一次性加载所有数据
• 支持大数据集处理
• 可以随时停止处理
• 适合流式数据处理

10.2.流式数据处理

1class StreamProcessor:
2    """流式数据处理器"""
3
4    def __init__(self, data_stream):
5        self.data_stream = data_stream
6
7    def filter_valid(self):
8        """过滤有效数据"""
9        return filter(lambda x: x is not None and x != '', self.data_stream)
10
11    def transform_data(self):
12        """转换数据"""
13        return map(str.upper, self.filter_valid())
14
15    def batch_process(self, batch_size=100):
16        """批量处理"""
17        batch = []
18        for item in self.transform_data():
19            batch.append(item)
20            if len(batch) >= batch_size:
21                yield batch
22                batch = []
23        if batch:
24            yield batch
25
26    def process(self):
27        """处理流水线"""
28        total_processed = 0
29        for batch in self.batch_process(3):  # 小批量用于演示
30            total_processed += len(batch)
31            print(f"处理批次: {batch}")
32        print(f"总共处理: {total_processed} 条记录")
33
34def data_stream():
35    """模拟数据流"""
36    data = ['hello', '', 'world', None, 'python', 'stream', '', 'processing']
37    for item in data:
38        yield item
39
40# 使用示例
41processor = StreamProcessor(data_stream())
42processor.process()

优势：

• 内存效率高，适合处理大数据流
• 支持批量处理，提高处理效率
• 易于扩展和组合
• 支持实时数据处理

11.最佳实践和注意事项

11.1.最佳实践

1# 1. 使用迭代器处理大数据集
2def process_large_data_correct():
3    """正确处理大数据集"""
4    # 正确：使用迭代器
5    with open('large_file.txt', 'r') as file:
6        for line in file:  # 文件对象本身就是迭代器
7            process_line(line.strip())
8
9    # 错误：一次性读取所有内容
10    # with open('large_file.txt', 'r') as file:
11    #     lines = file.readlines()  # 可能内存溢出
12    #     for line in lines:
13    #         process_line(line)
14
15# 2. 合理使用 itertools
16def efficient_combinations():
17    """高效组合操作"""
18    import itertools
19
20    data = range(1000000)
21
22    # 高效：链式操作
23    result = itertools.islice(
24        filter(lambda x: x % 2 == 0, data),
25        100
26    )
27    return list(result)
28
29# 3. 资源管理
30class ResourceIterator:
31    """正确管理资源的迭代器"""
32    def __init__(self, resource):
33        self.resource = resource
34
35    def __iter__(self):
36        return self
37
38    def __next__(self):
39        line = self.resource.readline()
40        if line:
41            return line
42        else:
43            raise StopIteration
44
45    def __del__(self):
46        """确保资源被释放"""
47        if hasattr(self.resource, 'close'):
48            self.resource.close()
49
50# 更推荐的方式：使用 with 语句
51def safe_file_processing(filename):
52    """更安全的文件处理方式"""
53    try:
54        with open(filename, 'r') as file:
55            for line in file:
56                process_line(line.strip())
57    except FileNotFoundError:
58        print(f"文件 {filename} 不存在")
59    except Exception as e:
60        print(f"处理文件时出错: {e}")

最佳实践总结：

• 使用迭代器处理大数据集
• 合理使用 itertools 模块
• 及时关闭资源，优先使用 with 语句
• 避免在迭代过程中修改集合
• 使用 islice 限制无限迭代器

11.2.常见陷阱

1import itertools
2
3# 陷阱1: 迭代器只能使用一次
4numbers = [1, 2, 3]
5iterator = iter(numbers)
6print(list(iterator))  # [1, 2, 3]
7print(list(iterator))  # [] - 空的！
8
9# 陷阱2: 在迭代过程中修改集合
10numbers = [1, 2, 3, 4, 5]
11try:
12    for num in numbers:
13        if num == 3:
14            numbers.remove(num)  # 危险！
15except RuntimeError as e:
16    print(f"错误: {e}")
17
18# 陷阱3: 无限迭代器没有终止条件
19infinite = itertools.count()
20# for i in infinite:  # 这将永远运行
21#     print(i)
22
23# 正确做法：使用 islice 限制
24limited = itertools.islice(infinite, 5)
25print(list(limited))  # [0, 1, 2, 3, 4]
26
27# 陷阱4: 忘记处理 StopIteration 异常
28def manual_iteration():
29    numbers = [1, 2, 3]
30    iterator = iter(numbers)
31    while True:
32        try:
33            value = next(iterator)
34            print(value)
35        except StopIteration:
36            break
37
38# 陷阱5: 在生成器中使用 return 而不是 yield
39def wrong_generator():
40    for i in range(3):
41        return i  # 错误！应该使用 yield
42
43def correct_generator():
44    for i in range(3):
45        yield i  # 正确

避免陷阱的方法：

• 理解迭代器的单向性
• 避免在迭代过程中修改集合
• 使用 islice 限制无限迭代器
• 正确处理 StopIteration 异常
• 在生成器中使用 yield 而不是 return