Python 基础体系 · 第 15/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python 可迭代对象与迭代器：协议、惰性、耗尽和组合

Python 中的 for 循环、推导式、生成器表达式、map()、zip()、itertools 以及文件逐行读取，都建立在同一组机制上：可迭代对象（iterable）提供数据来源，迭代器（iterator）维护遍历状态，__next__() 一次产生一个值，StopIteration 表示数据流结束。

如果只把“可迭代对象”理解成“可以放进 for 循环的对象”，就很容易忽略几个决定程序行为的重要事实：

• 可迭代对象不一定是迭代器；
• 迭代器通常是有状态、单向、一次性的；
• 惰性并不等于不消耗，也不等于低成本；
• 一个迭代器被某个消费者读取后，其他消费者看到的内容会发生变化；
• 组合迭代器时，输入是否被预先物化、是否共享状态，直接决定内存占用和结果语义。

本文以 Python 3.14 的同步迭代协议为主线，逐步说明这些机制如何连接起来，并在最后讨论生成器和 itertools 中最容易被忽略的边界。

---

一、先区分四个概念：序列、容器、可迭代对象和迭代器

1. 序列不是可迭代对象的全部

**序列（sequence）**是具有顺序、通常可以通过非负整数索引访问的一类对象，例如：

items = ["a", "b", "c"]

print(items[0])  # a
print(len(items))  # 3

列表、元组、字符串、字节串都是典型序列。序列强调的是：

1. 元素有顺序；
2. 可以通过下标定位；
3. 通常可以通过切片取得子序列；
4. 通常可以重复遍历。

但是，Python 的可迭代对象远不止序列。例如：

data = {10, 20, 30}
mapping = {"name": "Ada", "age": 36}
file_obj = open("data.txt", encoding="utf-8")

集合、字典、文件对象都可以参与迭代，但它们的访问语义不同：

for value in data:
    print(value)

for key in mapping:
    print(key)

for line in file_obj:
    print(line)

字典默认迭代的是键，而不是键值对；文件对象迭代的是文本行。可迭代性描述的是“能否逐项提供值”，并不要求对象支持索引、切片或长度。

Python 数据模型将序列、集合、映射等容器分别定义，但这些容器通常通过 __iter__() 提供高效迭代能力。(docs.python.org)

2. 可迭代对象是一个协议概念

**可迭代对象（iterable）**是可以被 iter(obj) 转换成迭代器的对象。

形式化地说，对象 x 满足可迭代协议，可以理解为：

$$\operatorname{iter}(x) \rightarrow i$$

其中：

• x 是数据来源；
• i 是迭代器；
• 后续通过 next(i) 逐项取得数据。

通常，一个类通过定义：

def __iter__(self):
    ...

来实现可迭代协议。__iter__() 应该返回一个迭代器，而不是直接返回单个元素。

例如：

class Numbers:
    def __init__(self, values):
        self.values = values

    def __iter__(self):
        return iter(self.values)


numbers = Numbers([10, 20, 30])

print(list(numbers))
# [10, 20, 30]

这里的 Numbers 是可迭代对象，但它本身不负责维护当前遍历位置。每次调用 iter(numbers)，它都从内部列表创建一个新的列表迭代器。

3. 迭代器是带状态的可迭代对象

**迭代器（iterator）**不仅能被迭代，还能通过 __next__() 产生下一个元素。

它满足两个核心条件：

class Iterator:
    def __iter__(self):
        return self

    def __next__(self):
        ...

也就是说：

$$\operatorname{iter}(i) \mathrel{\text{is}} i$$

迭代器的 __iter__() 必须返回自身。collections.abc.Iterator 也正是以 __iter__() 和 __next__() 为核心接口。(docs.python.org)

一个最小的计数迭代器如下：

class CountDown:
    def __init__(self, start):
        self.current = start

    def __iter__(self):
        return self

    def __next__(self):
        if self.current < 0:
            raise StopIteration

        value = self.current
        self.current -= 1
        return value


counter = CountDown(2)

print(next(counter))  # 2
print(next(counter))  # 1
print(next(counter))  # 0

try:
    print(next(counter))
except StopIteration:
    print("exhausted")

输出：

2
1
0
exhausted

这里的 current 就是迭代器状态。每调用一次 next(counter)，状态就发生一次变化：

调用前 current	返回值	调用后 current
2	2	1
1	1	0
0	0	-1
-1	StopIteration	-1

注意：StopIteration 不是“异常情况”，而是同步迭代协议中表示“没有下一个元素”的控制信号。

---

二、iter() 和 next() 如何组成迭代协议

1. iter(obj) 的主要路径

调用：

iterator = iter(obj)

时，Python 首先尝试取得对象的迭代器。对普通自定义类而言，主要路径是调用 obj.__iter__()。

class Words:
    def __iter__(self):
        return iter(["Python", "iterator"])


words = Words()
iterator = iter(words)

print(iterator)
print(list(iterator))

iter() 的第二种形式接收一个可调用对象和哨兵值：

iter(callable, sentinel)

它会不断调用 callable()，直到返回值等于 sentinel。例如，按固定大小读取文件：

from functools import partial

with open("data.bin", "rb") as file:
    chunks = iter(partial(file.read, 4096), b"")

    for chunk in chunks:
        process(chunk)

这个组合的逻辑等价于：

while True:
    chunk = file.read(4096)
    if chunk == b"":
        break
    process(chunk)

iter(callable, sentinel) 适合把“反复调用直到终止值”的 API 转换成标准迭代器。其停止条件是值相等，不是对象身份，因此哨兵值应当选择不会与正常结果混淆的值。Python 3.14 内置函数文档列出了 iter(iterable) 和 iter(callable, sentinel) 两种调用形式。(docs.python.org)

2. next(iterator) 的语义

调用：

value = next(iterator)

相当于请求迭代器执行：

value = iterator.__next__()

如果还有数据，返回下一个元素；如果已经耗尽，抛出 StopIteration。也可以提供默认值：

iterator = iter([])

print(next(iterator, "empty"))
# empty

此时：

next(iterator, default)

在迭代器耗尽时返回 default，不会抛出 StopIteration。(docs.python.org)

手动消费迭代器的完整写法是：

iterator = iter([1, 2, 3])

while True:
    try:
        item = next(iterator)
    except StopIteration:
        break

    print(item)

输出：

1
2
3

for 循环只是把这套流程封装起来。

3. for 循环的展开模型

下面的代码：

for item in source:
    handle(item)
else:
    finish()

可以近似理解为：

iterator = iter(source)

while True:
    try:
        item = next(iterator)
    except StopIteration:
        finish()
        break

    handle(item)

如果循环体中执行了 break，则不会执行 else 分支：

for item in [1, 2, 3]:
    if item == 2:
        break
else:
    print("completed")

这里不会打印 "completed"，因为迭代没有通过正常耗尽结束。Python 语言参考明确规定，for 会先对目标表达式调用 iter()，随后不断取得迭代器提供的元素；迭代器耗尽后结束循环，并在没有 break 时执行 else。(docs.python.org)

可以用下面的类观察 for 的真实调用过程：

class Traced:
    def __iter__(self):
        print("__iter__()")
        return iter([10, 20])

source = Traced()

for value in source:
    print("value:", value)

输出：

__iter__()
value: 10
value: 20

__next__() 没有出现在输出中，因为这里返回的是列表迭代器。若自己实现迭代器，就能看到每一次 next() 调用。

---

三、可迭代对象与迭代器的关键区别

1. 可迭代对象通常可以反复创建迭代器

values = [1, 2, 3]

first = iter(values)
second = iter(values)

print(first is second)
# False

print(list(first))
# [1, 2, 3]

print(list(second))
# [1, 2, 3]

列表是可迭代对象。iter(values) 每次返回一个独立的列表迭代器，因此两个迭代过程互不影响。

2. 迭代器通常只能沿一个方向推进

values = [1, 2, 3]
iterator = iter(values)

print(list(iterator))
# [1, 2, 3]

print(list(iterator))
# []

第一次 list(iterator) 已经把迭代器推进到末尾，第二次只能得到空列表。

生成器也是如此：

def numbers():
    yield 1
    yield 2
    yield 3


generator = numbers()

print(list(generator))
# [1, 2, 3]

print(list(generator))
# []

因此，下面的类型注解表达了不同的使用承诺：

from collections.abc import Iterable, Iterator

def consume_once(source: Iterator[int]) -> int:
    return sum(source)

def consume_many_times(source: Iterable[int]) -> int:
    return sum(source) + sum(source)

第一个函数明确接受一次性数据流；第二个函数隐含要求 source 可以重复产生独立迭代器。如果把生成器传给第二个函数，第二次求和将得到 0：

print(consume_many_times([1, 2, 3]))
# 12

print(consume_many_times(iter([1, 2, 3])))
# 6

计算过程如下：

• 列表：第一次 sum 得到 6，第二次重新迭代仍得到 6，总和为 12；
• 列表迭代器：第一次得到 6，迭代器耗尽，第二次得到 0，总和为 6。

这不是类型注解造成的差异，而是对象的状态语义不同。

3. 自身就是迭代器的对象

很多对象同时是可迭代对象和迭代器：

iterator = iter([1, 2, 3])

print(iter(iterator) is iterator)
# True

而列表不是：

values = [1, 2, 3]

print(iter(values) is values)
# False

可用下图概括两者关系：

flowchart LR
    A[可迭代对象 Iterable] -->|iter| B[迭代器 Iterator]
    B -->|next| C[一个元素]
    B -->|next| D[继续推进内部状态]
    B -->|耗尽| E[StopIteration]
    B -->|iter| B

关键关系是：

• Iterable 的职责是提供迭代器；
• Iterator 的职责是提供下一个元素；
• 一个迭代器本身也必须是可迭代对象；
• 可迭代对象不一定是迭代器。

---

四、为什么 __getitem__() 也可能让对象可迭代

现代代码通常通过 __iter__() 实现可迭代对象，但 Python 还保留了一个历史兼容路径：如果对象没有 __iter__()，却支持从索引 0 开始的 __getitem__()，Python 可能通过递增索引的方式进行迭代，直到抛出 IndexError。

例如：

class LegacyRange:
    def __init__(self, stop):
        self.stop = stop

    def __getitem__(self, index):
        if index < 0 or index >= self.stop:
            raise IndexError(index)
        return index * 10


values = LegacyRange(3)

print(list(values))
# [0, 10, 20]

这个类没有定义 __iter__()，但仍然可以被 list() 消费。

这解释了一个容易误判的现象：

from collections.abc import Iterable

print(isinstance(LegacyRange(3), Iterable))
# False

print(iter(LegacyRange(3)))
# 可以成功取得迭代器

collections.abc.Iterable 的检查主要识别显式定义或注册的 __iter__()，不会识别只通过 __getitem__() 提供迭代能力的类。判断一个对象是否真正可迭代，可靠方法是直接调用 iter(obj)。(docs.python.org)

这也是为什么下面的代码比 isinstance(x, Iterable) 更可靠：

def is_iterable(value):
    try:
        iter(value)
    except TypeError:
        return False
    else:
        return True

但还要注意：iter(value) 成功，只能证明对象符合同步迭代入口；它不代表迭代一定不会在中途抛出其他业务异常。

---

五、惰性：值是在什么时候计算的

1. 惰性不是一种数据类型

**惰性（lazy evaluation）**在这里指的是：构造数据处理对象时不立即计算全部结果，而是在消费者请求下一个元素时，才推进上游并计算当前结果。

比较列表推导式和生成器表达式：

numbers = range(5)

list_result = [x * x for x in numbers]
generator_result = (x * x for x in numbers)

print(list_result)
# [0, 1, 4, 9, 16]

print(generator_result)
# <generator object ...>

列表推导式在创建时就完成全部计算并保存结果；生成器表达式创建的是生成器迭代器，表达式会在后续消费时逐项求值。生成器表达式在运行时产生生成器迭代器，而不是立即产生完整列表。(docs.python.org)

可以通过带副作用的函数观察差异：

def transform(value):
    print("transform:", value)
    return value * 10


eager = [transform(x) for x in range(3)]
print("after eager")

lazy = (transform(x) for x in range(3))
print("after lazy")

print(next(lazy))
print(next(lazy))

输出顺序为：

transform: 0
transform: 1
transform: 2
after eager
after lazy
transform: 0
0
transform: 1
10

生成器表达式创建后，transform() 尚未执行；第一次 next(lazy) 才处理第一个输入。

2. 惰性处理形成数据流

考虑这个管道：

source = range(1, 10)
filtered = (x for x in source if x % 2 == 0)
mapped = (x * 100 for x in filtered)

print(next(mapped))
# 200

为了产生 200，实际发生了：

1. 从 source 读取 1，过滤掉；
2. 读取 2，过滤通过；
3. 计算 2 * 100；
4. 向调用者返回 200。

没有必要先生成所有偶数，也没有必要先生成所有乘法结果。数据按需从右向左触发、从左向右流动：

flowchart LR
    S["range(1, 10)"] --> F[过滤偶数]
    F --> M[乘以 100]
    M --> C["next(mapped)"]
    C --> O[返回 200]

如果只请求一个元素，上游通常也只需要推进到能够产生这一个元素的位置。

3. 惰性并不意味着零内存

下面的链条通常只保存少量中间状态：

result = (
    value * value
    for value in range(10_000_000)
    if value % 2 == 0
)

但某些迭代器为了工作，必须缓存输入：

from itertools import product

result = product(range(10_000), range(10_000))

product() 在开始产生结果前，会完全消费输入迭代对象并保存各个输入池，因此输入必须是有限的；它的输出虽然是惰性的，但输入池并不是零内存。(docs.python.org)

因此应区分：

• 输出惰性：结果元组不是一次性全部生成；
• 输入物化：为了生成输出，输入可能先被完整保存；
• 中间缓存：某些组合器可能保留尚未被其他分支消费的数据。

惰性只回答“何时产生结果”，不自动回答“需要多少内存”。

---

六、耗尽：迭代器是一个状态机

1. 正常状态转换

一个有限迭代器至少可以抽象成两个状态：

ACTIVE  --__next__()--> ACTIVE
ACTIVE  --__next__()--> EXHAUSTED
EXHAUSTED --__next__()--> EXHAUSTED

更具体地说：

class OneShot:
    def __init__(self):
        self.done = False

    def __iter__(self):
        return self

    def __next__(self):
        if self.done:
            raise StopIteration

        self.done = True
        return "only once"

使用：

iterator = OneShot()

print(next(iterator))
# only once

try:
    next(iterator)
except StopIteration:
    print("finished")

try:
    next(iterator)
except StopIteration:
    print("still finished")

输出：

only once
finished
still finished

在协议层面，迭代器耗尽后应继续报告耗尽，而不应重新产生数据。

2. StopIteration.value 与普通 for 的差异

迭代器可以抛出带值的 StopIteration：

class WithReturn:
    def __iter__(self):
        return self

    def __next__(self):
        raise StopIteration("final result")

直接捕获时可以读取这个值：

iterator = WithReturn()

try:
    next(iterator)
except StopIteration as exc:
    print(exc.value)
    # final result

但普通 for 循环会把 StopIteration 当作结束信号，不会把 .value 传给循环体。生成器的 return value 则可以通过委托机制被外层生成器获取，这属于 yield from 的控制流语义，后文会说明。

3. 消费者会改变数据源状态

下面的函数名为 peek，但它并不能真正“查看而不消耗”：

def peek(iterator):
    return next(iterator, None)


source = iter([10, 20, 30])

print(peek(source))  # 10
print(list(source))  # [20, 30]

peek() 已经消费了第一个元素。对于迭代器而言，“读取”通常就意味着“推进状态”。

如果需要保留已经读出的元素，可以显式缓存：

source = iter([10, 20, 30])
first = next(source, None)
remaining = list(source)

print(first)
print(remaining)

或者使用 itertools.tee() 创建多个逻辑分支，但这并不是免费的复制，后文会讨论其缓存代价。

---

七、如何实现正确的自定义迭代器

1. 直接实现迭代器

下面实现一个闭区间整数迭代器：

class IntRange:
    def __init__(self, start, stop, step=1):
        if step == 0:
            raise ValueError("step must not be zero")

        self.current = start
        self.stop = stop
        self.step = step

    def __iter__(self):
        return self

    def __next__(self):
        if self.step > 0:
            if self.current >= self.stop:
                raise StopIteration
        else:
            if self.current <= self.stop:
                raise StopIteration

        value = self.current
        self.current += self.step
        return value

测试：

print(list(IntRange(0, 5)))
# [0, 1, 2, 3, 4]

print(list(IntRange(5, 0, -2)))
# [5, 3, 1]

这里采用半开区间 [start, stop)，与内置 range() 的习惯一致。实现时最容易出错的地方是：

• step == 0 时无限循环；
• 正步长与负步长的终止条件相反；
• 抛出 StopIteration 后不能重新返回元素；
• __iter__() 如果返回新对象，就不再是典型的自迭代器结构。

2. 可迭代对象与迭代器分离通常更灵活

直接让容器对象自身维护位置，会导致容器只能被安全地遍历一次。更好的结构是：

class IntRangeIterable:
    def __init__(self, start, stop, step=1):
        self.start = start
        self.stop = stop
        self.step = step

    def __iter__(self):
        current = self.start

        while (
            current < self.stop
            if self.step > 0
            else current > self.stop
        ):
            yield current
            current += self.step

使用：

values = IntRangeIterable(0, 3)

print(list(values))
# [0, 1, 2]

print(list(values))
# [0, 1, 2]

每次调用 iter(values) 都会取得一个新的生成器，因此两次遍历彼此独立。

可以把两种设计对比如下：

设计	__iter__() 返回	能否自然重复遍历
容器型可迭代对象	新迭代器	通常可以
迭代器对象	self	通常不可以
生成器函数返回的生成器	生成器自身	通常不可以

---

八、生成器：用 yield 实现迭代器

1. 生成器函数调用时不会立即执行函数体

含有 yield 的函数称为生成器函数。调用生成器函数时，得到的是生成器迭代器；函数体从第一次调用 next() 时才开始执行。执行到 yield 时暂停，并保留局部变量、指令位置和异常处理状态。(docs.python.org)

def events():
    print("start")
    yield "A"
    print("middle")
    yield "B"
    print("end")


generator = events()
print("created")

print(next(generator))
print(next(generator))

try:
    next(generator)
except StopIteration:
    print("done")

输出：

created
start
A
middle
B
end
done

时间顺序说明：

1. events() 调用只创建生成器对象；
2. 第一次 next() 执行到第一个 yield；
3. 第二次 next() 从上次暂停位置继续；
4. 第三次 next() 执行到函数末尾并抛出 StopIteration。

2. yield 是暂停点，不是普通返回

普通 return 结束整个函数；yield 只把一个值交给调用者，然后保留执行现场。

def counter():
    value = 0

    while value < 3:
        received = yield value
        print("received:", received)
        value += 1

第一次启动时：

generator = counter()
print(next(generator))
# 0

此时生成器暂停在：

received = yield value

因为 next(generator) 是继续执行的方式，当前 yield 表达式恢复后的值为 None：

print(next(generator))
# received: None
# 1

3. send() 将值送回暂停点

send(value) 会恢复生成器，并让暂停处的 yield 表达式求值为 value：

def accumulator():
    total = 0

    while True:
        value = yield total
        if value is None:
            return
        total += value


generator = accumulator()

print(next(generator))
# 0

print(generator.send(10))
# 10

print(generator.send(5))
# 15

执行过程：

操作	暂停处 yield 的结果	total	产生的下一个值
next() 启动	None	0	0
send(10)	10	10	10
send(5)	5	15	15

生成器尚未启动时，第一次 send() 必须传入 None：

generator = accumulator()
generator.send(10)

会失败，因为生成器还没有暂停在任何一个 yield 表达式上。Python 文档规定，启动生成器时只能使用 send(None)，通常直接调用 next() 更清晰。(docs.python.org)

4. throw() 把异常注入生成器

throw() 会在生成器暂停的 yield 位置抛出异常：

def resilient():
    while True:
        try:
            value = yield "waiting"
        except ValueError as exc:
            print("handled:", exc)
        else:
            print("received:", value)


generator = resilient()

print(next(generator))
# waiting

print(generator.throw(ValueError("bad input")))
# handled: bad input
# waiting

如果生成器内部没有捕获这个异常，异常会传播给调用者：

def simple():
    yield 1


generator = simple()
print(next(generator))

try:
    generator.throw(RuntimeError("failure"))
except RuntimeError as exc:
    print(type(exc).__name__, exc)

throw() 的重要意义在于：消费者可以把外部故障传递给数据生产逻辑，而不是只能通过共享变量通知生成器。

5. close() 请求生成器清理并结束

close() 会在暂停位置注入 GeneratorExit：

def resource_stream():
    print("open resource")
    try:
        yield 1
        yield 2
    finally:
        print("close resource")


generator = resource_stream()

print(next(generator))
# open resource
# 1

generator.close()
# close resource

生成器中的 finally 会执行，因此可以用于清理资源。若生成器在收到 GeneratorExit 后仍然产生值，Python 会抛出 RuntimeError。生成器方法 send()、throw() 和 close() 的行为由生成器协议定义；Python 3.13 起，如果生成器在关闭时通过 return 返回值，该值会由 close() 返回。(docs.python.org)

不过，不能把垃圾回收当作资源管理策略。尤其是文件、网络连接、数据库游标等外部资源，应通过 with 或显式关闭保证生命周期。Python 数据模型也明确建议显式释放外部资源，而不要依赖对象何时被垃圾回收。(docs.python.org)

---

九、yield from：委托迭代与返回值传递

1. 直接转发子迭代器的值

下面是一个简单的委托：

def child():
    yield "child-1"
    yield "child-2"


def parent():
    yield "parent-1"
    yield from child()
    yield "parent-2"


print(list(parent()))
# ['parent-1', 'child-1', 'child-2', 'parent-2']

yield from child() 不只是一个简化版循环，它还会把外部对父生成器的操作转发给子迭代器：

• send() 可以转发给子生成器；
• throw() 可以转发给子生成器；
• close() 会参与子生成器的关闭；
• 子生成器结束时的 StopIteration.value 会成为 yield from 表达式的结果。

2. 用 return 从子生成器返回结果

def child():
    yield 1
    yield 2
    return 99


def parent():
    result = yield from child()
    print("child result:", result)
    yield 3


print(list(parent()))

输出：

child result: 99
[1, 2, 3]

这里的执行过程是：

1. parent() 委托给 child()；
2. child() 产生 1 和 2；
3. child() 执行 return 99；
4. 子生成器结束，内部表现为 StopIteration(99)；
5. yield from child() 表达式的值为 99；
6. parent() 打印结果后继续产生 3。

语言参考规定，yield from <expr> 的表达式必须是可迭代对象；子迭代器结束时，StopIteration.value 成为 yield from 表达式的结果。(docs.python.org)

---

十、组合：把多个数据流连接起来

1. map()、过滤和生成器表达式构成流水线

source = range(1, 11)

pipeline = (
    value * value
    for value in source
    if value % 2 == 0
)

print(list(pipeline))
# [4, 16, 36, 64, 100]

这个流水线可以拆成：

$$x \in [1, 10]$$

先保留满足：

$$x \bmod 2 = 0$$

的值，再计算：

$$f(x) = x^2$$

因此输入 2, 4, 6, 8, 10 被转换为 4, 16, 36, 64, 100。

如果只取前两个结果：

from itertools import islice

source = range(1, 10_000_000)

pipeline = (
    value * value
    for value in source
    if value % 2 == 0
)

print(list(islice(pipeline, 2)))
# [4, 16]

islice() 会从输入迭代器中选取指定位置的元素，不支持负索引；当输入是迭代器时，跳过和读取都会推进原迭代器。(docs.python.org)

2. chain() 连接多个输入

from itertools import chain

result = chain([1, 2], (3, 4), "56")

print(list(result))
# [1, 2, 3, 4, '5', '6']

chain() 逐个耗尽输入对象：

第一个输入：[1, 2]
第二个输入：(3, 4)
第三个输入："56"

它不会把所有输入先合并成一个新列表，因此适合顺序连接多个数据源。但如果某个输入本身是一次性迭代器，chain() 消费它之后，该输入就已经耗尽。

3. zip() 是同步推进

left = iter([1, 2, 3])
right = iter(["a", "b"])

print(list(zip(left, right)))
# [(1, 'a'), (2, 'b')]

print(list(left))
# [3]

print(list(right))
# []

每次产生一个元组，zip() 同时从各个输入迭代器读取一个值。默认情况下，最短输入耗尽后停止，因此第三个元素 3 没有配对结果。

当长度不一致属于错误时，可以使用严格模式：

left = [1, 2, 3]
right = ["a", "b"]

try:
    print(list(zip(left, right, strict=True)))
except ValueError as exc:
    print(exc)

严格模式会把输入长度不一致报告为 ValueError，比静默截断更适合数据对齐场景。

4. zip() 的分组技巧依赖“同一个迭代器”

itertools 文档中的分组模式是：

from itertools import zip_longest

def grouper(iterable, size, fillvalue=None):
    iterators = [iter(iterable)] * size
    return zip_longest(*iterators, fillvalue=fillvalue)

测试：

print(list(grouper("ABCDEFG", 3)))
# [('A', 'B', 'C'), ('D', 'E', 'F'), ('G', None, None)]

关键不是创建了三个独立迭代器，而是：

[iter(iterable)] * size

复制的是同一个迭代器引用。每一轮 zip_longest() 会连续从同一个底层迭代器取 size 个元素，因此形成固定大小分组。

如果误写成：

[iter(iterable) for _ in range(size)]

对于列表等可重复迭代对象，将得到多个独立迭代器，结果会变成同一批元素被重复读取，而不是连续分组。

---

十一、itertools 的组合语义与内存边界

itertools 的核心价值不是“把循环写短”，而是提供一组遵循迭代协议的组合器。它们的行为必须从三个问题判断：

1. 是否惰性产生输出；
2. 是否完全消费输入；
3. 是否共享或缓存输入状态。

1. groupby() 是连续分组，不是全局聚合

from itertools import groupby

data = [
    ("error", 1),
    ("error", 2),
    ("info", 3),
    ("error", 4),
]

for key, group in groupby(data, key=lambda item: item[0]):
    print(key, list(group))

输出：

error [('error', 1), ('error', 2)]
info [('info', 3)]
error [('error', 4)]

groupby() 只把相邻且键相同的元素放到同一组。因此它与 SQL 的 GROUP BY 不同：

输入：error, error, info, error
输出：error组, info组, error组

如果希望所有相同键进入同一组，必须先按相同键排序：

data = sorted(data, key=lambda item: item[0])

for key, group in groupby(data, key=lambda item: item[0]):
    print(key, list(group))

groupby() 返回的每个分组本身也是迭代器，并且与外层 groupby() 共享底层输入。外层向前推进后，之前的分组通常就不能再读取，因此需要保留时必须立即物化为列表。(docs.python.org)

错误示例：

for key, group in groupby("AAABBC"):
    groups = list(group)
    # 这里如果继续推进外层，group 已经没有可读内容

正确的保留方式：

saved = []

for key, group in groupby("AAABBC"):
    saved.append((key, list(group)))

print(saved)
# [('A', ['A', 'A', 'A']), ('B', ['B', 'B']), ('C', ['C'])]

2. takewhile() 会消费第一个失败元素

from itertools import takewhile

source = iter([1, 2, 6, 3, 4])

prefix = takewhile(lambda x: x < 5, source)

print(list(prefix))
# [1, 2]

print(list(source))
# [3, 4]

值 6 不满足条件，因此没有出现在 prefix 中；但它已经被从底层输入读取出来并丢弃。takewhile() 的语义是“持续读取直到第一次失败”，而不是“找到一个边界但保留边界元素”。官方文档明确指出，第一个不满足条件的元素已经被消费，无法通过原输入迭代器再次访问。(docs.python.org)

这在解析协议流、日志流或分隔记录时尤其重要。若边界元素也有意义，就不能直接假设 takewhile() 会把它留在输入中。

3. tee() 提供分支，但需要缓存领先数据

from itertools import tee

source = iter([1, 2, 3])
first, second = tee(source)

print(next(first))
# 1

print(next(first))
# 2

print(list(second))
# [1, 2, 3]

为了让 second 仍然能够读取 1 和 2，tee() 必须缓存 first 已经消费、而 second 尚未消费的元素。

因此，tee() 的内存成本取决于分支之间的消费速度差异：

first 读取很快
second 读取很慢
=> 缓存不断增长

如果两个分支几乎同步消费，缓存通常较小；如果一个分支远远领先，缓存可能接近领先分支已经读取的数据量。官方文档将 tee() 描述为从一个可迭代对象返回多个独立迭代器，并给出了基于共享链表的近似实现。(docs.python.org)

tee() 不是“复制任意迭代器的完整快照”，它更接近“为多个读取游标维护共享历史”。

4. product() 输出惰性，但输入要先保存

from itertools import product

pairs = product([1, 2], ["a", "b"])

print(next(pairs))
# (1, 'a')

print(list(pairs))
# [(1, 'b'), (2, 'a'), (2, 'b')]

输出逐个产生，但输入池会在开始输出前被保存。对于有限小输入，这通常很方便；对于大型输入，输入物化本身可能成为内存瓶颈；对于无限输入，product() 无法正常开始，因为它需要先完成输入消费。官方文档明确说明，product() 会完全消费输入，并且只适用于有限输入。(docs.python.org)

5. 无限迭代器必须配合边界

from itertools import count, islice

infinite = count(10, 2)

print(list(islice(infinite, 5)))
# [10, 12, 14, 16, 18]

count() 可以产生无限序列；但下面的代码不会结束：

list(count())

即使生成器和 itertools 都是惰性的，list() 仍然要求把输入全部消费并保存，因此对无限输入会无限运行并持续占用内存。

安全组合通常需要至少一个终止条件：

from itertools import count, takewhile

values = takewhile(lambda x: x < 10, count(0, 2))
print(list(values))
# [0, 2, 4, 6, 8]

终止条件必须能够在有限步骤内成立；如果谓词永远为真，takewhile() 仍然不会结束。

---

十二、常见误解与失败表现

1. 误以为所有可迭代对象都能调用 len()

values = (x for x in range(3))

len(values)

结果：

TypeError: object of type 'generator' has no len()

可迭代性和可计数性是两个不同协议：

• 可迭代：能否通过 iter() 逐项读取；
• 有大小：是否实现 __len__()；
• 容器：通常同时支持成员测试、迭代和长度，但不是所有可迭代对象都是容器。

对流式数据，要求 len() 往往意味着必须先消费或物化数据，这与惰性模型冲突。

2. 误以为迭代器可以重新开始

iterator = iter([1, 2, 3])

print(next(iterator))  # 1
print(list(iterator))  # [2, 3]
print(list(iterator))  # []

要重新开始，应保留可迭代对象，而不是只保留迭代器：

values = [1, 2, 3]

for _ in range(2):
    print(list(values))

如果源数据来自网络、文件或数据库游标，则通常不能简单重播，需要显式缓存、重新打开源或设计可重放日志。

3. 误以为 list() 只是查看内容

iterator = iter([1, 2, 3])

snapshot = list(iterator)

print(snapshot)
# [1, 2, 3]

print(list(iterator))
# []

list(iterator) 是一个消费者，它会把迭代器推进到耗尽。调试时直接打印 list(iterator) 可能改变后续业务逻辑。

4. 误以为生成器函数调用就执行了代码

def load():
    print("loading")
    yield 1


generator = load()
print("created")

输出只有：

created

"loading" 要等到：

next(generator)

才会出现。若生产代码依赖函数调用立即完成校验、打开文件或建立连接，使用生成器函数时需要重新检查生命周期假设。

5. 在迭代期间修改容器

values = [1, 2, 3, 4]

for value in values:
    if value % 2 == 0:
        values.remove(value)

print(values)

结果可能是：

[1, 3]

但这种写法容易跳过元素，因为列表迭代器的位置与列表内容变化相互作用。更清晰的做法是创建新列表：

values = [1, 2, 3, 4]
values = [value for value in values if value % 2 != 0]

print(values)
# [1, 3]

这里不是迭代协议保证“修改一定报错”，而是具体容器和实现对修改期间行为的保证有限。工程代码不应依赖修改与遍历同时进行时的偶然结果。

---

十三、诊断迭代问题的方法

1. 先确认对象类型和身份

def inspect_iterable(value):
    print("type:", type(value))
    print("is iterator:", iter(value) is value)

例如：

inspect_iterable([1, 2, 3])
# type: <class 'list'>
# is iterator: False

inspect_iterable(iter([1, 2, 3]))
# type: <class 'list_iterator'>
# is iterator: True

iter(value) is value 是判断一个对象是否表现为自身迭代器的直接实验，但它不能替代对协议语义的完整理解。

2. 用带日志的迭代器定位提前消费

class LoggedIterator:
    def __init__(self, iterable):
        self._iterator = iter(iterable)

    def __iter__(self):
        return self

    def __next__(self):
        value = next(self._iterator)
        print("consume:", value)
        return value

测试：

source = LoggedIterator([1, 2, 3])
pipeline = (x * 10 for x in source if x > 1)

print(next(pipeline))

输出：

consume: 1
consume: 2
20

为了生成第一个结果 20，流水线消费了两个输入元素。这个方法可以发现：

• 某个过滤器是否提前读取了边界元素；
• 某个组合器是否比预期多消费；
• 多个消费者是否共享同一个底层迭代器；
• 是否在调试代码中意外调用了 list() 或 tuple()。

3. 区分“未执行”和“已耗尽”

生成器有两个容易混淆的状态：

未启动：函数体尚未执行
已耗尽：函数体已经结束，不能再产生值

示例：

def source():
    print("start")
    yield 1


generator = source()
print("created")

print(next(generator))
print(list(generator))
print(list(generator))

输出：

created
start
1
[]
[]

第一次 list(generator) 为空，可能有两种完全不同的原因：

1. 生成器本来就没有更多元素；
2. 生成器之前已经被其他代码消费过。

因此诊断时要追踪“谁先调用了 next()、list()、sum() 或 for”。

---

十四、同步迭代与异步迭代不是同一个协议

本文讨论的是同步迭代：

iter(obj)
next(iterator)
StopIteration

异步迭代使用另一组协议：

aiter(obj)
anext(iterator)
StopAsyncIteration

对应的语法是：

async for item in async_source:
    ...

异步生成器使用 async def 和 yield，产生异步迭代器；其 __anext__() 返回可等待对象，结束时通过 StopAsyncIteration 表示完成。Python 3.14 语言参考将同步生成器与异步生成器分别定义，二者不能仅通过把 for 改成 async for 来互换。(docs.python.org)

这一区分很重要：

for item in async_source:
    ...

不能替代：

async for item in async_source:
    ...

因为同步协议不会自动等待异步迭代器返回的可等待对象。

---

十五、如何选择数据表示：可迭代对象还是迭代器

可以根据调用方需要的语义选择接口。

1. 当数据可以重复读取时，返回可迭代对象

例如：

class UserCollection:
    def __init__(self, users):
        self._users = tuple(users)

    def __iter__(self):
        return iter(self._users)

每次遍历都从同一份稳定数据创建新迭代器。适合：

• 内存中的集合；
• 可重复查询的索引；
• 需要多次遍历的配置或元数据；
• 调用方不应影响对象本身状态的 API。

2. 当数据来自一次性来源时，明确返回迭代器

例如：

def read_lines(path):
    with open(path, encoding="utf-8") as file:
        for line in file:
            yield line.rstrip("\n")

这个函数返回生成器，调用方应理解为：

lines = read_lines("data.txt")

for line in lines:
    ...

而不是：

print(list(lines))
print(list(lines))  # 第二次为空

如果业务需要重复处理，应在边界处明确物化：

lines = list(read_lines("data.txt"))

代价是内存使用增加，但生命周期和重放语义变得清晰。

3. 类型注解应表达消费方式

from collections.abc import Iterable, Iterator

def summarize(values: Iterable[int]) -> int:
    return sum(values)

def stream(values: Iterable[int]) -> Iterator[int]:
    for value in values:
        yield value * 2

Iterable[int] 表示调用方只要求“能够取得迭代器”；Iterator[int] 表示返回的是一次性推进的数据流。若函数会多次遍历参数，文档和类型设计都应反映“可重复遍历”的要求，而不能只写成含糊的 list 或 object。

---

十六、核心结论

Python 迭代系统可以压缩为一条精确的数据流链：

可迭代对象
    --iter()-->
迭代器
    --next()-->
一个元素
    --重复推进-->
更多元素
    --StopIteration-->
耗尽

其中最重要的边界是：

1. 可迭代对象是数据来源，迭代器是带状态的读取游标。
2. 迭代器的 __iter__() 返回自身；可迭代对象的 __iter__() 通常返回新的迭代器。
3. for 循环本质上反复调用 next()，并把 StopIteration 转换为正常结束。
4. 惰性意味着按需计算，不意味着没有缓存、不消耗输入或没有资源成本。
5. 迭代器通常是一次性的；任何 next()、for、list()、sum() 都可能推进它。
6. groupby()、tee()、product() 等组合器有不同的共享、缓存和物化语义，不能只根据“返回迭代器”判断其内存行为。
7. 生成器在此协议上增加了暂停、恢复、发送值、注入异常、关闭和委托控制流。

掌握这些协议后，列表、文件、生成器、数据库游标、map()、zip() 和 itertools 就不再是彼此孤立的 API，而是同一种“按需推进的数据流”在不同场景下的具体实现。

---

系列导航与关联阅读

• 系列入口：Python 完整学习路线：从语言模型、并发到 Web、数据、AI 与生产交付
• 上一篇：Python 装饰器：函数包装、参数化、类装饰器与元数据
• 下一篇：Python 生成器：yield、send、throw、close 与委托
• 延伸：Python itertools：惰性组合、无限序列、分组和内存边界

官方资料

• Python 3.14 Documentation
• Python 3.14 Language Reference

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。