02-自定义迭代器
Python 版本要求：Python 3.11+
贯穿项目：日志分析系统 本节目标：实现 LogIterator 类，按时间范围过滤日志
概念铺垫
为什么需要自定义迭代器？
问题场景：按时间范围过滤日志
继续日志分析系统，现在需要：
- 只分析某段时间的日志（如 10:00-11:00 的错误）
- 大文件中筛选符合条件的数据
python
# ❌ 方案一：列表过滤（内存爆炸）
def filter_by_time_bad(filename: str, start: str, end: str) -> list[str]:
with open(filename) as f:
all_lines = f.readlines() # 加载全部！
return [line for line in all_lines if start <= line[:10] <= end]
# ✅ 方案二：自定义迭代器（内存友好）
class LogTimeIterator:
"""按时间范围过滤的日志迭代器"""
# ... 实现 __iter__ 和 __next__问题：如何让对象支持 for line in iterator？需要实现什么？
生活类比：自助餐厅取餐
把迭代器协议想象成自助餐厅：
┌─────────────────────────────────────────────────────────────┐
│ 自助餐厅类比 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 可迭代对象 = 餐厅菜单 │
│ • 列出所有菜品 │
│ • 你可以"遍历"它 │
│ │
│ 迭代器 = 取餐窗口 │
│ • 有取餐规则（__iter__：确认窗口） │
│ • 有取餐动作（__next__：取下一份） │
│ • 取完后窗口关闭（StopIteration） │
│ │
│ for循环 = 自动取餐 │
│ • 自动调用取餐动作 │
│ • 取完自动离开 │
│ │
│ 迭代器协议 = 餐厅规则 │
│ • 必须有 __iter__ 方法（证明是取餐窗口） │
│ • 必须有 __next__ 方法（能取下一份） │
│ • 没有更多时抛出 StopIteration（窗口关闭） │
│ │
└─────────────────────────────────────────────────────────────┘一句话：
迭代器协议 = 两个方法：
__iter__确认身份，__next__取下一份
L1 理解层：会用
迭代器协议详解
两个必需方法
┌─────────────────────────────────────────────────────────────┐
│ 迭代器协议 (Iterator Protocol) │
├─────────────────────────────────────────────────────────────┤
│ │
│ __iter__(self) │
│ ───────────────── │
│ • 返回迭代器本身（return self） │
│ • 让对象能在 for 循环中使用 │
│ • 被 iter() 函数调用 │
│ │
│ __next__(self) │
│ ───────────────── │
│ • 返回下一个值 │
│ • 被 next() 函数调用 │
│ • 没有更多元素时抛出 StopIteration │
│ • 必须维护内部状态（当前位置） │
│ │
│ 注意： │
│ ───────────── │
│ • 迭代器是一次性的，用完即弃 │
│ • StopIteration 是信号，不是错误 │
│ │
└─────────────────────────────────────────────────────────────┘最简示例：空迭代器
python
from collections.abc import Iterator
class EmptyIterator(Iterator[None]):
"""最简单的迭代器：立即停止"""
def __iter__(self) -> "EmptyIterator":
return self
def __next__(self) -> None:
raise StopIteration
empty = EmptyIterator()
print(list(empty)) # []迭代器模板
python
from collections.abc import Iterator
from typing import Generic, TypeVar
T = TypeVar("T")
class MyIterator(Iterator[T]):
"""自定义迭代器模板"""
def __init__(self, data: list[T]) -> None:
self._data: list[T] = data
self._index: int = 0
def __iter__(self) -> "MyIterator[T]":
return self
def __next__(self) -> T:
if self._index >= len(self._data):
raise StopIteration
value: T = self._data[self._index]
self._index += 1
return value
my_iter = MyIterator([1, 2, 3])
for item in my_iter:
print(item, end=" ")
# 1 2 3for 循环如何使用迭代器
执行流程时间线
┌─────────────────────────────────────────────────────────────┐
│ for 循环与自定义迭代器 │
│ │
│ 代码：for item in MyIterator([1, 2, 3]): │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 时间线 │ │
│ │ │ │
│ │ 1. 创建迭代器对象 │ │
│ │ MyIterator([1, 2, 3]) │ │
│ │ → self._index = 0 │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ 2. 调用 __iter__ │ │
│ │ iter(iterator) → 返回 self │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ 3. 调用 __next__ 第1次 │ │
│ │ self._index=0 < 3? → 返回 1 │ │
│ │ self._index → 1 │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ 4. 执行循环体 │ │
│ │ print(item) → 打印 1 │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ 5. 调用 __next__ 第2次 │ │
│ │ self._index=1 < 3? → 返回 2 │ │
│ │ self._index → 2 │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ 6. 执行循环体 │ │
│ │ print(item) → 打印 2 │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ 7. 调用 __next__ 第3次 │ │
│ │ self._index=2 < 3? → 返回 3 │ │
│ │ self._index → 3 │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ 8. 执行循环体 │ │
│ │ print(item) → 打印 3 │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ 9. 调用 __next__ 第4次 │ │
│ │ self._index=3 >= 3? → StopIteration │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ 10. for 循环捕获 StopIteration │ │
│ │ 循环结束 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ⚠️ 关键：每次 __next__ 都要更新状态 │
│ ⚠️ 关键：状态耗尽时必须抛出 StopIteration │
│ │
└─────────────────────────────────────────────────────────────┘手动迭代 vs for 循环
python
from collections.abc import Iterator
class Countdown(Iterator[int]):
"""倒数计时器"""
def __init__(self, start: int) -> None:
self.current: int = start
def __iter__(self) -> "Countdown":
return self
def __next__(self) -> int:
if self.current <= 0:
raise StopIteration
result: int = self.current
self.current -= 1
return result
# 方式一：for 循环（自动处理 StopIteration）
for num in Countdown(5):
print(num, end=" ")
# 5 4 3 2 1
# 方式二：手动迭代（需要处理异常）
counter = Countdown(3)
print(next(counter)) # 3
print(next(counter)) # 2
print(next(counter)) # 1
# print(next(counter)) # StopIteration贯穿实战：LogIterator 按时间过滤
场景说明
日志分析系统中，需要按时间范围过滤日志：
日志格式示例：
[2024-01-19 10:00:15] INFO server started
[2024-01-19 10:05:30] ERROR connection failed
[2024-01-19 10:10:00] WARN timeout
[2024-01-19 11:00:00] INFO new request需求：只读取 10:00-10:30 时间段的日志。
实现 LogIterator
python
from collections.abc import Iterator
from datetime import datetime
from typing import Self
class LogTimeIterator(Iterator[str]):
"""按时间范围过滤的日志迭代器
贯穿项目：日志分析系统
功能：只产出指定时间范围内的日志行
使用：
for line in LogTimeIterator("app.log", "10:00", "10:30"):
print(line)
"""
def __init__(
self,
filename: str,
start_time: str,
end_time: str
) -> None:
self.filename: str = filename
self.start_time: str = start_time # 如 "10:00"
self.end_time: str = end_time # 如 "10:30"
self._file = None
self._exhausted: bool = False
def __iter__(self) -> Self:
"""打开文件，返回自身"""
if self._file is None:
self._file = open(self.filename, "r", encoding="utf-8")
return self
def __next__(self) -> str:
"""返回下一行符合条件的日志"""
if self._exhausted:
raise StopIteration
if self._file is None:
self.__iter__() # 确保文件已打开
while True:
line: str = self._file.readline()
if not line: # 文件结束
self._file.close()
self._exhausted = True
raise StopIteration
# 解析时间（日志格式：[2024-01-19 HH:MM:SS]）
time_str: str = self._extract_time(line)
if time_str and self._in_range(time_str):
return line.strip()
def _extract_time(self, line: str) -> str | None:
"""提取日志时间（HH:MM 格式）"""
if not line.startswith("["):
return None
# [2024-01-19 10:00:15] → 取 "10:00"
try:
return line[12:17] # HH:MM 位置
except IndexError:
return None
def _in_range(self, time_str: str) -> bool:
"""检查时间是否在范围内"""
return self.start_time <= time_str <= self.end_time
def __del__(self) -> None:
"""清理资源"""
if self._file and not self._file.closed:
self._file.close()使用示例
python
# 创建迭代器，只读取 10:00-10:30 的日志
log_iter = LogTimeIterator("app.log", "10:00", "10:30")
# for 循环自动处理
for line in log_iter:
if "ERROR" in line:
print(line)
# 也可以手动控制
log_iter2 = LogTimeIterator("app.log", "10:00", "10:30")
first_match = next(log_iter2) # 第一条符合条件的关键代码说明：
| 代码 | 含义 | 为什么这样写 |
|---|---|---|
def __iter__(self) -> Self | 返回自身 | 迭代器协议要求，让对象可在 for 中使用 |
self._file.readline() | 逐行读取 | 惰性处理，不会一次性加载全部 |
while True: 循环 | 跳过不符合条件的行 | 直到找到符合条件的才返回 |
self._exhausted = True | 标记耗尽状态 | 防止再次调用时重复关闭文件 |
def __del__(self) | 析构方法 | 保证文件句柄被释放 |
更多示例
斐波那契数列迭代器
python
from collections.abc import Iterator
from typing import Self
class Fibonacci(Iterator[int]):
"""斐波那契数列迭代器"""
def __init__(self, max_count: int | None = None) -> None:
self.max_count: int | None = max_count
self.count: int = 0
self.a: int = 0
self.b: int = 1
def __iter__(self) -> Self:
return self
def __next__(self) -> int:
if self.max_count is not None and self.count >= self.max_count:
raise StopIteration
result: int = self.a
self.a, self.b = self.b, self.a + self.b
self.count += 1
return result
# 前10个斐波那契数
fib = Fibonacci(max_count=10)
print(list(fib))
# [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]分页数据迭代器
python
from collections.abc import Iterator
from typing import Any, Callable, Self
class PageIterator(Iterator[dict[str, Any]]):
"""分页数据迭代器
模拟 API 分页获取，每次只加载一页数据
"""
def __init__(
self,
fetch_func: Callable[[int, int], list[dict]],
page_size: int = 10
) -> None:
self.fetch_func: Callable = fetch_func
self.page_size: int = page_size
self.current_page: int = 0
self.current_items: list[dict] = []
self.current_index: int = 0
def __iter__(self) -> Self:
return self
def __next__(self) -> dict[str, Any]:
if self.current_index >= len(self.current_items):
self.current_page += 1
self.current_items = self.fetch_func(
self.current_page, self.page_size
)
self.current_index = 0
if not self.current_items:
raise StopIteration
item: dict[str, Any] = self.current_items[self.current_index]
self.current_index += 1
return item
# 模拟 API
def fetch_page(page: int, size: int) -> list[dict[str, Any]]:
if page > 3:
return []
return [
{"id": (page - 1) * size + i, "page": page}
for i in range(size)
]
# 使用：自动翻页
for item in PageIterator(fetch_page, page_size=5):
print(f"ID: {item['id']}, Page: {item['page']}")
# ID: 0, Page: 1 ... ID: 14, Page: 3L2 实践层：用好
推荐做法与反模式
推荐做法
| 做法 | 原因 | 示例 |
|---|---|---|
| 优先用生成器 | 代码更简洁，Python 自动处理协议 | def gen(): yield x |
| iter 返回 self | 迭代器本身就是迭代器 | def __iter__(self): return self |
| 用类型注解 | 提高可读性和 IDE 支持 | def __next__(self) -> int: |
| 继承 Iterator[T] | 简化协议实现 | class MyIter(Iterator[int]): |
| 维护耗尽状态 | 防止重复调用问题 | self._exhausted = True |
反模式：不要这样做
python
# ❌ 错误：__iter__ 返回新对象（状态丢失）
class WrongIterator:
def __init__(self, data):
self.data = data
self.index = 0
def __iter__(self):
return MyIterator(self.data) # 新对象！
def __next__(self):
# ...
# ✅ 正确：__iter__ 返回 self
class CorrectIterator:
def __iter__(self):
return selfpython
# ❌ 错误：忘记抛出 StopIteration（无限循环）
class InfiniteIterator:
def __next__(self):
return self.data[self.index] # 无终止条件
# ✅ 正确：耗尽时抛出异常
class FiniteIterator:
def __next__(self):
if self.index >= len(self.data):
raise StopIteration
# ...适用场景
| 场景 | 推荐 | 原因 |
|---|---|---|
| 简单序列遍历 | ❌ | 用生成器更简洁 |
| 自定义数据结构遍历 | ✅ | 需要复杂状态管理 |
| 分页数据加载 | ✅ | 需要维护页码状态 |
| 文件流处理 | ✅ | 需要控制资源生命周期 |
| 无限序列 | ✅ | 需要维护计算状态 |
迭代器 vs 可迭代对象
两种设计模式
┌─────────────────────────────────────────────────────────────┐
│ 两种模式：迭代器 vs 可迭代对象 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 模式一：迭代器（Iterator） │
│ ─────────────────────────── │
│ • __iter__ 返回 self │
│ • 一次性使用，耗尽后需重新创建 │
│ • 适合：数据流、文件、无限序列 │
│ │
│ 模式二：可迭代对象（Iterable） │
│ ───────────────────────────── │
│ • __iter__ 返回新的迭代器对象 │
│ • 可多次遍历，每次创建新迭代器 │
│ • 适合：容器类（list、dict） │
│ │
│ 选择依据： │
│ ───────────── │
│ • 需要多次遍历 → 可迭代对象模式 │
│ • 只需一次遍历或数据流 → 迭代器模式 │
│ │
└─────────────────────────────────────────────────────────────┘可迭代对象示例
python
from collections.abc import Iterator, Iterable
from typing import Self
class MyContainer(Iterable[int]):
"""可迭代对象：可多次遍历"""
def __init__(self, data: list[int]) -> None:
self._data: list[int] = data
def __iter__(self) -> Iterator[int]:
"""每次返回新的迭代器"""
return MyContainerIterator(self._data)
class MyContainerIterator(Iterator[int]):
"""独立的迭代器类"""
def __init__(self, data: list[int]) -> None:
self._data: list[int] = data
self._index: int = 0
def __iter__(self) -> Self:
return self
def __next__(self) -> int:
if self._index >= len(self._data):
raise StopIteration
result: int = self._data[self._index]
self._index += 1
return result
# 使用：可多次遍历
container = MyContainer([1, 2, 3])
for x in container:
print(x) # 1 2 3
for x in container: # 新迭代器，再次遍历成功
print(x) # 1 2 3L3 专家层：深入
Python 如何实现迭代器协议
迭代器协议底层实现：
for item in obj:
│
▼
① iter(obj) ──────► obj.__iter__()
│ │
│ ▼
│ 返回迭代器（通常 return self）
▼
② next(iterator) ─► iterator.__next__()
│ │
│ ▼
│ 返回元素 或 raise StopIteration
▼
③ 赋值给 item，执行循环体
│
▼
④ 重复 ②-③，直到 StopIteration → for 终止
关键细节：
• CPython 中 list_iterator 是 C 结构体，包含 it_seq（原列表）和 it_index（当前位置）
• StopIteration 在 for 循环中被内部捕获，用户代码看不到
• 迭代器只存储对原数据的引用和索引，不复制数据验证：
python
import dis
def demo():
for x in [1, 2, 3]:
print(x)
dis.dis(demo)
# 关键字节码：
# GET_ITER → 调用 __iter__ 获取迭代器
# FOR_ITER → 调用 __next__，处理 StopIteration
# STORE_FAST → 存储当前元素
# JUMP_ABSOLUTE → 跳回 FOR_ITER
# 查看 list_iterator 内部结构
it = iter([1, 2, 3])
print(type(it)) # <class 'list_iterator'>
print(it.__length_hint__()) # 3（C 实现特有，预估剩余元素数）StopIteration 的内部处理机制
python
# CPython 处理流程：
# ceval.c 中的 FOR_ITER 指令：
# 1. 调用 tp_iternext (即 __next__)
# 2. 如果返回 NULL 且没有异常 → 已耗尽，跳转到循环末尾
# 3. 如果返回 NULL 且有 StopIteration → 消耗异常，跳转
# 4. 否则 → 正常值，STORE_FAST 存入变量
# for 循环等价于：
it = iter(obj)
while True:
try:
x = next(it)
except StopIteration:
break
# 循环体生成器：迭代器的快捷方式
自定义迭代器 vs 生成器：
自定义迭代器（手写类）： 生成器（yield）：
┌─────────────────────┐ ┌─────────────────────┐
│ class MyIter: │ │ def my_gen(): │
│ def __iter__(self):│ │ for item in data:│
│ return self │ │ if cond: │
│ def __next__(self):│ │ yield x │
│ ...状态管理... │ │ │
│ │ │ Python 自动生成： │
│ 需要手动实现： │ │ • __iter__ │
│ • 状态变量 │ │ • __next__ │
│ • StopIteration │ │ • 状态保存/恢复 │
│ • 资源清理 │ │ • StopIteration │
└─────────────────────┘ └─────────────────────┘
生成器自动满足迭代器协议，推荐优先使用 yield。性能考量
| 操作 | 时间复杂度 | 空间复杂度 | 说明 |
|---|---|---|---|
__iter__ 调用 | O(1) | O(1) | 仅返回 self |
__next__ 调用 | O(1) | O(1) | 索引递增，常数时间 |
| 创建迭代器 | O(1) | O(1) | ~56 字节（list_iterator） |
| 遍历 N 个元素 | O(N) | O(1) | 每元素 O(1)，总空间常数 |
| list(iterator) | O(N) | O(N) | 申请 N 个元素的新列表 |
知识关联
迭代器协议知识关联：
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ __iter__ │────→│ __next__ │────→│ StopIteration│
│ 获取迭代器 │ │ 获取元素 │ │ 终止信号 │
└──────────────┘ └──────────────┘ └──────────────┘
│ │ │
↓ ↓ ↓
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ for 循环 │ │ next() 函数 │ │ 字节码指令 │
│ 自动处理 │ │ 手动控制 │ │ GET_ITER │
│ │ │ │ │ FOR_ITER │
└──────────────┘ └──────────────┘ └──────────────┘
│
↓
┌──────────────┐ ┌──────────────┐
│ 生成器 │ │ itertools │
│ yield 语法 │────→│ 链式处理 │
│ 自动协议 │ │ 组合迭代 │
└──────────────┘ └──────────────┘回答以下问题，检查你是否掌握了核心概念：
- 迭代器协议需要实现哪两个方法？
__iter__方法应该返回什么？- 迭代器耗尽时应该怎么做？
- 迭代器和可迭代对象有什么区别？
- 为什么迭代器是一次性的？
答案：
__iter__和__next__- 返回
self（迭代器本身） - 抛出
StopIteration异常 - 迭代器：
__iter__返回 self，一次性；可迭代对象：__iter__返回新迭代器，可多次遍历 - 迭代器维护状态指针，用完即弃，无法重置
本章能力清单
学完本章，你能够：
- [x] 实现迭代器协议（
__iter__+__next__） - [x] 编写 LogTimeIterator 按条件过滤日志
- [x] 区分迭代器模式与可迭代对象模式
- [x] 理解
StopIteration异常的作用 - [x] 控制迭代器的资源生命周期（文件句柄等）
前置知识检查：
- 你是否理解了迭代基础（for 循环背后的机制）？ ← 第1节
下一步学习：
- 生成器基础（用 yield 简化迭代器） → 第3节
- 生成器高级特性 → 第4节
导航：