Python 与数据实践 · 第 11 章

迭代协议与生成器

for 循环你从第 02 章用到现在，早就成了肌肉记忆。但你有没有追问过一句：for 凭什么知道"下一个值"是谁？凭什么列表、字符串、字典、文件都能往 for 里塞，一个数字却会被当场拒绝？这一章我们就把 for 的机箱…

约 24 分钟9,589 字30 个代码示例

for 循环你从第 02 章用到现在，早就成了肌肉记忆。但你有没有追问过一句：for 凭什么知道"下一个值"是谁？凭什么列表、字符串、字典、文件都能往 for 里塞，一个数字却会被当场拒绝？这一章我们就把 for 的机箱拆开——里面是一套叫迭代协议的约定。吃透它，你不仅能造出自己"可被 for 的对象"，还能解锁 Python 里最省内存的取值姿势：生成器。

1. 可迭代对象：for 眼里的合格供应商

先立规矩：

一个对象只要实现了 __iter__() 方法，并且这个方法返回一个迭代器，它就是可迭代对象（Iterable）。

你叫得出名字的容器——list、tuple、str、dict、set——全都签了这份合约。

动手验证一下，iter() 这个内置函数能从可迭代对象身上"领"到一个迭代器：

cart = ["键盘", "鼠标", "显示器"]

# iter() 等价于 cart.__iter__()，拿到一个迭代器
it = iter(cart)
print(type(it))   # <class 'list_iterator'>

# next() 等价于 it.__next__()，逐个把值"吐"出来
print(next(it))   # 键盘
print(next(it))   # 鼠标
print(next(it))   # 显示器

# print(next(it))  # 再要就没了：抛 StopIteration

顺手记住一条推理：迭代器自己也实现了 __iter__（约定俗成返回自身），所以迭代器必然是可迭代对象；但反过来，可迭代对象不一定是迭代器。

2. 迭代器：真正吐值的角色

把上面的推理摊开，就是迭代器的完整定义——同时实现了 __iter__ 和 __next__ 两个方法的对象。骨架长这样：

class ValueEmitter:
    """迭代器骨架：两个方法缺一不可"""

    def __iter__(self):
        """
        要求：返回一个迭代器。
        绝大多数情况下，把自己 return 出去就是标准答案
        """
        return self

    def __next__(self):
        """交出下一个值；实在没了就抛 StopIteration"""
        pass


obj = ValueEmitter()  # obj 就是一个迭代器

iter() 和 next() 这两个内置函数，不过是这两个魔术方法（第 08 章的老朋友）的马甲而已。

应用：造一条停不下来的序列

容器得先把所有值装进内存才敢开工，迭代器没有这个包袱——它算一步、给一步。比如培养皿里的细胞，每观察一次数量翻一倍：

class CellCulture:
    """细胞分裂迭代器：数量无限翻倍，没有尽头"""

    def __init__(self):
        self.count = 1

    def __iter__(self):
        return self

    def __next__(self):
        observed = self.count
        self.count *= 2
        return observed


dish = CellCulture()

print(next(dish))  # 1
print(next(dish))  # 2
print(next(dish))  # 4
print(next(dish))  # 8
print(next(dish))  # 16

这条序列没有终点，但谁需要谁取——"无限"在惰性的世界里根本不值钱。

3. 各司其职：可迭代对象配一个专职迭代器

第 1 节那条协议说 __iter__ 要返回迭代器，最讲究的写法是拆成两个类：可迭代对象管"能被遍历"这件事，迭代器管"遍历到哪儿了"这件事。看一部从 6 层往 1 层降的电梯：

class ElevatorDescent:
    """可迭代对象：电梯从 top 层一路降到 1 层"""

    def __init__(self, top):
        self.top = top

    def __iter__(self):
        # 每次调用都发一个崭新的迭代器
        return FloorStepper(self.top)


class FloorStepper:
    """专职迭代器：记住电梯当前停在哪层"""

    def __init__(self, top):
        self.floor = top

    def __iter__(self):
        return self

    def __next__(self):
        if self.floor < 1:
            raise StopIteration  # 1 层以下没楼层可给了
        current = self.floor
        self.floor -= 1
        return current


lift = ElevatorDescent(6)

step = iter(lift)
print(next(step))  # 6
print(next(step))  # 5
print(next(step))  # 4
print(next(step))  # 3
print(next(step))  # 2
print(next(step))  # 1
# print(next(step))  # StopIteration!

为什么非要拆开？玄机在 ElevatorDescent.__iter__ 每次都 new 一个新的 FloorStepper：进度状态存在迭代器身上，不污染可迭代对象本身，所以同一个对象能一遍又一遍地从头遍历。列表能 for 两遍，靠的就是这个机制。

4. 隐形的消费者：谁在替你调 iter 和 next

除了 for，还有一大票语法和内置函数在背后偷偷走"领迭代器 → 反复取值"的流程。

4.1 `for` 循环

# for 的内部剧本：先 iter() 领迭代器，再不停 next()，直到收到 StopIteration
for f in ElevatorDescent(3):
    print(f)  # 3 2 1

4.2 `list()`、`tuple()`、`set()` 构造函数

print(list(ElevatorDescent(3)))   # [3, 2, 1]
print(tuple(ElevatorDescent(3)))  # (3, 2, 1)
print(set(ElevatorDescent(3)))    # {1, 2, 3}

4.3 `*` 解包

top, *below = ElevatorDescent(3)
print(top)    # 3
print(below)  # [2, 1]

floors = [*ElevatorDescent(3)]
print(floors)  # [3, 2, 1]

4.4 `in` 成员判断

print(6 in ElevatorDescent(3))  # False
print(2 in ElevatorDescent(3))  # True

4.5 `sum()`、`max()`、`min()`

print(sum(ElevatorDescent(3)))  # 6
print(max(ElevatorDescent(3)))  # 3
print(min(ElevatorDescent(3)))  # 1

4.6 `zip()`、`map()`、`filter()`

# zip：把多个可迭代对象按位置配成对
goods = ["口罩", "酒精", "创可贴"]
prices = [12, 35, 8]
for good, price in zip(goods, prices):
    print(f"{good}: {price} 元")
# 口罩: 12 元
# 酒精: 35 元
# 创可贴: 8 元

# map：对每个值做一次转换
doubled = map(lambda n: n * 2, ElevatorDescent(3))
print(list(doubled))  # [6, 4, 2]

# filter：按条件筛掉不要的
odds = filter(lambda n: n % 2 == 1, ElevatorDescent(3))
print(list(odds))  # [3, 1]

4.7 `any()`、`all()`

floors = ElevatorDescent(3)
print(any(floors))  # True（只要有一个值为真，any 就是 True）
print(all(floors))  # True（3、2、1 全是真值）

print(all([3, 2, 1, 0]))  # False（混进一个 0，all 立刻翻脸）

5. range：天天用，未必真懂

range() 大概是你最常打交道的可迭代对象，专门负责生产整数序列。三种参数形态：

# range(stop)：从 0 数到 stop-1
for i in range(4):
    print(i, end=" ")  # 0 1 2 3
print()

# range(start, stop)：从 start 数到 stop-1
for i in range(3, 7):
    print(i, end=" ")  # 3 4 5 6
print()

# range(start, stop, step)：按步长跳着数
for i in range(1, 10, 3):
    print(i, end=" ")  # 1 4 7
print()

# 步长为负：倒着数
for i in range(4, 0, -1):
    print(i, end=" ")  # 4 3 2 1

三条重要特性，逐条说清：

特性一：惰性求值。 range 从不预先生成一长串数字，用到哪个才算哪个。而且和列表不同，它支持索引、切片和长度查询：

r = range(0, 100, 2)

print(len(r))    # 50
print(r[4])      # 8
print(r[2:6])    # range(4, 12, 2)
print(8 in r)    # True
print(9 in r)    # False

特性二：不是迭代器。 range 是可迭代对象，但不亲自当迭代器——每次 iter() 都发新的，于是可以反复遍历：

r = range(3)

for i in r:
    print(i, end=" ")  # 0 1 2
print()

for i in r:
    print(i, end=" ")  # 0 1 2（再来一遍毫无压力）

特性三：内存恒定。 range 内部只保存 start、stop、step 三个数，区间再大也只占这点地方——range(10**10) 随手就建，换成列表试试？

6. 推导式：一行代码顶一小段循环

推导式（Comprehension）是一种紧凑语法：从已有的可迭代对象出发，一步到位造出新容器。

6.1 列表推导式

# 基本款：[表达式 for 变量 in 可迭代对象]
triples = [n * 3 for n in range(5)]
print(triples)  # [0, 3, 6, 9, 12]

# 带筛选：[表达式 for 变量 in 可迭代对象 if 条件]
multiples_of_5 = [n for n in range(30) if n % 5 == 0]
print(multiples_of_5)  # [0, 5, 10, 15, 20, 25]

# if-else 放在表达式前部（注意：这种写法没有"尾巴上的 if"）
scores = [92, 58, 77, 61]
verdicts = ["通过" if s >= 60 else "不通过" for s in scores]
print(verdicts)  # ['通过', '不通过', '通过', '通过']

6.2 字典推导式

# 基本款：{键表达式: 值表达式 for 变量 in 可迭代对象}
cube_map = {n: n ** 3 for n in range(5)}
print(cube_map)  # {0: 0, 1: 1, 2: 8, 3: 27, 4: 64}

# 带筛选：只保留 3 的倍数
by_three = {n: n ** 3 for n in range(12) if n % 3 == 0}
print(by_three)  # {0: 0, 3: 27, 6: 216, 9: 729}

6.3 集合推导式

# 基本款：{表达式 for 变量 in 可迭代对象}（花括号里没有冒号）
multiples_of_4 = {n for n in range(20) if n % 4 == 0}
print(multiples_of_4)  # {0, 4, 8, 12, 16}

# 集合自动去重：四个单词只留下三个首字母
words = ["apple", "avocado", "banana", "cherry"]
initials = {w[0] for w in words}
print(initials)  # {'a', 'b', 'c'}

6.4 嵌套推导式

# 把二维名单拍扁成一维
roster = [["张三", "李四"], ["欧阳锋"], ["王五", "赵六"]]
names = [name for group in roster for name in group]
print(names)  # ['张三', '李四', '欧阳锋', '王五', '赵六']

# 等价于：
# names = []
# for group in roster:
#     for name in group:
#         names.append(name)

# 嵌套 + 筛选：只留两个字的名字
short = [name for group in roster for name in group if len(name) == 2]
print(short)  # ['张三', '李四', '王五', '赵六']

读嵌套推导式有个窍门：从左往右读，for 出现的顺序就等于普通循环从外到内的顺序。

6.5 没有元组推导式这回事

Python 里不存在元组推导式。你用圆括号写出来的那个东西，是下一节的主角——生成器表达式：

result = (n * 2 for n in range(5))
print(type(result))  # <class 'generator'>，不是 tuple

6.6 推导式与循环怎么选

# 推导式写法（更 Pythonic，通常也更快）
cubes = [n ** 3 for n in range(8)]

# 等价的循环写法
cubes = []
for n in range(8):
    cubes.append(n ** 3)

推导式短小精悍，解释器对它还有专门优化。但别走火入魔：一旦逻辑复杂起来（多层嵌套加多重条件），硬压成一行只会变成没人敢碰的天书——那种时候，老老实实写循环才是对读者负责。

7. 生成器函数：会踩刹车的函数

普通函数一路狂奔，撞到 return 交出结果就散场。但只要函数体内出现一个 yield 关键字，这个函数的性质就彻底变了——调用它不执行任何代码，只生产一个生成器对象；而生成器本身就是一个迭代器。

# 函数体里有 yield，Python 就把它登记为生成器函数
def game_levels():
    print("载入草原关卡")
    yield "草原"
    print("载入沙漠关卡")
    yield "沙漠"
    print("载入火山关卡")
    yield "火山"


quest = game_levels()  # 注意：此刻函数体一行都没跑，只拿到了生成器
print(type(quest))     # <class 'generator'>

print(next(quest))  # 载入草原关卡（开始执行，到第一个 yield 踩下刹车）
                    # 草原
print(next(quest))  # 载入沙漠关卡（从上次的刹车点无缝续跑）
                    # 沙漠
print(next(quest))  # 载入火山关卡
                    # 火山
# print(next(quest))  # 函数体跑完，抛 StopIteration

生成器是编写迭代器的捷径。第 3 节那个电梯，两个类几十行，换成生成器只要几行：

def descend(top):
    """生成器版本：从 top 层降到 1 层"""
    while top >= 1:
        yield top
        top -= 1


lift = descend(5)
print(type(lift))     # <class 'generator'>

for f in lift:
    print(f, end=" ")  # 5 4 3 2 1

生成器有三个鲜明特点：

惰性求值：值按需生产，用一个算一个，内存开销可以忽略
现场封存：每次 yield 处暂停，局部变量和执行位置原样冻结，下次调用继续接着演
一遍生命：跟迭代器一样，取完即止，没法倒带重放

8. yield from：把产量外包出去

生成器内部可以调用别的生成器，把人家的产出逐个转发：

def opening_ceremony():
    yield "运动员入场"
    yield "点燃火炬"


def sports_day():
    yield "运动会开始"
    for moment in opening_ceremony():
        yield moment
    yield "仪式结束，比赛开始"


for moment in sports_day():
    print(moment)
# 运动会开始
# 运动员入场
# 点燃火炬
# 仪式结束，比赛开始

中间那个 for 纯属搬运工体力活，Python 给了个语法糖 yield from，一步完成"委托"：

def sports_day():
    yield "运动会开始"
    yield from opening_ceremony()  # 中间环节整体转交子生成器
    yield "仪式结束，比赛开始"

9. send：不仅会吐，还会收

yield 不只是出料口，同时也是进料口。调用生成器的 send(值) 方法，这个值会成为 yield 表达式的运算结果，被函数体接住：

def checkout():
    """收银台：不停接收商品价格，随时回报累计账单"""
    bill = 0
    while True:
        price = yield bill  # 吐出当前账单，同时等待下一件商品的价格
        if price is None:
            break
        bill += price


counter = checkout()
print(next(counter))      # 0（先点火一次，让生成器走到 yield 处待命）
print(counter.send(15))   # 15（收到 15 元，累计后吐出新账单）
print(counter.send(42))   # 57
print(counter.send(8))    # 65

有个细节必须记住：第一次不能直接 send 具体值——生成器还没跑到 yield 那一行，送进去的值没人接，会直接报错。先 next()（或 send(None)）把它点火。

10. 生成器表达式：换一对括号，立刻变懒

把列表推导式的方括号换成圆括号，得到的不是元组，而是生成器表达式——一次性把"算"和"存"分开，只算不存：

# 惰性版本：内存里只有一个生成器，两百万个值一个都没提前算
lazy_squares = (n * n for n in range(2_000_000))

print(type(lazy_squares))  # <class 'generator'>

# 要一个，算一个：
print(next(lazy_squares))  # 0
print(next(lazy_squares))  # 1
print(next(lazy_squares))  # 4

11. itertools：标准库自带的迭代工具箱

itertools 模块收录了一大批现成的高效迭代器，认识几个最常用的：

import itertools

# count(起点, 步长)：从起点出发，无限往下数
seat_no = itertools.count(7, 3)
print(next(seat_no))  # 7
print(next(seat_no))  # 10
print(next(seat_no))  # 13

# cycle(可迭代对象)：循环播放，永不停止
light = itertools.cycle(["红灯", "绿灯", "黄灯"])
print(next(light))  # 红灯
print(next(light))  # 绿灯
print(next(light))  # 黄灯
print(next(light))  # 红灯（又转回来了）

# repeat(值, 次数)：把同一个值连给 N 遍
padding = list(itertools.repeat("待定", 3))
print(padding)  # ['待定', '待定', '待定']

# chain(*可迭代对象)：几段序列首尾相接
merged = list(itertools.chain([7, 8], [9, 10], [11]))
print(merged)  # [7, 8, 9, 10, 11]

# islice(可迭代对象, 起点, 结束, 步长)：给无限序列"切片"
head = list(itertools.islice(itertools.count(), 6))
print(head)  # [0, 1, 2, 3, 4, 5]

# permutations(可迭代对象, r)：排列（讲究先后顺序）
podium = list(itertools.permutations(["甲", "乙", "丙"], 2))
print(podium)
# [('甲', '乙'), ('甲', '丙'), ('乙', '甲'), ('乙', '丙'), ('丙', '甲'), ('丙', '乙')]

# combinations(可迭代对象, r)：组合（不分先后）
matchups = list(itertools.combinations(["A", "B", "C", "D"], 2))
print(matchups)
# [('A', 'B'), ('A', 'C'), ('A', 'D'), ('B', 'C'), ('B', 'D'), ('C', 'D')]

12. 作业

动笔之前，先用费曼学习法把三样东西讲给自己听：迭代器、可迭代对象、生成器各是什么，彼此什么关系。讲卡壳的地方，就是需要回头补的地方。

一、多层储物清单摊平

编写生成器函数 flatten_boxes，把任意深度嵌套的列表摊平成一层：

def flatten_boxes(boxes):
    # 你的代码
    pass


storage = ["拖把", ["抹布", ["扫帚", "垃圾袋"], "水桶"], "梯子", ["灯泡", "灯管"]]
print(list(flatten_boxes(storage)))
# ['拖把', '抹布', '扫帚', '垃圾袋', '水桶', '梯子', '灯泡', '灯管']

二、模拟分批拉取记录

编写一个生成器，模拟从接口分批拉取数据：

def fetch_batches(total, batch_size):
    """
    模拟分批拉取
    total: 记录总数
    batch_size: 每批条数
    每轮 yield 一批记录（列表），记录编号从 1 开始
    """
    # 你的代码
    pass


for batch in fetch_batches(25, 10):
    print(batch)
# [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# [11, 12, 13, 14, 15, 16, 17, 18, 19, 20]
# [21, 22, 23, 24, 25]

三、思考题

下面代码的输出是什么？为什么？

def tea_pot():
    print("第一泡，出汤")
    yield 1
    print("第二泡，出汤")
    yield 2
    print("第三泡，出汤")
    yield 3
    print("茶叶已泡乏")


pot = tea_pot()
print("茶具已备好")
print(next(pot))
print("---")
print(next(pot))
print("---")
pot.close()
print("茶具已收拾")
print(next(pot))

下一章：12 · with 与上下文协议