处理无限数据流需“按需取值+主动终止”，迭代器天然适用；生成器函数用yield实现惰性无限迭代，内存恒定且可中断；itertools.islice可安全截断；终止逻辑可封装进迭代器内部并抛stopiteration。
不能靠“跑完”来处理无限数据流，关键在于“按需取值 + 主动终止”。迭代器天然适合这种场景，因为它不预加载全部数据，只在每次 next() 或循环中 for 的一次迭代时生成一个值。
用生成器函数构造惰性无限迭代器
生成器是最常用、最自然的方式。它用 yield 暂停并返回当前值，状态自动保存，下次调用继续执行。内存占用恒定，且可随时用 break、return 或异常中断。
- Python 示例：每秒产出时间戳，取 5 次后停止
import time
def clock_stream():
while True:
yield time.time()
time.sleep(1)
for i, ts in enumerate(clock_stream()):
print(f"Tick {i+1}: {ts:.2f}")
if i >= 4:
break
用 itertools 工具做安全截断
面对已有的无限迭代器（如 itertools.count()、自定义素数生成器），不必自己写计数逻辑。直接用 itertools.islice() 限定取前 N 个，简洁又不易出错。
- islice(iterator, n) 返回前 n 个元素，到头自动停止，不会抛 StopIteration
- 配合 list() 或 for 直接消费，无需手动管理计数或异常
from itertools import count, islice
# 无限计数器
infinite_counter = count(start=100, step=10)
# 只取前 3 个：100, 110, 120
for x in islice(infinite_counter, 3):
print(x)
封装终止条件到迭代器内部
如果希望迭代器本身具备“智能结束”能力（比如超时、阈值、信号响应），可以把控制逻辑写进生成器或自定义迭代器的 __next__() 中。
- 检查时间戳是否超限，或累计次数是否达上限
- 遇到终止条件时，直接 raise StopIteration，符合 Python 迭代协议
- 这样外部代码可直接用 for，无需额外判断
def limited_clock(max_seconds=5):
start = time.time()
while time.time() - start yield time.time()
time.sleep(1)
# 自动在 5 秒后停止
for t in limited_clock():
print(f"Alive at {t:.1f}")
生产环境需加固：超时、异常隔离与背压
真实系统中，无限流常来自传感器、日志、网络事件等，必须防止单次卡死或资源耗尽：
- 给 yield 前的操作加超时（如 asyncio.wait_for 或信号 alarm）
- 把可能失败的数据源操作包在 try/except 内，避免整个迭代崩溃
- 若消费者处理慢，考虑引入简单背压，比如 yield 前检查缓冲区长度或用 queue.Full 控制
