Python中的函数式编程工具：itertools模块详解

描述：
Python的itertools模块是函数式编程的核心工具库，提供了高效、内存友好的迭代器构建块。这些工具专门用于处理迭代数据流，能够组合出强大的迭代器模式，避免创建大型中间数据结构，显著提升内存效率和执行性能。在数据处理、算法实现和流式计算中广泛应用。

解题过程循序渐进讲解：

1. itertools模块的设计哲学

• 模块中的所有函数都返回迭代器（iterator），采用惰性求值（lazy evaluation），只在需要时生成下一个元素
• 避免一次性加载所有数据到内存，适合处理大型或无限数据流
• 遵循"组合子"（combinator）设计模式，简单迭代器可以组合成复杂的数据管道

2. 无限迭代器（Infinite Iterators）
这些迭代器会无限生成值，必须结合其他限制条件使用：

import itertools

# count(start=0, step=1)：从start开始无限递增
counter = itertools.count(10, 2)  # 10, 12, 14, 16, ...
print(next(counter), next(counter))  # 10, 12

# cycle(iterable)：无限重复输入的可迭代对象
cycler = itertools.cycle('AB')
print([next(cycler) for _ in range(5)])  # ['A', 'B', 'A', 'B', 'A']

# repeat(object, times=None)：重复对象指定次数或无限重复
repeater = itertools.repeat('X', 3)
print(list(repeater))  # ['X', 'X', 'X']

3. 有限迭代器（Iterators Terminating on Shortest Input）
这些迭代器在最短输入耗尽时停止：

# accumulate(iterable, func=operator.add)：累积计算
data = [1, 2, 3, 4]
result = itertools.accumulate(data)  # 默认使用加法
print(list(result))  # [1, 3, 6, 10]

# chain(*iterables)：连接多个可迭代对象
chained = itertools.chain('ABC', 'DEF')
print(list(chained))  # ['A', 'B', 'C', 'D', 'E', 'F']

# chain.from_iterable(iterable)：从可迭代对象的可迭代对象中连接
nested = [['A', 'B'], ['C', 'D']]
flattened = itertools.chain.from_iterable(nested)
print(list(flattened))  # ['A', 'B', 'C', 'D']

# compress(data, selectors)：根据选择器筛选数据
data = ['A', 'B', 'C', 'D']
selectors = [1, 0, 1, 0]  # 1表示选择，0表示跳过
result = itertools.compress(data, selectors)
print(list(result))  # ['A', 'C']

# dropwhile(predicate, iterable)：跳过满足条件的元素，返回剩余所有
data = [1, 4, 6, 4, 1]
result = itertools.dropwhile(lambda x: x < 5, data)
print(list(result))  # [6, 4, 1]  # 从第一个不满足条件的开始全部保留

# takewhile(predicate, iterable)：返回满足条件的元素，遇到第一个不满足的停止
data = [1, 4, 6, 4, 1]
result = itertools.takewhile(lambda x: x < 5, data)
print(list(result))  # [1, 4]  # 遇到6时停止

# filterfalse(predicate, iterable)：返回不满足条件的元素
data = [1, 2, 3, 4, 5]
result = itertools.filterfalse(lambda x: x % 2, data)  # 过滤掉奇数
print(list(result))  # [2, 4]

# groupby(iterable, key=None)：按key函数分组连续相同元素
data = ['AA', 'AB', 'BA', 'BB', 'CA', 'CB']
# 注意：groupby只对连续相同元素分组，需要先排序
sorted_data = sorted(data, key=lambda x: x[0])  # 按首字母排序
for key, group in itertools.groupby(sorted_data, key=lambda x: x[0]):
    print(key, list(group))
# 输出：A ['AA', 'AB'], B ['BA', 'BB'], C ['CA', 'CB']

# islice(iterable, stop) 或 islice(iterable, start, stop, step)：切片迭代器
data = range(10)
result = itertools.islice(data, 2, 8, 2)  # 从索引2到8，步长为2
print(list(result))  # [2, 4, 6]

# starmap(function, iterable)：将可迭代对象的每个元素解包作为函数参数
pairs = [(2, 3), (4, 5)]
result = itertools.starmap(pow, pairs)  # pow(2,3), pow(4,5)
print(list(result))  # [8, 1024]

# tee(iterable, n=2)：复制迭代器为n个独立迭代器
original = iter([1, 2, 3])
a, b, c = itertools.tee(original, 3)  # 创建3个独立迭代器
print(list(a), list(b), list(c))  # 都是[1, 2, 3]

# zip_longest(*iterables, fillvalue=None)：以最长可迭代对象为准进行zip
a = [1, 2]
b = ['A', 'B', 'C']
result = itertools.zip_longest(a, b, fillvalue='-')
print(list(result))  # [(1, 'A'), (2, 'B'), ('-', 'C')]

4. 组合迭代器（Combinatoric Iterators）
用于排列组合计算：

# product(*iterables, repeat=1)：笛卡尔积
result = itertools.product('AB', '12')
print(list(result))  # [('A', '1'), ('A', '2'), ('B', '1'), ('B', '2')]

# 重复自身
result = itertools.product('AB', repeat=2)  # 相当于product('AB', 'AB')
print(list(result))  # [('A', 'A'), ('A', 'B'), ('B', 'A'), ('B', 'B')]

# permutations(iterable, r=None)：排列
result = itertools.permutations('ABC', 2)  # 从3个元素中取2个的排列
print(list(result))  # [('A', 'B'), ('A', 'C'), ('B', 'A'), ('B', 'C'), ('C', 'A'), ('C', 'B')]

# combinations(iterable, r)：组合（不考虑顺序）
result = itertools.combinations('ABC', 2)  # 从3个元素中取2个的组合
print(list(result))  # [('A', 'B'), ('A', 'C'), ('B', 'C')]

# combinations_with_replacement(iterable, r)：可重复的组合
result = itertools.combinations_with_replacement('AB', 3)
print(list(result))  # [('A', 'A', 'A'), ('A', 'A', 'B'), ('A', 'B', 'B'), ('B', 'B', 'B')]

5. 实际应用案例

# 案例1：滑动窗口计算移动平均值
def moving_average(data, window_size=3):
    it = iter(data)
    # 初始化第一个窗口
    window = list(itertools.islice(it, window_size))
    if len(window) < window_size:
        return
    yield sum(window) / window_size
    
    # 滑动窗口
    for value in it:
        window.pop(0)  # 移除最旧值
        window.append(value)  # 添加最新值
        yield sum(window) / window_size

data = [1, 2, 3, 4, 5, 6, 7]
print(list(moving_average(data, 3)))  # [2.0, 3.0, 4.0, 5.0, 6.0]

# 案例2：扁平化嵌套列表的惰性版本
def flatten_lazy(nested_iterable):
    for item in nested_iterable:
        if isinstance(item, (list, tuple)):
            yield from flatten_lazy(item)
        else:
            yield item

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

# 案例3：批次处理大数据流
def batched(iterable, batch_size):
    it = iter(iterable)
    while True:
        batch = list(itertools.islice(it, batch_size))
        if not batch:
            return
        yield batch

# 处理大型文件或数据库结果
data = range(1000)
for batch in batched(data, 100):
    process_batch(batch)  # 每次处理100个元素

6. 性能优势与内存效率对比

import itertools
import time
import sys

# 传统方法：创建完整列表
def squares_list(n):
    return [i**2 for i in range(n)]

# itertools方法：惰性生成
def squares_iter(n):
    return (i**2 for i in range(n))  # 生成器表达式

# 测试内存使用
n = 1000000
print("列表方法内存大小:", sys.getsizeof(squares_list(n)))  # 约8MB
print("生成器内存大小:", sys.getsizeof(squares_iter(n)))  # 约128字节

# 测试执行时间
start = time.time()
sum(squares_list(n))
print("列表方法时间:", time.time() - start)

start = time.time()
sum(squares_iter(n))
print("生成器方法时间:", time.time() - start)

7. 高级组合技巧

# 管道式数据处理
import operator

data = [1, 2, 3, 2, 1, 4, 5, 3, 2]

# 数据处理管道：去重 → 过滤 → 累加
pipeline = (
    data
    | (lambda it: itertools.groupby(sorted(it)))  # 去重
    | (lambda it: (key for key, _ in it))  # 提取key
    | (lambda it: itertools.filterfalse(lambda x: x < 2, it))  # 过滤小于2的值
    | (lambda it: itertools.accumulate(it, operator.mul))  # 累积乘积
)

# 自定义管道操作符
class Pipe:
    def __init__(self, iterable):
        self.iterable = iterable
    
    def __or__(self, transform):
        return Pipe(transform(self.iterable))
    
    def __iter__(self):
        return iter(self.iterable)

result = Pipe(data) | (lambda it: itertools.groupby(sorted(it))) \
                   | (lambda it: (key for key, _ in it)) \
                   | (lambda it: itertools.filterfalse(lambda x: x < 2, it)) \
                   | (lambda it: itertools.accumulate(it, operator.mul))

print(list(result))

关键要点总结：

1. itertools所有函数都返回迭代器，支持惰性求值，内存效率高
2. 无限迭代器（count, cycle, repeat）需要与限制条件结合使用
3. 组合迭代器（product, permutations, combinations）用于数学计算
4. 可以组合多个迭代器构建复杂的数据处理管道
5. 相比列表操作，在处理大数据时具有显著的内存和性能优势
6. 注意groupby只对连续相同元素分组，通常需要先排序