Python itertools 高级模式itertools 提供高效的迭代器工具, 所有函数均返回迭代器——惰性求值, 内存友好.1. 无限迭代器: cycle / islice / tee-------------------------------------from itertools import cycle, islice, tee# cycle 无限循环可迭代对象counter 0for item in cycle([红, 黄, 绿]):print(循环信号灯:, item)counter 1if counter 5:break# islice 对迭代器做切片 (惰性)# islice(iterable, start, stop, step)data range(100)subset islice(data, 10, 30, 2) # 索引10到30, 步长2print(islice 切片结果:, list(subset))# tee 将一个迭代器克隆为 N 个独立副本src range(5)a, b, c tee(src, 3) # 克隆出3个独立迭代器print(tee 迭代器a:, list(a)) # [0,1,2,3,4]print(tee 迭代器b:, list(b)) # [0,1,2,3,4]print(tee 迭代器c:, list(c)) # [0,1,2,3,4]# 注意: tee 会缓存数据, 如果原始迭代器很大且多个副本进度差异大, 会消耗内存2. 组合数学: combinations / permutations / product-----------------------------------------------------from itertools import combinations, permutations, product# combinations: 无放回组合, 元素不重复, 顺序无关items [A, B, C]print(组合(2个):, list(combinations(items, 2)))# [(A,B), (A,C), (B,C)]# permutations: 排列, 顺序有关, 元素不重复print(排列(2个):, list(permutations(items, 2)))# [(A,B), (A,C), (B,A), (B,C), (C,A), (C,B)]# product: 笛卡尔积, 相当于嵌套循环, repeat 参数允许重复print(笛卡尔积:, list(product([0, 1], repeat2)))# [(0,0), (0,1), (1,0), (1,1)]# 实战: 枚举所有可能的密码组合digits 0123456789# product(digits, repeat4) 生成所有4位数字密码for pin in islice(product(digits, repeat4), 3):print(密码组合:, .join(pin))3. groupby — 分组聚合-----------------------groupby 要求输入已排序, 只对连续相同键分组.from itertools import groupbydata [(北京, 100), (北京, 200), (上海, 150), (上海, 300), (广州, 80)]# 按城市分组前必须先排序sorted_data sorted(data, keylambda x: x[0])result {}for city, group in groupby(sorted_data, keylambda x: x[0]):values [item[1] for item in group]result[city] {总量: sum(values), 次数: len(values), 平均: sum(values)/len(values)}print(groupby 分组统计:, result)# groupby 返回的 group 是迭代器, 需要遍历前消费4. accumulate — 累计运算--------------------------from itertools import accumulateimport operatornums [1, 2, 3, 4, 5]# 默认累加print(累计求和:, list(accumulate(nums))) # [1, 3, 6, 10, 15]# 累计乘积print(累计乘积:, list(accumulate(nums, operator.mul))) # [1, 2, 6, 24, 120]# 自定义函数: 累计最大值print(累计最大值:, list(accumulate(nums, max))) # [1, 2, 3, 4, 5]# 斐波那契数列def fib():yield 0yield 1yield from accumulate(range(2, 20), lambda a, b: a b)# 注意: accumulate 的 func 接收 (上一次结果, 当前元素)5. chain / chain.from_iterable — 展平-----------------------------------------from itertools import chain# chain 连接多个可迭代对象lists [[1, 2], [3, 4], [5]]flattened chain.from_iterable(lists) # 等价于 chain(*lists)print(展平列表:, list(flattened)) # [1, 2, 3, 4, 5]# chain 可混合不同类型mixed chain(ABC, [1, 2], (3,))print(混合类型连接:, list(mixed)) # [A,B,C,1,2,3]6. zip_longest — 最长拉链--------------------------from itertools import zip_longestnames [Alice, Bob, Charlie]scores [85, 92]# 默认 zip 会在最短处截断, zip_longest 用 fillvalue 填充zipped zip_longest(names, scores, fillvalue缺考)print(zip_longest 结果:, list(zipped))# [(Alice, 85), (Bob, 92), (Charlie, 缺考)]7. takewhile / dropwhile — 条件筛选--------------------------------------from itertools import takewhile, dropwhilenums [2, 4, 6, 7, 8, 9, 10]# takewhile: 拿取满足条件的元素, 遇到第一个不满足就停止print(takewhile 偶数:, list(takewhile(lambda x: x % 2 0, nums)))# [2, 4, 6]# dropwhile: 丢弃满足条件的元素, 遇到第一个不满足就保留后续全部print(dropwhile 偶数:, list(dropwhile(lambda x: x % 2 0, nums)))# [7, 8, 9, 10]8. pairwise (Python 3.10) / batched (Python 3.12)------------------------------------------------------from itertools import pairwise, batched# pairwise: 返回连续重叠对 (s - (s0,s1), (s1,s2), ...)items [1, 2, 3, 4, 5]print(pairwise 相邻对:, list(pairwise(items)))# [(1,2), (2,3), (3,4), (4,5)]# 应用: 检查序列是否单调递增print(是否单调递增:, all(a b for a, b in pairwise(items))) # True# batched: 将迭代器分成指定大小的块data_batches list(batched(range(10), 3))print(batched 分块:, data_batches)# [(0,1,2), (3,4,5), (6,7,8), (9,)]9. 自定义迭代器链式模式-------------------------def custom_iterator_chain(data):链式组合多个 itertools 操作from itertools import filterfalse, compress, starmap# filterfalse: 保留不满足条件的元素return (data# 这里用函数组合展示模式, 实际需顺序调用)# 典型流水线模式def pipeline_example(nums):数据处理流水线: 过滤 - 变换 - 分组 - 聚合# 过滤正数positives filter(lambda x: x 0, nums)# 映射平方squares map(lambda x: x ** 2, positives)# 按奇偶分组sorted_squares sorted(squares, keylambda x: x % 2)grouped groupby(sorted_squares, keylambda x: x % 2)result {}for is_even, group in grouped:result[偶数 if is_even 0 else 奇数] list(group)return resultprint(流水线结果:, pipeline_example([-3, -2, 1, 2, 3, 4]))# compress: 按选择器筛选data [a, b, c, d]selectors [1, 0, 1, 0]print(compress 筛选:, list(compress(data, selectors))) # [a, c]# starmap: 用 * 解包参数pairs [(2, 3), (4, 5), (6, 7)]print(starmap 求和:, list(starmap(lambda a, b: a b, pairs)))# [5, 9, 13]总结: itertools 函数组合可构建声明式数据管道, 避免中间列表, 代码更简洁高效.
Pythonitertools高级模式
发布时间：2026/5/29 23:48:26
Python itertools 高级模式itertools 提供高效的迭代器工具, 所有函数均返回迭代器——惰性求值, 内存友好.1. 无限迭代器: cycle / islice / tee-------------------------------------from itertools import cycle, islice, tee# cycle 无限循环可迭代对象counter 0for item in cycle([红, 黄, 绿]):print(循环信号灯:, item)counter 1if counter 5:break# islice 对迭代器做切片 (惰性)# islice(iterable, start, stop, step)data range(100)subset islice(data, 10, 30, 2) # 索引10到30, 步长2print(islice 切片结果:, list(subset))# tee 将一个迭代器克隆为 N 个独立副本src range(5)a, b, c tee(src, 3) # 克隆出3个独立迭代器print(tee 迭代器a:, list(a)) # [0,1,2,3,4]print(tee 迭代器b:, list(b)) # [0,1,2,3,4]print(tee 迭代器c:, list(c)) # [0,1,2,3,4]# 注意: tee 会缓存数据, 如果原始迭代器很大且多个副本进度差异大, 会消耗内存2. 组合数学: combinations / permutations / product-----------------------------------------------------from itertools import combinations, permutations, product# combinations: 无放回组合, 元素不重复, 顺序无关items [A, B, C]print(组合(2个):, list(combinations(items, 2)))# [(A,B), (A,C), (B,C)]# permutations: 排列, 顺序有关, 元素不重复print(排列(2个):, list(permutations(items, 2)))# [(A,B), (A,C), (B,A), (B,C), (C,A), (C,B)]# product: 笛卡尔积, 相当于嵌套循环, repeat 参数允许重复print(笛卡尔积:, list(product([0, 1], repeat2)))# [(0,0), (0,1), (1,0), (1,1)]# 实战: 枚举所有可能的密码组合digits 0123456789# product(digits, repeat4) 生成所有4位数字密码for pin in islice(product(digits, repeat4), 3):print(密码组合:, .join(pin))3. groupby — 分组聚合-----------------------groupby 要求输入已排序, 只对连续相同键分组.from itertools import groupbydata [(北京, 100), (北京, 200), (上海, 150), (上海, 300), (广州, 80)]# 按城市分组前必须先排序sorted_data sorted(data, keylambda x: x[0])result {}for city, group in groupby(sorted_data, keylambda x: x[0]):values [item[1] for item in group]result[city] {总量: sum(values), 次数: len(values), 平均: sum(values)/len(values)}print(groupby 分组统计:, result)# groupby 返回的 group 是迭代器, 需要遍历前消费4. accumulate — 累计运算--------------------------from itertools import accumulateimport operatornums [1, 2, 3, 4, 5]# 默认累加print(累计求和:, list(accumulate(nums))) # [1, 3, 6, 10, 15]# 累计乘积print(累计乘积:, list(accumulate(nums, operator.mul))) # [1, 2, 6, 24, 120]# 自定义函数: 累计最大值print(累计最大值:, list(accumulate(nums, max))) # [1, 2, 3, 4, 5]# 斐波那契数列def fib():yield 0yield 1yield from accumulate(range(2, 20), lambda a, b: a b)# 注意: accumulate 的 func 接收 (上一次结果, 当前元素)5. chain / chain.from_iterable — 展平-----------------------------------------from itertools import chain# chain 连接多个可迭代对象lists [[1, 2], [3, 4], [5]]flattened chain.from_iterable(lists) # 等价于 chain(*lists)print(展平列表:, list(flattened)) # [1, 2, 3, 4, 5]# chain 可混合不同类型mixed chain(ABC, [1, 2], (3,))print(混合类型连接:, list(mixed)) # [A,B,C,1,2,3]6. zip_longest — 最长拉链--------------------------from itertools import zip_longestnames [Alice, Bob, Charlie]scores [85, 92]# 默认 zip 会在最短处截断, zip_longest 用 fillvalue 填充zipped zip_longest(names, scores, fillvalue缺考)print(zip_longest 结果:, list(zipped))# [(Alice, 85), (Bob, 92), (Charlie, 缺考)]7. takewhile / dropwhile — 条件筛选--------------------------------------from itertools import takewhile, dropwhilenums [2, 4, 6, 7, 8, 9, 10]# takewhile: 拿取满足条件的元素, 遇到第一个不满足就停止print(takewhile 偶数:, list(takewhile(lambda x: x % 2 0, nums)))# [2, 4, 6]# dropwhile: 丢弃满足条件的元素, 遇到第一个不满足就保留后续全部print(dropwhile 偶数:, list(dropwhile(lambda x: x % 2 0, nums)))# [7, 8, 9, 10]8. pairwise (Python 3.10) / batched (Python 3.12)------------------------------------------------------from itertools import pairwise, batched# pairwise: 返回连续重叠对 (s - (s0,s1), (s1,s2), ...)items [1, 2, 3, 4, 5]print(pairwise 相邻对:, list(pairwise(items)))# [(1,2), (2,3), (3,4), (4,5)]# 应用: 检查序列是否单调递增print(是否单调递增:, all(a b for a, b in pairwise(items))) # True# batched: 将迭代器分成指定大小的块data_batches list(batched(range(10), 3))print(batched 分块:, data_batches)# [(0,1,2), (3,4,5), (6,7,8), (9,)]9. 自定义迭代器链式模式-------------------------def custom_iterator_chain(data):链式组合多个 itertools 操作from itertools import filterfalse, compress, starmap# filterfalse: 保留不满足条件的元素return (data# 这里用函数组合展示模式, 实际需顺序调用)# 典型流水线模式def pipeline_example(nums):数据处理流水线: 过滤 - 变换 - 分组 - 聚合# 过滤正数positives filter(lambda x: x 0, nums)# 映射平方squares map(lambda x: x ** 2, positives)# 按奇偶分组sorted_squares sorted(squares, keylambda x: x % 2)grouped groupby(sorted_squares, keylambda x: x % 2)result {}for is_even, group in grouped:result[偶数 if is_even 0 else 奇数] list(group)return resultprint(流水线结果:, pipeline_example([-3, -2, 1, 2, 3, 4]))# compress: 按选择器筛选data [a, b, c, d]selectors [1, 0, 1, 0]print(compress 筛选:, list(compress(data, selectors))) # [a, c]# starmap: 用 * 解包参数pairs [(2, 3), (4, 5), (6, 7)]print(starmap 求和:, list(starmap(lambda a, b: a b, pairs)))# [5, 9, 13]总结: itertools 函数组合可构建声明式数据管道, 避免中间列表, 代码更简洁高效.
