本文目录导读：
我来详细介绍如何自定义迭代对象的状态,包括基本方法和高级技巧。
基础迭代器实现
使用类实现迭代器
class RangeIterator:
"""自定义范围迭代器"""
def __init__(self, start, end, step=1):
self.current = start
self.end = end
self.step = step
self._state = 'ready' # 自定义状态
def __iter__(self):
self._state = 'iterating'
return self
def __next__(self):
if self.current >= self.end:
self._state = 'exhausted'
raise StopIteration
value = self.current
self.current += self.step
self._state = f'at_{value}'
return value
def get_state(self):
"""获取当前迭代器状态"""
return {
'state': self._state,
'current': self.current,
'end': self.end,
'progress': (self.current - self.end) / self.end * 100 if self.end else 0
}
# 使用示例
ri = RangeIterator(0, 10, 2)
for i in ri:
print(f"值: {i}, 状态: {ri.get_state()}")
使用生成器函数
def stateful_generator(data):
"""带状态的生成器"""
state = {'processed': 0, 'failed': 0, 'total': len(data)}
for index, item in enumerate(data):
try:
# 处理数据
result = item * 2
state['processed'] += 1
state['current_index'] = index
yield result, state.copy()
except Exception as e:
state['failed'] += 1
state['last_error'] = str(e)
yield None, state.copy()
# 使用示例
data = [1, 2, 3, 'error', 5]
gen = stateful_generator(data)
for value, state in gen:
print(f"值: {value}, 状态: {state}")
可重置迭代器
class ResettableIterator:
"""可重置状态的迭代器"""
def __init__(self, data):
self._original_data = data
self.reset()
def reset(self, new_data=None):
"""重置迭代器状态"""
if new_data:
self._original_data = new_data
self.data = list(self._original_data)
self.index = 0
self.iterations = 0
self.status = 'reset'
def __iter__(self):
self.iterations += 1
self.status = f'iteration_{self.iterations}'
return self
def __next__(self):
if self.index >= len(self.data):
self.status = 'completed'
raise StopIteration
value = self.data[self.index]
self.index += 1
self.status = f'processing_{self.index}/{len(self.data)}'
return value
def peek(self, offset=0):
"""预览下一个元素而不改变状态"""
peek_index = self.index + offset
if peek_index < len(self.data):
return self.data[peek_index]
return None
# 使用示例
ri = ResettableIterator([1, 2, 3, 4, 5])
for item in ri:
if item == 3:
print(f"批处理状态: {ri.status}, 当前位置: {ri.index}")
# 重置迭代器
ri.reset()
并行迭代器（带状态同步）
class ParallelIterator:
"""并行迭代器，支持多个序列同步迭代"""
def __init__(self, *iterables):
self.iterables = iterables
self.index = 0
self.states = []
# 初始化每个迭代器的状态
for i, iterable in enumerate(iterables):
if hasattr(iterable, '__iter__'):
self.states.append({
'index': 0,
'iter': iter(iterable),
'exhausted': False
})
def __iter__(self):
return self
def __next__(self):
result = []
self.states = []
for i, state in enumerate(self.states):
if state['exhausted']:
result.append(None)
else:
try:
value = next(state['iter'])
state['index'] += 1
result.append(value)
except StopIteration:
state['exhausted'] = True
result.append(None)
if all(s['exhausted'] for s in self.states):
raise StopIteration
return result, self.index
# 使用示例
a = [1, 2, 3]
b = ['a', 'b', 'c', 'd']
pi = ParallelIterator(a, b)
for values, index in pi:
print(f"索引 {index}: {values}")
基于状态的迭代控制
class StateControlledIterator:
"""基于状态控制迭代行为"""
def __init__(self, data, config=None):
self.data = data
self.config = config or {}
self.state = {
'index': 0,
'skip_count': self.config.get('skip_threshold', 2),
'error_count': 0,
'last_operation': 'init'
}
def __iter__(self):
return self
def __next__(self):
while self.state['index'] < len(self.data):
value = self.data[self.state['index']]
self.state['index'] += 1
# 基于状态的条件跳过
if self._should_skip(value):
self.state['skip_count'] -= 1
self.state['last_operation'] = 'skip'
continue
# 错误处理
if self._is_error(value):
self.state['error_count'] += 1
self.state['last_operation'] = 'error'
if self.state['error_count'] >= 3:
raise StopIteration("Too many errors")
continue
self.state['last_operation'] = 'return'
return value
raise StopIteration
def _should_skip(self, value):
"""判断是否应该跳过"""
if self.config.get('skip_values'):
return value in self.config['skip_values']
return False
def _is_error(self, value):
"""判断是否为错误值"""
return value < 0 if isinstance(value, (int, float)) else False
# 使用示例
data = [1, 2, -1, 3, 4, 5, -2, 6]
config = {'skip_values': [4], 'skip_threshold': 2}
sci = StateControlledIterator(data, config)
for value in sci:
print(f"处理值: {value}, 状态: {sci.state}")
异步迭代器状态
import asyncio
class AsyncStateIterator:
"""异步迭代器状态管理"""
def __init__(self, data):
self.data = data
self.state = {
'total': len(data),
'processed': 0,
'status': 'created',
'errors': []
}
def __aiter__(self):
self.state['status'] = 'started'
return self
async def __anext__(self):
if self.state['processed'] >= self.state['total']:
self.state['status'] = 'completed'
raise StopAsyncIteration
# 模拟异步操作
await asyncio.sleep(0.1)
value = self.data[self.state['processed']]
self.state['processed'] += 1
self.state['status'] = f'processing_{self.state["processed"]}'
return value, self.state.copy()
# 使用示例
async def main():
data = [1, 2, 3, 4, 5]
iterator = AsyncStateIterator(data)
async for value, state in iterator:
print(f"值: {value}, 状态: {state['status']}")
asyncio.run(main())
实际应用示例
class ChunkedFileIterator:
"""分块文件读取迭代器"""
def __init__(self, filename, chunk_size=1024):
self.filename = filename
self.chunk_size = chunk_size
self.state = {
'total_size': 0,
'read_size': 0,
'chunk_count': 0,
'done': False,
'last_error': None
}
def __iter__(self):
self.file = open(self.filename, 'rb')
self.state['total_size'] = self._get_file_size()
return self
def __next__(self):
if self.state['done']:
self.file.close()
raise StopIteration
chunk = self.file.read(self.chunk_size)
if not chunk:
self.state['done'] = True
self.state['status'] = 'completed'
raise StopIteration
self.state['read_size'] += len(chunk)
self.state['chunk_count'] += 1
self.state['progress'] = (self.state['read_size'] /
self.state['total_size'] * 100)
return chunk, self.state.copy()
def _get_file_size(self):
import os
return os.path.getsize(self.filename)
def __del__(self):
if hasattr(self, 'file') and not self.file.closed:
self.file.close()
# 使用示例
iterator = ChunkedFileIterator('example.txt', chunk_size=512)
for chunk, state in iterator:
print(f"读取块 {state['chunk_count']}: {state['progress']:.1f}%")
最佳实践
- 状态持久化：使用
copy()确保状态不变 - 错误处理：在状态中记录错误信息
- 性能优化：避免在每次迭代中创建新对象
- 线程安全：使用锁保护共享状态
这些方法可以根据实际需求进行组合和扩展,创建灵活的状态管理迭代器。
