字数 2425，阅读大约需 13 分钟
Python迭代器：什么是迭代器？
学习目标
• 理解迭代器的概念和作用
• 掌握迭代器协议的实现方法
• 学会创建和使用自定义迭代器
• 了解迭代器与可迭代对象的区别
1. 迭代器基础概念
什么是迭代器？
迭代器（Iterator）是一个包含可数个值的对象，可以被遍历访问其中的所有值。在Python中，迭代器是实现了迭代器协议的对象，该协议包含两个特殊方法：
•
__iter__()：返回迭代器对象本身•
__next__()：返回序列中的下一个值
迭代器的特点
1. 惰性计算：只在需要时计算下一个值
2. 内存高效：不需要一次性加载所有数据到内存
3. 单向遍历：只能向前遍历，不能回退
4. 一次性使用：遍历完成后就被"耗尽"了
2. 基础示例
示例1：理解内置迭代器
# 列表是可迭代对象，但不是迭代器
numbers = [1, 2, 3, 4, 5]
print(f"列表本身是迭代器吗？{hasattr(numbers, '__next__')}") # False
# 通过iter()函数获取迭代器
iterator = iter(numbers)
print(f"iter()返回的对象是迭代器吗？{hasattr(iterator, '__next__')}") # True
# 使用next()函数逐一获取元素
print(next(iterator)) # 1
print(next(iterator)) # 2
print(next(iterator)) # 3
# 使用for循环遍历剩余元素
for num in iterator:
print(num) # 4, 5示例2：迭代器耗尽演示
# 创建一个简单的迭代器
data = iter([10, 20, 30])
# 第一次遍历
print("第一次遍历：")
for item in data:
print(item)
# 第二次遍历（迭代器已耗尽）
print("第二次遍历：")
for item in data:
print(item) # 不会打印任何内容3. 创建自定义迭代器
示例3：计数器迭代器
class Counter:
"""一个简单的计数器迭代器"""
def __init__(self, start, end):
self.current = start
self.end = end
def __iter__(self):
"""返回迭代器对象本身"""
return self
def __next__(self):
"""返回序列中的下一个值"""
if self.current < self.end:
current = self.current
self.current += 1
return current
else:
raise StopIteration # 迭代结束时抛出此异常
# 使用自定义迭代器
counter = Counter(1, 5)
print("使用for循环：")
for num in counter:
print(num) # 1, 2, 3, 4
# 手动调用next()
counter2 = Counter(10, 13)
print("\n手动调用next()：")
print(next(counter2)) # 10
print(next(counter2)) # 11
print(next(counter2)) # 12
# print(next(counter2)) # 会抛出StopIteration异常示例4：斐波那契数列迭代器
class Fibonacci:
"""斐波那契数列迭代器"""
def __init__(self, max_count):
self.max_count = max_count
self.count = 0
self.a, self.b = 0, 1
def __iter__(self):
return self
def __next__(self):
if self.count < self.max_count:
if self.count == 0:
self.count += 1
return self.a
elif self.count == 1:
self.count += 1
return self.b
else:
self.a, self.b = self.b, self.a + self.b
self.count += 1
return self.b
else:
raise StopIteration
# 使用斐波那契迭代器
fib = Fibonacci(10)
print("前10个斐波那契数：")
for num in fib:
print(num, end=" ") # 0 1 1 2 3 5 8 13 21 344. 实际应用场景
示例5：文件行读取器
class FileLineReader:
"""文件行读取迭代器"""
def __init__(self, filename):
self.filename = filename
self.file = None
def __iter__(self):
self.file = open(self.filename, 'r', encoding='utf-8')
return self
def __next__(self):
if self.file:
line = self.file.readline()
if line:
return line.strip() # 移除换行符
else:
self.file.close()
raise StopIteration
else:
raise StopIteration
# 使用示例（假设有一个test.txt文件）
"""
# 创建测试文件
with open('test.txt', 'w', encoding='utf-8') as f:
f.write("第一行\n第二行\n第三行\n")
# 使用文件迭代器
file_reader = FileLineReader('test.txt')
for line in file_reader:
print(f"读取到：{line}")
"""示例6：数据分批处理迭代器
class BatchIterator:
"""数据分批处理迭代器"""
def __init__(self, data, batch_size):
self.data = data
self.batch_size = batch_size
self.index = 0
def __iter__(self):
return self
def __next__(self):
if self.index >= len(self.data):
raise StopIteration
batch = self.data[self.index:self.index + self.batch_size]
self.index += self.batch_size
return batch
# 使用分批迭代器
data = list(range(1, 21)) # 1到20的数字
batch_iter = BatchIterator(data, 5)
print("分批处理数据：")
for batch in batch_iter:
print(f"当前批次：{batch}")5. 迭代器 vs 可迭代对象
概念区别
• 可迭代对象（Iterable）：实现了
__iter__()方法的对象• 迭代器（Iterator）：实现了
__iter__()和__next__()方法的对象
示例7：区别演示
# 可迭代对象但不是迭代器
class NumberCollection:
def __init__(self, numbers):
self.numbers = numbers
def __iter__(self):
return iter(self.numbers) # 返回一个新的迭代器
# 既是可迭代对象又是迭代器
class NumberIterator:
def __init__(self, numbers):
self.numbers = numbers
self.index = 0
def __iter__(self):
return self
def __next__(self):
if self.index < len(self.numbers):
value = self.numbers[self.index]
self.index += 1
return value
raise StopIteration
# 测试区别
collection = NumberCollection([1, 2, 3])
iterator = NumberIterator([1, 2, 3])
print("可迭代对象可以多次遍历：")
for num in collection:
print(num, end=" ")
print()
for num in collection:
print(num, end=" ") # 可以再次遍历
print("\n\n迭代器只能遍历一次：")
for num in iterator:
print(num, end=" ")
print()
for num in iterator:
print(num, end=" ") # 不会输出任何内容（已耗尽）6. 实际编程示例：智能数据流处理器
以下是一个完整的实际编程示例，展示如何使用迭代器来处理实时数据流：
import time
import random
from datetime import datetime
class SensorDataStream:
"""模拟传感器数据流的迭代器"""
def __init__(self, sensor_id, duration_seconds=10):
self.sensor_id = sensor_id
self.duration_seconds = duration_seconds
self.start_time = time.time()
self.data_count = 0
def __iter__(self):
return self
def __next__(self):
current_time = time.time()
if current_time - self.start_time >= self.duration_seconds:
print(f"传感器 {self.sensor_id} 数据流结束，共产生 {self.data_count} 条数据")
raise StopIteration
# 模拟传感器读取延迟
time.sleep(0.5)
# 生成模拟数据
self.data_count += 1
temperature = round(random.uniform(20.0, 35.0), 2)
humidity = round(random.uniform(40.0, 80.0), 2)
return {
'sensor_id': self.sensor_id,
'timestamp': datetime.now().strftime('%H:%M:%S'),
'temperature': temperature,
'humidity': humidity,
'data_point': self.data_count
}
class DataProcessor:
"""数据处理器，使用迭代器处理数据流"""
def __init__(self, temp_threshold=30.0):
self.temp_threshold = temp_threshold
self.processed_count = 0
self.alert_count = 0
def process_data_stream(self, data_stream):
"""处理数据流的主要方法"""
print(f"开始处理传感器数据流...")
print(f"温度预警阈值设定为: {self.temp_threshold}°C")
print("-" * 60)
try:
for data in data_stream:
self.processed_count += 1
# 数据处理逻辑
status = self._analyze_data(data)
# 显示处理结果
self._display_data(data, status)
# 检查是否需要预警
if status == "HIGH_TEMP":
self.alert_count += 1
self._send_alert(data)
except KeyboardInterrupt:
print("\n\n手动停止数据流处理...")
# 显示处理统计
self._show_statistics()
def _analyze_data(self, data):
"""分析单条数据"""
temp = data['temperature']
humidity = data['humidity']
if temp > self.temp_threshold:
return "HIGH_TEMP"
elif humidity > 70:
return "HIGH_HUMIDITY"
else:
return "NORMAL"
def _display_data(self, data, status):
"""显示数据处理结果"""
status_icon = {
"NORMAL": "✓",
"HIGH_TEMP": "🌡️ ",
"HIGH_HUMIDITY": "💧"
}
print(f"[{data['timestamp']}] 传感器{data['sensor_id']} | "
f"温度: {data['temperature']}°C | "
f"湿度: {data['humidity']}% | "
f"状态: {status_icon.get(status, '?')} {status}")
def _send_alert(self, data):
"""发送高温预警"""
print(f" ⚠️ 高温预警！当前温度 {data['temperature']}°C 超过阈值 {self.temp_threshold}°C")
def _show_statistics(self):
"""显示处理统计信息"""
print("-" * 60)
print("数据处理完成！")
print(f"总处理数据条数: {self.processed_count}")
print(f"高温预警次数: {self.alert_count}")
if self.processed_count > 0:
alert_rate = (self.alert_count / self.processed_count) * 100
print(f"预警率: {alert_rate:.1f}%")
class MultiSensorManager:
"""多传感器管理器，演示多个迭代器的使用"""
def __init__(self):
self.sensors = []
self.processor = DataProcessor(temp_threshold=28.0)
def add_sensor(self, sensor_id, duration=8):
"""添加传感器"""
sensor_stream = SensorDataStream(sensor_id, duration)
self.sensors.append(sensor_stream)
print(f"已添加传感器 {sensor_id}，运行时长 {duration} 秒")
def process_all_sensors(self):
"""依次处理所有传感器数据"""
print(f"开始处理 {len(self.sensors)} 个传感器的数据流...")
print("=" * 60)
for i, sensor_stream in enumerate(self.sensors, 1):
print(f"\n>>> 正在处理第 {i} 个传感器数据流 <<<")
self.processor.process_data_stream(sensor_stream)
if i < len(self.sensors):
print("\n等待处理下一个传感器...")
time.sleep(1)
# 完整的使用演示
def main():
"""主函数，演示完整的数据流处理过程"""
print("智能传感器数据流处理系统")
print("=" * 60)
# 方式1：处理单个传感器数据流
print("\n【方式1：单传感器数据流处理】")
single_sensor = SensorDataStream("TEMP001", duration_seconds=6)
processor = DataProcessor(temp_threshold=25.0)
processor.process_data_stream(single_sensor)
print("\n" + "="*60)
# 方式2：处理多个传感器数据流
print("\n【方式2：多传感器数据流处理】")
manager = MultiSensorManager()
manager.add_sensor("TEMP002", duration=5)
manager.add_sensor("TEMP003", duration=4)
manager.process_all_sensors()
print("\n" + "="*60)
print("系统演示完成！")
if __name__ == "__main__":
# 运行演示
# 注意：这个示例会实际运行并产生输出
# 如果不想等待，可以将duration_seconds设置为更小的值
print("准备运行传感器数据流处理演示...")
print("提示：程序运行过程中可以按 Ctrl+C 提前终止")
input("按回车键开始演示...")
main()代码运行结果示例：
智能传感器数据流处理系统
============================================================
【方式1：单传感器数据流处理】
开始处理传感器数据流...
温度预警阈值设定为: 25.0°C
------------------------------------------------------------
[14:32:15] 传感器TEMP001 | 温度: 23.4°C | 湿度: 65.2% | 状态: ✓ NORMAL
[14:32:16] 传感器TEMP001 | 温度: 28.7°C | 湿度: 58.9% | 状态: 🌡️ HIGH_TEMP
⚠️ 高温预警！当前温度 28.7°C 超过阈值 25.0°C
[14:32:16] 传感器TEMP001 | 温度: 22.1°C | 湿度: 75.3% | 状态: 💧 HIGH_HUMIDITY
传感器 TEMP001 数据流结束，共产生 12 条数据
------------------------------------------------------------
数据处理完成！
总处理数据条数: 12
高温预警次数: 3
预警率: 25.0%这个示例展现了迭代器的以下优势：
1. 内存高效：不需要预先加载所有数据，实时处理
2. 代码清晰：使用for循环简洁地处理数据流
3. 易于扩展：可以轻松添加新的传感器和处理逻辑
4. 实时处理：模拟真实的数据流处理场景
5. 资源管理：迭代器自动处理数据流的开始和结束
7. 总结
关键要点
1. 迭代器实现了
__iter__()和__next__()方法2. 迭代器提供了内存高效的数据遍历方式
3. 迭代器是一次性的，遍历完就被耗尽
4. 可以创建自定义迭代器来处理特定的数据访问模式
最佳实践
• 在处理大量数据时优先考虑使用迭代器
• 合理处理
StopIteration异常• 区分可迭代对象和迭代器的使用场景
• 在自定义迭代器中正确实现协议方法
进阶学习
• 生成器（Generator）：更简洁的迭代器创建方式
•
itertools模块：Python标准库中的迭代器工具• 异步迭代器：支持异步编程的迭代器
评论区