1. 为什么要纠结 deque、yield、next 这三个东西先说个直觉很多 Python 教程会把collections.deque、yield、next()拆成三个独立章节来讲学的时候每个都懂一写代码就发现用不上。我早年也这样直到做实时数据流的处理才真正意识到这三个东西放到一起就是一套完整的流式数据消费框架。deque解决的是“高效缓存最近 N 条数据”的问题yield解决的是“数据生产者按需生成”的问题next()解决的是“手动驱动生产节奏”的问题。三者组合起来你能写出一套既不占内存、又能随时取数的数据处理管线。比如日志实时监控、传感器数据滑动窗口、视频帧序列缓存、任务调度队列这些场景本质上都是同一套模式一边不断产生数据一边只关心最近一段数据还要能一步步按节奏消费。这篇博客就把这套模式彻底讲透从底层原理到实际代码再到你一定会踩的坑一次说清楚。适合刚学完 Python 基础语法但不知道怎么把生成器、队列用起来的读者也适合想优化内存占用或数据流处理模式的开发者。2. 先把三个概念彻底讲透它们到底解决什么问题2.1 deque 不只是“列表的替代品”deque是collections模块里的双端队列全称是 double-ended queue。它最大的优势是两端操作都是 O(1) 时间复杂度。你用list做append(0, item)是 O(n)因为后面所有元素都要往后挪用deque.appendleft()却永远是 O(1)。底层实现上deque是一个块状双向链表结构每个块里存一组元素块与块之间用指针连接。这意味着它在两端增删极快但在中间位置做随机访问比list慢得多。你用deque的时候心里要有数它擅长当“队列”不擅长当“数组”。from collections import deque # 创建一个最大长度 3 的 deque d deque(maxlen3) d.append(1) d.append(2) d.append(3) print(d) # deque([1, 2, 3], maxlen3) # 再追加一个最左边的 1 会被自动挤掉 d.append(4) print(d) # deque([2, 3, 4], maxlen3) # 左端添加也一样 d.appendleft(0) print(d) # deque([0, 2, 3], maxlen3)maxlen这个参数是神器。它让 deque 变成了一个自动滚动的窗口不需要你手动判断长度、手动 pop 左侧元素。你只管往里塞数据超出长度后最旧的数据自动消失。这个特性在滑动窗口、实时统计场景里极其好用。2.2 yield 把函数变成了“可暂停的工厂”yield是生成器函数的核心。普通函数遇到return就彻底结束但包含yield的函数在被调用时不会立即执行函数体而是返回一个生成器对象。每次你向这个生成器要数据它才执行到下一个yield把值交出来然后原地暂停。我习惯用一个类比普通函数是食堂打饭你点完菜师傅一口气炒完端给你生成器是流水线你按一下按钮机器往前走一步、产出一个零件再按再产一个。你不再需要一次性处理一整批数据而是按需取用。def counter(): n 0 while True: n 1 yield n gen counter() print(next(gen)) # 1 print(next(gen)) # 2 print(next(gen)) # 3执行到yield n时函数会保存当前所有的局部变量状态包括n的值、代码执行到的位置。下次调用next()的时候从上次暂停的地方继续往下走。这就是生成器和普通函数最本质的区别它有状态而普通函数每次调用都是全新的。2.3 next() —— 手动控制生产节奏的油门next()是 Python 的内置函数作用是从迭代器里取下一个值。它的行为等价于调用迭代器对象的__next__()方法。两者在绝大多数场景下可以互换但next()多了一个默认值参数。gen counter() print(next(gen, None)) # 1 print(next(gen, None)) # 2next(gen, None)的意思是能取到值就取值取不到迭代器耗尽就返回None而不是抛出StopIteration。这个差异在小细节里特别有用后面实战部分我会演示。next()的意义在于它把“迭代”这件事从一个 for 循环中解放出来。for 循环本质上是不断调用next()直到StopIteration但很多时候你不想一次性吃完整条流水线而是想吃一口、干点别的、再回来吃一口。这时候就需要手动next()。2.4 三者的组合逻辑为什么刚好能配成一套单独看这三个东西威力都不大。但它们组合在一起就能构造出“有状态的队列 惰性生产 精确消费”的完整链路。数据源yield负责按需生成数据不一次性全部载入内存。临时存储deque负责缓存最近 N 条数据供消费方随时查看历史。驱动开关next负责决定“产一条新数据”还是“先看缓存”。打个比方yield 是后厨deque 是出餐口的保温柜next() 是你跟服务员说的“再上一道”。后厨不会一次性把所有菜都炒完惰性求值保温柜只放最近几道菜deque 的滑动窗口你什么时候说“下一道”后厨才动手炒新的。3. deque yield next 的四个实战场景3.1 实时日志监控模拟 tail -f 加上滑动窗口很多服务端开发者都写过日志监控脚本。需求一般是这样持续读取日志文件的新增内容并且实时输出最近 N 条异常日志。用open()读文件配合seek()太繁琐用deque yield next可以让代码很干净。import time from collections import deque def tail_follow(filepath, interval0.5): with open(filepath, r, encodingutf-8) as f: # 先跳到文件末尾 f.seek(0, 2) while True: line f.readline() if line: yield line.strip() else: time.sleep(interval) def sliding_log_viewer(filepath, window5): history deque(maxlenwindow) for line in tail_follow(filepath): history.append(line) # 只打印包含 ERROR 的行同时输出最近 window 条上下文 if ERROR in line: yield list(history) if __name__ __main__: viewer sliding_log_viewer(/tmp/app.log) # 手动控制只取 3 次 print(next(viewer, 暂时没有异常)) print(next(viewer, 暂时没有异常)) print(next(viewer, 暂时没有异常))这里tail_follow是一个无限生成器只产出新增的行。sliding_log_viewer内部维护了一个deque(maxlen5)每当检测到 ERROR 日志就把最近 5 行作为上下文输出。你用next(viewer)手动驱动想取几次取几次。这个模式真正有价值的地方在于它不会因为你“取了一次”就把整个文件读完也不会因为日志量大而把全部内容塞进内存。deque的maxlen参数天然限制了内存占用无论系统运行多久历史窗口大小恒定。注意deque的maxlen只能限制“最近 N 条”如果你需要的是“最近 N 条且按时间范围过滤”得自己再加判断逻辑deque 本身不做时间维度的事。3.2 滑动窗口均值与极值传感器数据流分析硬件开发、IoT 数据采集场景里经常需要对连续的数据流做均值、最大值、最小值计算。如果用list每次新数据进来都要del lst[0]时间复杂度是 O(n)数据密集时会非常吃力。from collections import deque class MovingStats: def __init__(self, window_size): self.window deque(maxlenwindow_size) def push(self, value): self.window.append(value) if len(self.window) self.window.maxlen: return None # 窗口未满不输出统计结果 return { avg: sum(self.window) / len(self.window), max: max(self.window), min: min(self.window) } # 模拟传感器数据 def sensor_stream(): data [12.3, 12.5, 12.8, 13.1, 12.9, 13.4, 13.7] for v in data: yield v stats MovingStats(3) for reading in sensor_stream(): result stats.push(reading) if result: print(f窗口统计: avg{result[avg]:.2f}, max{result[max]:.1f}, min{result[min]:.1f})这个场景里yield扮演了数据源的角色deque承担了窗口缓存的职责而驱动方式可以选择for循环或next()。代码里我用了for因为除了处理窗口统计之外没有其他需要手动控制next()的场景。但如果是“窗口统计 特定条件下再取数据”比如连续三个窗口的平均值都超过阈值才告警你就需要手动next()来控制节奏。这种组合用法我会在第 5 节演示。3.3 生产-消费模式手动分批处理任务队列后台任务调度里有个经典模型任务不断产生但消费端一次只处理一个或者一批。队列太长会占内存队列太短又会频繁 IO。deque当缓冲区yield当生产源next()当消费开关这个组合写起来天然匹配。from collections import deque import itertools def task_generator(): 模拟任务生成器 task_id 0 while True: task_id 1 yield {task_id: task_id, payload: fdata-{task_id}} def batch_processor(batch_size3): 批次处理器攒满 batch_size 个任务才输出一个批次 buffer deque() tasks task_generator() while True: task next(tasks) buffer.append(task) if len(buffer) batch_size: # 取出当前批次并清空缓冲 batch list(buffer) buffer.clear() yield batch processor batch_processor(3) print(next(processor)) # 第一批 3 个任务 print(next(processor)) # 第二批 3 个任务这里的关键是deque的clear()方法。它在 O(n)、实际上是 O(len(d)) 的时间内清空队列释放内部引用。重新添加数据时deque 不会保留任何上一次的数据痕迹。你可以把task_generator换成任何真实的任务源比如从 Redis 拉取任务、从 CSV 文件读取行、从 API 分页拉取数据。batch_processor的逻辑完全不变。3.4 环形缓冲固定大小的数据暂存区前面三个场景都用到了maxlen参数但你有没有想过如果maxlen不设deque会无限增长吗答案是不会不设maxlen它就会无限增长直到你把内存耗尽。但有些场景你确实需要“不自动丢弃旧数据”的环形缓冲这时候就要自己控制。from collections import deque class RingBuffer: def __init__(self, capacity): self.capacity capacity self._buffer deque() def write(self, item): if len(self._buffer) self.capacity: self._buffer.popleft() # 手动挤掉最旧的数据 self._buffer.append(item) def read_all(self): return list(self._buffer) # 结合生成器做流式写入 def data_source(): for i in range(100): yield i def ring_buffer_writer(source_gen, capacity5): rb RingBuffer(capacity) for item in source_gen: rb.write(item) if len(rb._buffer) capacity: yield rb.read_all() # 当缓冲区满时把当前内容交出去 rb._buffer.clear() # 清空以便继续写入 writer ring_buffer_writer(data_source(), capacity5) chunks list(writer) print(chunks)这里展示的是“手动管理满与不满”的写法。区别在于使用maxlen的 deque 是自动淘汰旧数据而手动 popleft 则是在 push 之前主动控制容量两者的语义有细微差别maxlen语义数据永远保留最近 N 条不管你读没读旧的直接消失。手动popleft语义你可以在弹出之前把内容取走因为有“读”的动作参与。选择哪种取决于你的需求是“缓存防丢”还是“滑窗丢弃”。4. 核心组合技巧与易踩的坑4.1 StopIteration 是朋友不是敌人StopIteration是生成器正常的结束信号。很多新手一看到异常就慌但其实它和IndexError、KeyError不一样它代表“数据源已经穷尽了”这是一个预期中的控制流事件不是错误。在手动next()的代码里我建议用默认值参数或者显式捕获不要让它裸奔到调用栈上层。gen (x for x in range(3)) print(next(gen, END)) # 0 print(next(gen, END)) # 1 print(next(gen, END)) # 2 print(next(gen, END)) # END这一点在你写“生产者-消费者”协作代码时特别重要。你不知道生成器什么时候耗尽但你不能因为耗尽就让整个程序崩溃。默认值参数最省心。注意next(gen, default)的 default 参数只在生成器正常耗尽时生效。如果生成器内部抛出了其他异常比如ValueError那么该异常仍然会向外传播default 不会拦截。4.2 生成器是一次性的deque 可以循环用生成器对象一旦被迭代完就永久耗尽了你不能“重置”它。这是一个新手特别容易踩的坑。def gen_func(): yield 1 yield 2 g gen_func() print(list(g)) # [1, 2] print(list(g)) # []第二次list(g)得到空列表因为 g 已经耗尽。但 deque 没有这个问题你随时可以清空再重用。这意味着如果某个函数接收生成器作为参数你最好不要“先检查一下再处理”因为检查用的next()已经把第一个数据消耗掉了。常见的方案是用itertools.tee()复制两份迭代器但这个操作会把元素缓存到内存数据量大时不一定划算。所以我在实际中尽量做到“一次性消费”不对生成器做两次遍历。4.3 next() 和第 4 个坑容易被忽视的 send() 和 yield fromnext(generator)本质上是generator.send(None)的简写。这意味着生成器体内能用received yield value的语法接收从外部传入的数据。一开始你可能不觉得这个有用但在需要双向通信的协程场景里这是刚需。def echo_processor(): received yield ready while True: received yield fprocessed: {received} proc echo_processor() print(next(proc)) # ready print(proc.send(hello)) # processed: hello print(proc.send(world)) # processed: world还有yield from语法它可以让一个生成器委托给另一个生成器相当于把子生成器的yield输出全部透传出去。下面这段代码里外层生成器不用手动写循环来转发数据。def inner(): yield 1 yield 2 def outer(): yield start yield from inner() yield end print(list(outer())) # [start, 1, 2, end]yield from还能处理子生成器的返回值子生成器return的值会被传回给yield from表达式本身。def inner_with_return(): yield 1 return done def outer_with_return(): result yield from inner_with_return() yield result print(list(outer_with_return())) # [1, done]这个特性在处理嵌套生成器逻辑的时候能省掉很多样板代码。我写过最爽的一次是嵌套三层数据源过滤最内层负责原始数据读取中间层负责清洗最外层负责格式化输出。用yield from链式调用代码读起来像管道一样清晰。4.4 容器嵌套时deque 的深拷贝陷阱deque里存的是元素引用。如果你往 deque 里塞的是 list 或 dict后续修改这些可变对象的内容deque 里“存的”也会跟着变。这在日志上下文的场景里尤其隐蔽。from collections import deque d deque(maxlen3) lst [1, 2] d.append(lst) lst.append(3) # 这会改变 deque 里的元素 print(d) # deque([[1, 2, 3]])如果你需要 deque 里保存的是“当时那一刻的数据快照”记得用copy.copy()或copy.deepcopy()或者干脆存不可变类型比如tuple。d deque(maxlen3) lst [1, 2] d.append(tuple(lst)) lst.append(3) print(d) # deque([(1, 2)]) # 不受影响4.5 性能差异deque 并不总比 list 快我见过很多追求性能的人把任何“最近 N 条”都换成 deque结果反而变慢了。原因是 deque 对中间位置的随机访问是 O(n)比 list 的 O(1) 慢得多如果你频繁做索引访问用list会更快。操作listdeque尾部 appendO(1)O(1)头部 append/popO(n)O(1)中间索引访问O(1)O(n)遍历所有元素O(n)O(n)内存占用紧凑数组较小块状链表较大所以我在实战中的选择逻辑是这样的只在一端操作用 list 就够了。频繁在两端操作用 deque。需要“最近 N 条”自动淘汰用 deque(maxlenN)。需要频繁随机访问中间元素用 list 或 numpy 数组。5. 实战把三者组合成一套流式处理框架5.1 需求定义假设你要做一个实时数据监控系统数据源是一个不断产生数字的生成器需求是维护最近 5 条数据作为上下文缓存。每产生一个新数据就判断当前最近 5 条的平均值是否超过阈值。只有超过阈值时才输出整个缓存上下文。消费方可以按需拉取一次输出一个上下文快照。5.2 完整代码from collections import deque import random import time def data_source(): 模拟实时数据源每 0.1 秒生成一个随机数 while True: yield random.uniform(10, 30) time.sleep(0.1) def anomaly_detector(threshold20.0, window_size5): 异常检测器 维护最近 window_size 条数据均值超过 threshold 时输出窗口快照。 使用 yield 实现惰性输出用 next() 手动驱动。 window deque(maxlenwindow_size) source data_source() while True: value next(source) # 从数据源取一个数 window.append(value) if len(window) window_size: # 窗口未满继续积累 continue avg sum(window) / len(window) if avg threshold: # 输出当前快照同时把窗口留空避免重复检测同一波数据 snapshot list(window) window.clear() yield snapshot # 使用 next() 手动拉取两个报警上下文 detector anomaly_detector(threshold20.0, window_size5) snapshot1 next(detector, None) if snapshot1: print(第一个报警上下文:, snapshot1, 平均:, sum(snapshot1) / len(snapshot1)) snapshot2 next(detector, None) if snapshot2: print(第二个报警上下文:, snapshot2, 平均:, sum(snapshot2) / len(snapshot2))这个代码同时用到了三个核心概念data_source用yield源源不断地产数据。anomaly_detector内部用deque(maxlen5)做窗口缓存。外层用next(detector, None)手动控制“什么时候要下一个异常快照”。如果你用for循环消费这个检测器程序会永远跑下去因为生成器是无限的。next()在这里本质上是给消费者一个“我要才给不要不产”的开关这在需要和其他事件循环协同开发的场景里非常有用。5.3 针对大数据量的优化策略上面的代码有个明显问题每次计算均值都用sum(window)如果窗口很大比如maxlen100000每次都要做 10 万次加法性能会很差。优化方法是用累加器维护窗口和class SlidingWindowStats: def __init__(self, window_size): self.window deque(maxlenwindow_size) self.total 0 def push(self, value): if len(self.window) self.window.maxlen: # 即将弹出最旧的元素先从累加和中减去 oldest self.window[0] self.total - oldest self.window.append(value) self.total value property def avg(self): if not self.window: return 0.0 return self.total / len(self.window)这样每次 push 都是 O(1) 操作不会因为窗口变大而变慢。这个技巧在处理实时传感器数据、高频交易数据时特别关键因为 O(n) 的求和会拖垮整个生产链路。6. 容易混淆的概念KMP 算法的 next 数组和 Python 的 next()搜索热度里频繁出现“kmp算法next数组的求法”“kmp算法next计算方法”其实这里的next和 Python 的next()完全是两码事但既然大家都在搜我顺嘴说清楚免得混淆。6.1 两个 next 完全不是一回事KMP 算法里的 next 数组是字符串匹配时用来记录“失配后模式串应该回退到哪个位置”的数组。它是一个预处理结果基于模式串自身的前缀和后缀匹配情况计算得出。比如模式串ABABC的 next 数组可能是[0, 0, 1, 2, 0]。Python 的next()则是驱动迭代器对象的方法它让迭代器吐出下一个值或抛出StopIteration。一个是数据结构数组一个是运行机制函数调用除了英文字面一样没有任何关联。6.2 容易碰到的命名冲突如果你在 Python 里也写了字符串匹配相关的算法并且把变量命名为next这就可能出问题局部变量next会遮蔽内置函数next。看这个例子pattern ABABC next_arr [0, 0, 1, 2, 0] def test(): next [0, 0, 1, 2, 0] # 局部变量遮蔽内置 next gen (x for x in range(3)) # 下一行会报 TypeError: list object is not callable # print(next(gen))解决办法很简单命名时用next_arr、next_table而不是next。如果一定要用内部名用builtins.next显式调用内置函数。import builtins gen (x for x in range(3)) value builtins.next(gen) # 强制走内置函数注意Python 的next和 C 的std::next、JavaScript 的next()方法概念上相似都是“取下一个”但用法细节完全不同。跨语言切换时要小心别用惯性思维。7. 常见问题与排查技巧实录7.1 问题next() 一直返回默认值但数据源明明没耗尽排查思路检查生成器是不是被提前部分消费了。next()每次调用只取一个值如果你在某段代码里用next()探测过一次后面再调用时第一个数据已经没了返回的可能是第二个或更后面的数据。source (x for x in [1, 2, 3, 4, 5]) # 探测第一个值 first next(source, None) print(探测到:, first) # 真正消费时第一个值已经不在了 print(实际消费:, list(source)) # [2, 3, 4, 5]如果数据源来自文件句柄或网络流这种“提前 consumption”会造成数据丢失且不可恢复。最稳妥的方案是不要提前探测用唯一的驱动点来控制消费。如果必须探测用itertools.tee()复制一份迭代器但要注意 tee 会缓存数据内存代价要自己评估。7.2 问题deque 满了但没有按预期淘汰很多人以为deque(maxlen3)会在append时自动淘汰旧数据这没错。但如果你用extend一次性添加多个元素行为会略有不同它会一口气把超长部分整体挤掉而不是逐个淘汰。from collections import deque d deque(maxlen3) d.extend([1, 2, 3, 4, 5, 6]) print(d) # deque([4, 5, 6], maxlen3)在实际代码里这个区别一般不会造成 bug但在调试时容易让人困惑为什么我明明append了 4 次只看到 1 条新数据大概率是用了extend而不是append。7.3 问题yield 生成器和 next() 在异常处理中卡死当生成器内部捕获了异常但没有正确往外抛或重置状态时外部调用next()可能会一直拿到旧数据或者卡在某个yield上。def buggy_gen(): while True: try: value yield waiting if value is None: continue if value 0: raise ValueError(负数不支持) except ValueError as e: yield ferror: {e} gen buggy_gen() print(next(gen)) # waiting print(gen.send(None)) # waiting - continue print(gen.send(5)) # 返回 5 被 yield 出去函数暂停 print(gen.send(-1)) # 捕获到 ValueError, 返回 error 字符串问题在于当yield同时出现在多个位置时生成器的状态机可能比你想象的更复杂。如果走到了一个既没有显式yield又没有return的路径生成器会直接耗尽下次next()抛StopIteration这会让外层代码以为数据源结束了其实只是某次异常分支没有正确产出值。排查技巧在生成器的每个分支末尾都放一个yield或者return保证无论走哪条路径都有明确的产出或终止行为。7.4 问题deque 无法直接序列化为 JSON把 deque 当成普通 list 去json.dumps()会报错因为 deque 不是 JSON 原生类型。import json from collections import deque d deque([1, 2, 3]) # 会报错: Object of type deque is not JSON serializable # json.dumps(d)解决方式有两种# 方式一转 list json.dumps(list(d)) # 方式二自定义 JSONEncoder class DequeEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, deque): return list(obj) return super().default(obj) json.dumps(d, clsDequeEncoder)建议在写日志输出、API 响应时明确将 deque 转成 list 再交出去。这不仅是 JSON 问题也避免下游代码误以为 deque 支持 list 的索引切片操作而产生潜在 bug。7.5 问题maxlen 为 0 的 deque 是什么行为如果你不小心把deque(maxlen0)这个队列会丢弃所有 append 进来的数据队列永远为空。这是一个隐藏陷阱特别是当maxlen由外部参数传进来时。d deque(maxlen0) d.append(1) print(d) # deque([])如果你看到代码里 deque 一直是空的先检查 maxlen 是不是被传成了 0 或负数负数和 0 行为类似。这个 bug 特别隐蔽因为完全没有报错只是数据“神秘消失”。8. 扩展结合第二个隐藏主角 send() 和 yield from8.1 为什么单独讲 send() 和 yield from如果你已经理解了next()那就不能只知道__next__()还得知道send(value)。前面聊过next(gen)等价于gen.send(None)而send(value)会把值传回生成器内部作为上一次yield表达式的结果。这个机制让生成器从“单向数据流”变成了“双向通信管道”。最常见的场景是协程外部发给生成器一个任务生成器处理完把结果 yield 回外部。8.2 结合 deque 做一个双向通信的桌面快捷键监听模拟这里模拟一个简单场景外部告诉生成器“记录一条数据”生成器在内部维护一个 deque 缓存当缓存满了就输出一个统计结果。from collections import deque def recorder(window_size5): 协程版本外部用 send 写入数据内部用 deque 缓存并统计 window deque(maxlenwindow_size) received yield ready # 等待外部第一个 send while True: if received is not None: window.append(received) if len(window) window.maxlen: avg sum(window) / len(window) received yield avg # 输出平均值并接收下一个输入 else: received yield f等待更多数据当前 {len(window)}/{window_size} else: received yield 请提供数据 rec recorder(3) print(next(rec)) # ready print(rec.send(10)) # 等待更多数据当前 1/3 print(rec.send(20)) # 等待更多数据当前 2/3 print(rec.send(30)) # 20.0 print(rec.send(40)) # 30.0这个例子是下一个层级的生成器玩法deque做窗口yield做输出send()做输入而next()只负责初始化驱动。宏观来看你得到的是一个可以“写入数据、随时取统计结果”的协程对象它天然就适合做流式处理的中转站。这个模式其实已经踩进了 Python 协程的门口后面如果再叠加asyncio.coroutine或async def、await就是完整的异步编程生态了。所以 trio、asyncio 的底层原理都可以从这个组合嗅到味道。8.3 yield from 和 deque 的配合合并多个窗口如果你有多个数据源每个都需要维护一个窗口最终要把所有窗口的结果合并起来输出yield from就很好用。def sensor_a(): for i in range(1, 10, 2): yield i def sensor_b(): for i in range(2, 11, 2): yield i def merge_windows(source_a, source_b, window_size2): window_a deque(maxlenwindow_size) window_b deque(maxlenwindow_size) gen_a source_a() gen_b source_b() while True: a_val next(gen_a, None) b_val next(gen_b, None) if a_val is None and b_val is None: break if a_val is not None: window_a.append(a_val) if b_val is not None: window_b.append(b_val) yield from _format_windows(window_a, window_b) def _format_windows(wa, wb): yield fA: {list(wa)} yield fB: {list(wb)} for line in merge_windows(sensor_a, sensor_b, window_size2): print(line)这个写法利用yield from把窗口快照逐条透传出去。通俗地说它像是一个流水线装了两条分支每个分支的最近进度被实时打包从总出口输出。纯用next()操作会啰嗦纯用 for 循环又没法在两个数据源之间快速交错deque 提供缓存yield from 提供数据转发刚好互补。9. 什么时候不要用这套组合再好的工具也有边界。deque yield next 这个组合在以下场景中并不合适需要随机访问历史数据的场景比如你要看第 10 分钟、第 20 分钟的数据deque 的索引访问是 O(n)还不如 list。数据源极小、一次性加载完全没压力的场景比如处理一个只有几百行的配置文件用生成器反而让逻辑更难懂。需要在线程/进程间共享数据队列的场景deque不是线程安全的不过你可以加锁使用Python 官方在文档里提到deque的append()、popleft()方法是原子的可以用在多线程生产者消费者模式中。这是它相对 list 的一个优势。需要持久化或序列化的场景生成器没有直接的持久化方案deque 也需要手动转 list。一句话总结这个组合适合“数据量大、一次只处理一小步、关心最近一段上下文”的场景不适合“数据量小、随便处理、关心全量历史”的场景。我在实际项目里最常用它的地方是运维脚本和数据处理管线。以前写数据处理动辄把整个文件读进内存遇到几个 GB 的日志文件直接卡死。改成yield逐行读、deque维护窗口、next()控制取数时机之后内存占用从几个 GB 降到了几 MB逻辑反而更清晰了。后来在传感器数据解析和 API 分页数据拉取里也复用了同一套思路基本没出过大问题。最后一个我建议你亲自做的小练习用deque(maxlen5)yieldnext()写一个“最近 5 个网页 URL 浏览记录”的缓存器每当有新 URL 时输出当前最近 5 条记录。写完之后再把这个缓存器放进一个for循环里观察yield的执行时机。这个练习做完你对这三个语法的掌握就不只是“见过”而是“能用了”。 Python流式数据处理：deque、yield与next的实战组合
1. 为什么要纠结 deque、yield、next 这三个东西先说个直觉很多 Python 教程会把collections.deque、yield、next()拆成三个独立章节来讲学的时候每个都懂一写代码就发现用不上。我早年也这样直到做实时数据流的处理才真正意识到这三个东西放到一起就是一套完整的流式数据消费框架。deque解决的是“高效缓存最近 N 条数据”的问题yield解决的是“数据生产者按需生成”的问题next()解决的是“手动驱动生产节奏”的问题。三者组合起来你能写出一套既不占内存、又能随时取数的数据处理管线。比如日志实时监控、传感器数据滑动窗口、视频帧序列缓存、任务调度队列这些场景本质上都是同一套模式一边不断产生数据一边只关心最近一段数据还要能一步步按节奏消费。这篇博客就把这套模式彻底讲透从底层原理到实际代码再到你一定会踩的坑一次说清楚。适合刚学完 Python 基础语法但不知道怎么把生成器、队列用起来的读者也适合想优化内存占用或数据流处理模式的开发者。2. 先把三个概念彻底讲透它们到底解决什么问题2.1 deque 不只是“列表的替代品”deque是collections模块里的双端队列全称是 double-ended queue。它最大的优势是两端操作都是 O(1) 时间复杂度。你用list做append(0, item)是 O(n)因为后面所有元素都要往后挪用deque.appendleft()却永远是 O(1)。底层实现上deque是一个块状双向链表结构每个块里存一组元素块与块之间用指针连接。这意味着它在两端增删极快但在中间位置做随机访问比list慢得多。你用deque的时候心里要有数它擅长当“队列”不擅长当“数组”。from collections import deque # 创建一个最大长度 3 的 deque d deque(maxlen3) d.append(1) d.append(2) d.append(3) print(d) # deque([1, 2, 3], maxlen3) # 再追加一个最左边的 1 会被自动挤掉 d.append(4) print(d) # deque([2, 3, 4], maxlen3) # 左端添加也一样 d.appendleft(0) print(d) # deque([0, 2, 3], maxlen3)maxlen这个参数是神器。它让 deque 变成了一个自动滚动的窗口不需要你手动判断长度、手动 pop 左侧元素。你只管往里塞数据超出长度后最旧的数据自动消失。这个特性在滑动窗口、实时统计场景里极其好用。2.2 yield 把函数变成了“可暂停的工厂”yield是生成器函数的核心。普通函数遇到return就彻底结束但包含yield的函数在被调用时不会立即执行函数体而是返回一个生成器对象。每次你向这个生成器要数据它才执行到下一个yield把值交出来然后原地暂停。我习惯用一个类比普通函数是食堂打饭你点完菜师傅一口气炒完端给你生成器是流水线你按一下按钮机器往前走一步、产出一个零件再按再产一个。你不再需要一次性处理一整批数据而是按需取用。def counter(): n 0 while True: n 1 yield n gen counter() print(next(gen)) # 1 print(next(gen)) # 2 print(next(gen)) # 3执行到yield n时函数会保存当前所有的局部变量状态包括n的值、代码执行到的位置。下次调用next()的时候从上次暂停的地方继续往下走。这就是生成器和普通函数最本质的区别它有状态而普通函数每次调用都是全新的。2.3 next() —— 手动控制生产节奏的油门next()是 Python 的内置函数作用是从迭代器里取下一个值。它的行为等价于调用迭代器对象的__next__()方法。两者在绝大多数场景下可以互换但next()多了一个默认值参数。gen counter() print(next(gen, None)) # 1 print(next(gen, None)) # 2next(gen, None)的意思是能取到值就取值取不到迭代器耗尽就返回None而不是抛出StopIteration。这个差异在小细节里特别有用后面实战部分我会演示。next()的意义在于它把“迭代”这件事从一个 for 循环中解放出来。for 循环本质上是不断调用next()直到StopIteration但很多时候你不想一次性吃完整条流水线而是想吃一口、干点别的、再回来吃一口。这时候就需要手动next()。2.4 三者的组合逻辑为什么刚好能配成一套单独看这三个东西威力都不大。但它们组合在一起就能构造出“有状态的队列 惰性生产 精确消费”的完整链路。数据源yield负责按需生成数据不一次性全部载入内存。临时存储deque负责缓存最近 N 条数据供消费方随时查看历史。驱动开关next负责决定“产一条新数据”还是“先看缓存”。打个比方yield 是后厨deque 是出餐口的保温柜next() 是你跟服务员说的“再上一道”。后厨不会一次性把所有菜都炒完惰性求值保温柜只放最近几道菜deque 的滑动窗口你什么时候说“下一道”后厨才动手炒新的。3. deque yield next 的四个实战场景3.1 实时日志监控模拟 tail -f 加上滑动窗口很多服务端开发者都写过日志监控脚本。需求一般是这样持续读取日志文件的新增内容并且实时输出最近 N 条异常日志。用open()读文件配合seek()太繁琐用deque yield next可以让代码很干净。import time from collections import deque def tail_follow(filepath, interval0.5): with open(filepath, r, encodingutf-8) as f: # 先跳到文件末尾 f.seek(0, 2) while True: line f.readline() if line: yield line.strip() else: time.sleep(interval) def sliding_log_viewer(filepath, window5): history deque(maxlenwindow) for line in tail_follow(filepath): history.append(line) # 只打印包含 ERROR 的行同时输出最近 window 条上下文 if ERROR in line: yield list(history) if __name__ __main__: viewer sliding_log_viewer(/tmp/app.log) # 手动控制只取 3 次 print(next(viewer, 暂时没有异常)) print(next(viewer, 暂时没有异常)) print(next(viewer, 暂时没有异常))这里tail_follow是一个无限生成器只产出新增的行。sliding_log_viewer内部维护了一个deque(maxlen5)每当检测到 ERROR 日志就把最近 5 行作为上下文输出。你用next(viewer)手动驱动想取几次取几次。这个模式真正有价值的地方在于它不会因为你“取了一次”就把整个文件读完也不会因为日志量大而把全部内容塞进内存。deque的maxlen参数天然限制了内存占用无论系统运行多久历史窗口大小恒定。注意deque的maxlen只能限制“最近 N 条”如果你需要的是“最近 N 条且按时间范围过滤”得自己再加判断逻辑deque 本身不做时间维度的事。3.2 滑动窗口均值与极值传感器数据流分析硬件开发、IoT 数据采集场景里经常需要对连续的数据流做均值、最大值、最小值计算。如果用list每次新数据进来都要del lst[0]时间复杂度是 O(n)数据密集时会非常吃力。from collections import deque class MovingStats: def __init__(self, window_size): self.window deque(maxlenwindow_size) def push(self, value): self.window.append(value) if len(self.window) self.window.maxlen: return None # 窗口未满不输出统计结果 return { avg: sum(self.window) / len(self.window), max: max(self.window), min: min(self.window) } # 模拟传感器数据 def sensor_stream(): data [12.3, 12.5, 12.8, 13.1, 12.9, 13.4, 13.7] for v in data: yield v stats MovingStats(3) for reading in sensor_stream(): result stats.push(reading) if result: print(f窗口统计: avg{result[avg]:.2f}, max{result[max]:.1f}, min{result[min]:.1f})这个场景里yield扮演了数据源的角色deque承担了窗口缓存的职责而驱动方式可以选择for循环或next()。代码里我用了for因为除了处理窗口统计之外没有其他需要手动控制next()的场景。但如果是“窗口统计 特定条件下再取数据”比如连续三个窗口的平均值都超过阈值才告警你就需要手动next()来控制节奏。这种组合用法我会在第 5 节演示。3.3 生产-消费模式手动分批处理任务队列后台任务调度里有个经典模型任务不断产生但消费端一次只处理一个或者一批。队列太长会占内存队列太短又会频繁 IO。deque当缓冲区yield当生产源next()当消费开关这个组合写起来天然匹配。from collections import deque import itertools def task_generator(): 模拟任务生成器 task_id 0 while True: task_id 1 yield {task_id: task_id, payload: fdata-{task_id}} def batch_processor(batch_size3): 批次处理器攒满 batch_size 个任务才输出一个批次 buffer deque() tasks task_generator() while True: task next(tasks) buffer.append(task) if len(buffer) batch_size: # 取出当前批次并清空缓冲 batch list(buffer) buffer.clear() yield batch processor batch_processor(3) print(next(processor)) # 第一批 3 个任务 print(next(processor)) # 第二批 3 个任务这里的关键是deque的clear()方法。它在 O(n)、实际上是 O(len(d)) 的时间内清空队列释放内部引用。重新添加数据时deque 不会保留任何上一次的数据痕迹。你可以把task_generator换成任何真实的任务源比如从 Redis 拉取任务、从 CSV 文件读取行、从 API 分页拉取数据。batch_processor的逻辑完全不变。3.4 环形缓冲固定大小的数据暂存区前面三个场景都用到了maxlen参数但你有没有想过如果maxlen不设deque会无限增长吗答案是不会不设maxlen它就会无限增长直到你把内存耗尽。但有些场景你确实需要“不自动丢弃旧数据”的环形缓冲这时候就要自己控制。from collections import deque class RingBuffer: def __init__(self, capacity): self.capacity capacity self._buffer deque() def write(self, item): if len(self._buffer) self.capacity: self._buffer.popleft() # 手动挤掉最旧的数据 self._buffer.append(item) def read_all(self): return list(self._buffer) # 结合生成器做流式写入 def data_source(): for i in range(100): yield i def ring_buffer_writer(source_gen, capacity5): rb RingBuffer(capacity) for item in source_gen: rb.write(item) if len(rb._buffer) capacity: yield rb.read_all() # 当缓冲区满时把当前内容交出去 rb._buffer.clear() # 清空以便继续写入 writer ring_buffer_writer(data_source(), capacity5) chunks list(writer) print(chunks)这里展示的是“手动管理满与不满”的写法。区别在于使用maxlen的 deque 是自动淘汰旧数据而手动 popleft 则是在 push 之前主动控制容量两者的语义有细微差别maxlen语义数据永远保留最近 N 条不管你读没读旧的直接消失。手动popleft语义你可以在弹出之前把内容取走因为有“读”的动作参与。选择哪种取决于你的需求是“缓存防丢”还是“滑窗丢弃”。4. 核心组合技巧与易踩的坑4.1 StopIteration 是朋友不是敌人StopIteration是生成器正常的结束信号。很多新手一看到异常就慌但其实它和IndexError、KeyError不一样它代表“数据源已经穷尽了”这是一个预期中的控制流事件不是错误。在手动next()的代码里我建议用默认值参数或者显式捕获不要让它裸奔到调用栈上层。gen (x for x in range(3)) print(next(gen, END)) # 0 print(next(gen, END)) # 1 print(next(gen, END)) # 2 print(next(gen, END)) # END这一点在你写“生产者-消费者”协作代码时特别重要。你不知道生成器什么时候耗尽但你不能因为耗尽就让整个程序崩溃。默认值参数最省心。注意next(gen, default)的 default 参数只在生成器正常耗尽时生效。如果生成器内部抛出了其他异常比如ValueError那么该异常仍然会向外传播default 不会拦截。4.2 生成器是一次性的deque 可以循环用生成器对象一旦被迭代完就永久耗尽了你不能“重置”它。这是一个新手特别容易踩的坑。def gen_func(): yield 1 yield 2 g gen_func() print(list(g)) # [1, 2] print(list(g)) # []第二次list(g)得到空列表因为 g 已经耗尽。但 deque 没有这个问题你随时可以清空再重用。这意味着如果某个函数接收生成器作为参数你最好不要“先检查一下再处理”因为检查用的next()已经把第一个数据消耗掉了。常见的方案是用itertools.tee()复制两份迭代器但这个操作会把元素缓存到内存数据量大时不一定划算。所以我在实际中尽量做到“一次性消费”不对生成器做两次遍历。4.3 next() 和第 4 个坑容易被忽视的 send() 和 yield fromnext(generator)本质上是generator.send(None)的简写。这意味着生成器体内能用received yield value的语法接收从外部传入的数据。一开始你可能不觉得这个有用但在需要双向通信的协程场景里这是刚需。def echo_processor(): received yield ready while True: received yield fprocessed: {received} proc echo_processor() print(next(proc)) # ready print(proc.send(hello)) # processed: hello print(proc.send(world)) # processed: world还有yield from语法它可以让一个生成器委托给另一个生成器相当于把子生成器的yield输出全部透传出去。下面这段代码里外层生成器不用手动写循环来转发数据。def inner(): yield 1 yield 2 def outer(): yield start yield from inner() yield end print(list(outer())) # [start, 1, 2, end]yield from还能处理子生成器的返回值子生成器return的值会被传回给yield from表达式本身。def inner_with_return(): yield 1 return done def outer_with_return(): result yield from inner_with_return() yield result print(list(outer_with_return())) # [1, done]这个特性在处理嵌套生成器逻辑的时候能省掉很多样板代码。我写过最爽的一次是嵌套三层数据源过滤最内层负责原始数据读取中间层负责清洗最外层负责格式化输出。用yield from链式调用代码读起来像管道一样清晰。4.4 容器嵌套时deque 的深拷贝陷阱deque里存的是元素引用。如果你往 deque 里塞的是 list 或 dict后续修改这些可变对象的内容deque 里“存的”也会跟着变。这在日志上下文的场景里尤其隐蔽。from collections import deque d deque(maxlen3) lst [1, 2] d.append(lst) lst.append(3) # 这会改变 deque 里的元素 print(d) # deque([[1, 2, 3]])如果你需要 deque 里保存的是“当时那一刻的数据快照”记得用copy.copy()或copy.deepcopy()或者干脆存不可变类型比如tuple。d deque(maxlen3) lst [1, 2] d.append(tuple(lst)) lst.append(3) print(d) # deque([(1, 2)]) # 不受影响4.5 性能差异deque 并不总比 list 快我见过很多追求性能的人把任何“最近 N 条”都换成 deque结果反而变慢了。原因是 deque 对中间位置的随机访问是 O(n)比 list 的 O(1) 慢得多如果你频繁做索引访问用list会更快。操作listdeque尾部 appendO(1)O(1)头部 append/popO(n)O(1)中间索引访问O(1)O(n)遍历所有元素O(n)O(n)内存占用紧凑数组较小块状链表较大所以我在实战中的选择逻辑是这样的只在一端操作用 list 就够了。频繁在两端操作用 deque。需要“最近 N 条”自动淘汰用 deque(maxlenN)。需要频繁随机访问中间元素用 list 或 numpy 数组。5. 实战把三者组合成一套流式处理框架5.1 需求定义假设你要做一个实时数据监控系统数据源是一个不断产生数字的生成器需求是维护最近 5 条数据作为上下文缓存。每产生一个新数据就判断当前最近 5 条的平均值是否超过阈值。只有超过阈值时才输出整个缓存上下文。消费方可以按需拉取一次输出一个上下文快照。5.2 完整代码from collections import deque import random import time def data_source(): 模拟实时数据源每 0.1 秒生成一个随机数 while True: yield random.uniform(10, 30) time.sleep(0.1) def anomaly_detector(threshold20.0, window_size5): 异常检测器 维护最近 window_size 条数据均值超过 threshold 时输出窗口快照。 使用 yield 实现惰性输出用 next() 手动驱动。 window deque(maxlenwindow_size) source data_source() while True: value next(source) # 从数据源取一个数 window.append(value) if len(window) window_size: # 窗口未满继续积累 continue avg sum(window) / len(window) if avg threshold: # 输出当前快照同时把窗口留空避免重复检测同一波数据 snapshot list(window) window.clear() yield snapshot # 使用 next() 手动拉取两个报警上下文 detector anomaly_detector(threshold20.0, window_size5) snapshot1 next(detector, None) if snapshot1: print(第一个报警上下文:, snapshot1, 平均:, sum(snapshot1) / len(snapshot1)) snapshot2 next(detector, None) if snapshot2: print(第二个报警上下文:, snapshot2, 平均:, sum(snapshot2) / len(snapshot2))这个代码同时用到了三个核心概念data_source用yield源源不断地产数据。anomaly_detector内部用deque(maxlen5)做窗口缓存。外层用next(detector, None)手动控制“什么时候要下一个异常快照”。如果你用for循环消费这个检测器程序会永远跑下去因为生成器是无限的。next()在这里本质上是给消费者一个“我要才给不要不产”的开关这在需要和其他事件循环协同开发的场景里非常有用。5.3 针对大数据量的优化策略上面的代码有个明显问题每次计算均值都用sum(window)如果窗口很大比如maxlen100000每次都要做 10 万次加法性能会很差。优化方法是用累加器维护窗口和class SlidingWindowStats: def __init__(self, window_size): self.window deque(maxlenwindow_size) self.total 0 def push(self, value): if len(self.window) self.window.maxlen: # 即将弹出最旧的元素先从累加和中减去 oldest self.window[0] self.total - oldest self.window.append(value) self.total value property def avg(self): if not self.window: return 0.0 return self.total / len(self.window)这样每次 push 都是 O(1) 操作不会因为窗口变大而变慢。这个技巧在处理实时传感器数据、高频交易数据时特别关键因为 O(n) 的求和会拖垮整个生产链路。6. 容易混淆的概念KMP 算法的 next 数组和 Python 的 next()搜索热度里频繁出现“kmp算法next数组的求法”“kmp算法next计算方法”其实这里的next和 Python 的next()完全是两码事但既然大家都在搜我顺嘴说清楚免得混淆。6.1 两个 next 完全不是一回事KMP 算法里的 next 数组是字符串匹配时用来记录“失配后模式串应该回退到哪个位置”的数组。它是一个预处理结果基于模式串自身的前缀和后缀匹配情况计算得出。比如模式串ABABC的 next 数组可能是[0, 0, 1, 2, 0]。Python 的next()则是驱动迭代器对象的方法它让迭代器吐出下一个值或抛出StopIteration。一个是数据结构数组一个是运行机制函数调用除了英文字面一样没有任何关联。6.2 容易碰到的命名冲突如果你在 Python 里也写了字符串匹配相关的算法并且把变量命名为next这就可能出问题局部变量next会遮蔽内置函数next。看这个例子pattern ABABC next_arr [0, 0, 1, 2, 0] def test(): next [0, 0, 1, 2, 0] # 局部变量遮蔽内置 next gen (x for x in range(3)) # 下一行会报 TypeError: list object is not callable # print(next(gen))解决办法很简单命名时用next_arr、next_table而不是next。如果一定要用内部名用builtins.next显式调用内置函数。import builtins gen (x for x in range(3)) value builtins.next(gen) # 强制走内置函数注意Python 的next和 C 的std::next、JavaScript 的next()方法概念上相似都是“取下一个”但用法细节完全不同。跨语言切换时要小心别用惯性思维。7. 常见问题与排查技巧实录7.1 问题next() 一直返回默认值但数据源明明没耗尽排查思路检查生成器是不是被提前部分消费了。next()每次调用只取一个值如果你在某段代码里用next()探测过一次后面再调用时第一个数据已经没了返回的可能是第二个或更后面的数据。source (x for x in [1, 2, 3, 4, 5]) # 探测第一个值 first next(source, None) print(探测到:, first) # 真正消费时第一个值已经不在了 print(实际消费:, list(source)) # [2, 3, 4, 5]如果数据源来自文件句柄或网络流这种“提前 consumption”会造成数据丢失且不可恢复。最稳妥的方案是不要提前探测用唯一的驱动点来控制消费。如果必须探测用itertools.tee()复制一份迭代器但要注意 tee 会缓存数据内存代价要自己评估。7.2 问题deque 满了但没有按预期淘汰很多人以为deque(maxlen3)会在append时自动淘汰旧数据这没错。但如果你用extend一次性添加多个元素行为会略有不同它会一口气把超长部分整体挤掉而不是逐个淘汰。from collections import deque d deque(maxlen3) d.extend([1, 2, 3, 4, 5, 6]) print(d) # deque([4, 5, 6], maxlen3)在实际代码里这个区别一般不会造成 bug但在调试时容易让人困惑为什么我明明append了 4 次只看到 1 条新数据大概率是用了extend而不是append。7.3 问题yield 生成器和 next() 在异常处理中卡死当生成器内部捕获了异常但没有正确往外抛或重置状态时外部调用next()可能会一直拿到旧数据或者卡在某个yield上。def buggy_gen(): while True: try: value yield waiting if value is None: continue if value 0: raise ValueError(负数不支持) except ValueError as e: yield ferror: {e} gen buggy_gen() print(next(gen)) # waiting print(gen.send(None)) # waiting - continue print(gen.send(5)) # 返回 5 被 yield 出去函数暂停 print(gen.send(-1)) # 捕获到 ValueError, 返回 error 字符串问题在于当yield同时出现在多个位置时生成器的状态机可能比你想象的更复杂。如果走到了一个既没有显式yield又没有return的路径生成器会直接耗尽下次next()抛StopIteration这会让外层代码以为数据源结束了其实只是某次异常分支没有正确产出值。排查技巧在生成器的每个分支末尾都放一个yield或者return保证无论走哪条路径都有明确的产出或终止行为。7.4 问题deque 无法直接序列化为 JSON把 deque 当成普通 list 去json.dumps()会报错因为 deque 不是 JSON 原生类型。import json from collections import deque d deque([1, 2, 3]) # 会报错: Object of type deque is not JSON serializable # json.dumps(d)解决方式有两种# 方式一转 list json.dumps(list(d)) # 方式二自定义 JSONEncoder class DequeEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, deque): return list(obj) return super().default(obj) json.dumps(d, clsDequeEncoder)建议在写日志输出、API 响应时明确将 deque 转成 list 再交出去。这不仅是 JSON 问题也避免下游代码误以为 deque 支持 list 的索引切片操作而产生潜在 bug。7.5 问题maxlen 为 0 的 deque 是什么行为如果你不小心把deque(maxlen0)这个队列会丢弃所有 append 进来的数据队列永远为空。这是一个隐藏陷阱特别是当maxlen由外部参数传进来时。d deque(maxlen0) d.append(1) print(d) # deque([])如果你看到代码里 deque 一直是空的先检查 maxlen 是不是被传成了 0 或负数负数和 0 行为类似。这个 bug 特别隐蔽因为完全没有报错只是数据“神秘消失”。8. 扩展结合第二个隐藏主角 send() 和 yield from8.1 为什么单独讲 send() 和 yield from如果你已经理解了next()那就不能只知道__next__()还得知道send(value)。前面聊过next(gen)等价于gen.send(None)而send(value)会把值传回生成器内部作为上一次yield表达式的结果。这个机制让生成器从“单向数据流”变成了“双向通信管道”。最常见的场景是协程外部发给生成器一个任务生成器处理完把结果 yield 回外部。8.2 结合 deque 做一个双向通信的桌面快捷键监听模拟这里模拟一个简单场景外部告诉生成器“记录一条数据”生成器在内部维护一个 deque 缓存当缓存满了就输出一个统计结果。from collections import deque def recorder(window_size5): 协程版本外部用 send 写入数据内部用 deque 缓存并统计 window deque(maxlenwindow_size) received yield ready # 等待外部第一个 send while True: if received is not None: window.append(received) if len(window) window.maxlen: avg sum(window) / len(window) received yield avg # 输出平均值并接收下一个输入 else: received yield f等待更多数据当前 {len(window)}/{window_size} else: received yield 请提供数据 rec recorder(3) print(next(rec)) # ready print(rec.send(10)) # 等待更多数据当前 1/3 print(rec.send(20)) # 等待更多数据当前 2/3 print(rec.send(30)) # 20.0 print(rec.send(40)) # 30.0这个例子是下一个层级的生成器玩法deque做窗口yield做输出send()做输入而next()只负责初始化驱动。宏观来看你得到的是一个可以“写入数据、随时取统计结果”的协程对象它天然就适合做流式处理的中转站。这个模式其实已经踩进了 Python 协程的门口后面如果再叠加asyncio.coroutine或async def、await就是完整的异步编程生态了。所以 trio、asyncio 的底层原理都可以从这个组合嗅到味道。8.3 yield from 和 deque 的配合合并多个窗口如果你有多个数据源每个都需要维护一个窗口最终要把所有窗口的结果合并起来输出yield from就很好用。def sensor_a(): for i in range(1, 10, 2): yield i def sensor_b(): for i in range(2, 11, 2): yield i def merge_windows(source_a, source_b, window_size2): window_a deque(maxlenwindow_size) window_b deque(maxlenwindow_size) gen_a source_a() gen_b source_b() while True: a_val next(gen_a, None) b_val next(gen_b, None) if a_val is None and b_val is None: break if a_val is not None: window_a.append(a_val) if b_val is not None: window_b.append(b_val) yield from _format_windows(window_a, window_b) def _format_windows(wa, wb): yield fA: {list(wa)} yield fB: {list(wb)} for line in merge_windows(sensor_a, sensor_b, window_size2): print(line)这个写法利用yield from把窗口快照逐条透传出去。通俗地说它像是一个流水线装了两条分支每个分支的最近进度被实时打包从总出口输出。纯用next()操作会啰嗦纯用 for 循环又没法在两个数据源之间快速交错deque 提供缓存yield from 提供数据转发刚好互补。9. 什么时候不要用这套组合再好的工具也有边界。deque yield next 这个组合在以下场景中并不合适需要随机访问历史数据的场景比如你要看第 10 分钟、第 20 分钟的数据deque 的索引访问是 O(n)还不如 list。数据源极小、一次性加载完全没压力的场景比如处理一个只有几百行的配置文件用生成器反而让逻辑更难懂。需要在线程/进程间共享数据队列的场景deque不是线程安全的不过你可以加锁使用Python 官方在文档里提到deque的append()、popleft()方法是原子的可以用在多线程生产者消费者模式中。这是它相对 list 的一个优势。需要持久化或序列化的场景生成器没有直接的持久化方案deque 也需要手动转 list。一句话总结这个组合适合“数据量大、一次只处理一小步、关心最近一段上下文”的场景不适合“数据量小、随便处理、关心全量历史”的场景。我在实际项目里最常用它的地方是运维脚本和数据处理管线。以前写数据处理动辄把整个文件读进内存遇到几个 GB 的日志文件直接卡死。改成yield逐行读、deque维护窗口、next()控制取数时机之后内存占用从几个 GB 降到了几 MB逻辑反而更清晰了。后来在传感器数据解析和 API 分页数据拉取里也复用了同一套思路基本没出过大问题。最后一个我建议你亲自做的小练习用deque(maxlen5)yieldnext()写一个“最近 5 个网页 URL 浏览记录”的缓存器每当有新 URL 时输出当前最近 5 条记录。写完之后再把这个缓存器放进一个for循环里观察yield的执行时机。这个练习做完你对这三个语法的掌握就不只是“见过”而是“能用了”。