我是一个平时喜欢折腾数据管线的工程师今天想好好聊聊数据流式编程里最基础也最关键的一个概念——执行单元。这个词听起来有点学院派但说白了它就是数据流里那个真正干活的节点收一条数据进来处理一下再把结果吐给下一个环节。不管你是接触过 Flink、Kafka Streams还是用过 Node-RED、RxJS甚至只是写过一条简单的异步 Promise 链背后都离不开执行单元的这套逻辑。这篇文章我会从设计原理讲到手写实现再聊一聊我在实际项目中踩过的坑。适合正在学习流式计算、想自己搭建轻量级数据处理框架、或者被异步并发搞得头疼的同学。概念我会讲得尽量通俗代码也给了可直接跑的版本保证不是那种看完就忘的科普。1. 数据流式编程的骨架执行单元到底解决了什么问题1.1 先从一个让我头疼的场景说起之前我维护过一个日志清洗服务逻辑不算复杂从消息队列拿原始日志解析成结构化字段过滤掉无用的调试信息做字段补全最后写到存储里。听起来很直接吧但用传统命令式写法没过多久代码就成了一团乱麻。核心问题不是逻辑本身难而是每一步之间耦合得太紧。解析方法要等拿到输入才执行过滤逻辑散落在各处如果某一环节需要并行处理还得手动管理线程和队列。后来我把这段逻辑用数据流的方式重写每个处理阶段拆成一个执行单元单元之间只通过消息传递数据。改动之后代码结构清晰了很多每个单元可以独立测试并且随时能在中间插入新功能比如加一个脱敏节点完全不用动上下游逻辑。这个经历其实反映了执行单元的核心价值它将一段连续的处理过程拆成了离散、可组合、可独立伸缩的步骤。每一步接收输入、处理、发射输出至于上游是谁、下游是谁单元本身不关心只要数据格式匹配就行。1.2 执行单元到底长什么样从抽象角度看一个执行单元通常具备这三样东西输入端口接收上游传来的数据。简单场景只有一个输入端口复杂场景可能有多个比如双流 join 就需要两个输入。输出端口把处理结果发给下游。大多数单元是单输出的但也存在条件分支单元按规则把数据路由到不同输出。生命周期方法负责初始化资源、处理单条数据、释放资源。我把它们统称为init、process、dispose。很多流处理框架里概念叫法不同但本质上是一致的Flink 管它叫算子OperatorAkka Streams 里叫 StageNode-RED 里叫 NodeJava Stream 里叫中间操作。换个马甲而已。有一个关键点值得强调执行单元的理想状态是无副作用的也就是它不修改外部共享变量只依赖输入和自身保存的状态。这样做的好处是单元可以随意移动、复制、并行扩展而不会产生奇怪的竞态问题。1.3 它和 Actor 模型、流水线、响应式编程的关系很多初学者搞不清数据流、Actor、响应式编程的区别我整理过一张熟悉的关系网概念侧重点和执行单元的关系Actor 模型并发、消息传递、状态隔离每个 Actor 可以看作一个带邮箱的执行单元单元间完全通过消息通信流水线阶段性加工、并行处理执行单元的有序组合就是一条数据流水线响应式编程异步数据流、声明式组合map/filter 等操作符就是现成的执行单元数据流编程图结构、数据驱动执行执行单元是图的节点边是数据通道这是最上层的抽象打个比方执行单元就像一条自动生产线上的工位。工人只管把自己工位上的活干完然后放到传送带上给下一个工位。传送带怎么调度、哪个工位多安排几个人、哪个工位需要缓冲库存这是框架层面要解决的问题。2. 执行单元的核心机制拆解2.1 输入输出契约消息与端口怎么定义才不容易踩坑执行单元之间的接口设计决定了整个数据流框架的灵活性和健壮性。我在这次实践里采用的是结构化消息而非裸对象每条消息包含三部分dataclass class StreamMessage: key: str # 分区键用于确定路由到哪个实例处理 data: Any # 实际负载 timestamp: float # 事件发生时间用于乱序处理和延迟统计为什么要带key和timestamp因为分布式场景下仅靠消息内容无法决定“这条数据该交给哪个执行单元实例”和“这条数据是不是来晚了”。很多初学者在设计接口时只留一个payload字段后面做并行分区和时间窗口时才发现缺信息被迫改接口。输入端口个数也有讲究。常规单元是单输入单输出map、filter 都是这种但多输入场景并不罕见。我在自己的框架里允许单元声明多个输入端口每个端口绑定一个上游输出内部用select多路复用接收数据。这样做能处理类似“订单流和支付流做关联”的需求不需要额外引入 join 算子。输出这块我建议至少给出emit和emit_many两个方法。前者发一条后者批量发。批量发在吞吐量要求高时很有用单条发射的方式调用开销太大。实测在 Python 环境中批量发射能减少大概 20%30% 的调度开销。2.2 状态管理执行单元里有状态怎么保证不出乱子执行单元分无状态和有状态两种。无状态单元很容易理解输入什么就输出什么比如格式转换、字段过滤。有状态单元就复杂了它需要在处理消息的同时维护内部数据例如滚动窗口计算、去重、计数。状态存放有三种常见策略纯内存状态状态存在对象字段里比如self.counter。实现最简单但一旦进程重启状态全丢只适合允许丢失的场景。外部存储状态把状态写到 Redis、数据库里。可靠性高但每次读写都有 IO 延迟性能会下降不少。本地持久化状态像 Flink 的 RocksDB 后端那样把状态保存在本地嵌入式数据库。可靠性和性能平衡得比较好但实现复杂。我在轻量框架里用的是纯内存状态但特意加了一个约束只有带相同 key 的数据才会被路由到同一个执行单元实例。这就把并发问题缩小到单实例内部避免两个线程同时修改同一个计数器的尴尬。这里给新手一个经验设计有状态执行单元时务必想清楚两个问题——数据按什么维度分组、状态生命周期从哪开始到哪结束。比如做一个“过去 5 分钟每个用户下单总金额”的计算器分组维度是用户 ID状态生命周期从第一次收到该用户消息开始到窗口结束释放。没有这两个答案代码怎么写都会别扭。2.3 调度模型推式、拉式和背压到底选哪种执行单元之间怎么传递数据是数据流框架最核心的决策之一。我遇到过三种模型推式Push上游处理完直接调用下游的process。实现简单响应快但问题是一旦下游处理速度跟不上数据会堆积在内存里最终 OOM。拉式Pull下游主动向上游要数据。下游消费多少上游才生产多少天然带背压。但实现起来要处理“上游没数据时下游等多久”的问题延迟偏高。动态推拉混合正常情况下上游推当下游缓冲队列达到阈值时触发反压信号通知上游停下来等一会儿。这是目前主流流处理引擎的通用方案。我的实践中背压不是理论问题而是真实血泪。最早版本用的是纯推式每条消息都立刻传给下游结果下游一个慢速的磁盘写入把内存直接打爆。后来我把连接单元之间的通道设计成有界队列当队列满时上游写入会被阻塞自然形成背压。这里用餐厅传菜做类比厨师做好菜放到出餐口传菜员取走去上菜。如果出餐口只有一个位置厨师就得放慢节奏等传菜员如果出餐口无限大菜就会堆满整个厨房。有界队列就是那个“只有一个位置”的出餐口用阻塞换取系统的稳定性。2.4 并行执行与数据分区为什么不能简单多开几个线程单个执行单元处理速度有上限于是自然想到并行让多个单元实例同时干活。但并行不是简单地把一个单元复制两份而是要处理数据该怎么分配的问题。我的做法是引入分区键就是消息里的key字段。框架内部维护多个单元实例按照hash(key) % 实例数把消息路由到不同实例。这样做的意义是同一条 key 的数据始终进入同一个实例局部有序性得到保证每个实例内部的状态也不会被别的实例破坏。刚开始并行时我犯过一个错误为了让某个无状态单元跑快点我把它复制成 4 个实例数据随机分配到各个实例。结果下游刚好有一个去重单元依赖上游“同一用户记录只经过一个实例”的假设导致同一个用户的两条记录被不同实例处理去重失效统计数字直接翻倍。修复方法就是在无状态单元上也保留分区路由保证相同 key 的消息走同一条路径。并行度也并非越大越好。每个实例都有调度和内存开销而且如果单元里访问了外部系统并行度过高还会压垮下游数据库。我一般遵循一个经验法则先看单个实例的吞吐瓶颈在 CPU 还是 IO瓶颈在 CPU 就按核数设置并行度瓶颈在 IO 就适当调高但给下游留 30% 余量。3. 实操手写一个轻量级执行单元框架3.1 框架整体设计思路这部分是全文的重头戏。目标不是把它做成生产级引擎而是让你看完后能理解流处理框架的核心环节是怎么串起来的。我选 Python 语言原因有三一是大家读起来没门槛二是 Python 的asyncio.Queue天然支持有界队列和背压正好和前面的调度模型对得上三是写单元测试方便。框架包含四个核心类StreamMessage消息载体包含 key、data、timestamp。BaseUnit执行单元基类声明生命周期钩子和输入输出逻辑。Pipeline管道类负责连接单元、调度执行、分配消息。Source特殊执行单元没有输入只有输出是数据流的起点。整体结构是Source 产生数据经过一系列 BaseUnit最终到达 Sink一个只收不发的特殊单元。Pipeline 内部维护一条有向无环的执行链数据从头流到尾。3.2 骨架代码执行单元和平管道的实现首先定义消息和执行单元基类import asyncio from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Optional dataclass class StreamMessage: 数据流中传递的消息体。key 用于分区路由data 是实际内容。 key: str data: Any timestamp: float 0.0 class BaseUnit(ABC): def __init__(self, parallelism: int 1): self.parallelism parallelism self._outputs: list[asyncio.Queue] [] self._event_loop: Optional[asyncio.AbstractEventLoop] None def bind(self, downstream_queue: asyncio.Queue, loop: asyncio.AbstractEventLoop): 把当前单元的输出绑定到一个下游队列。 self._outputs.append(downstream_queue) self._event_loop loop async def emit(self, msg: StreamMessage): 向下游发射一条消息。队列满时会自动阻塞形成背压。 for queue in self._outputs: await queue.put(msg) async def init(self): 初始化钩子比如建立数据库连接、加载模型。 pass abstractmethod async def process(self, msg: StreamMessage): 处理单条消息这是执行单元的核心。 pass async def dispose(self): 释放资源钩子。 pass这段代码里最关键的是emit方法。它不像传统做法那样直接调用下游的process而是把消息放进一个有界队列。如果下游处理慢队列空间不足await queue.put(msg)就会挂起整个链路自然减速而不会导致内存无限增长。接着是单元实现示例。我写一个字段提取、一个过滤、一个聚合窗口class ExtractUnit(BaseUnit): 从原始日志中解析出关键字段。 async def process(self, msg: StreamMessage): raw msg.data parsed { user_id: raw[user_id], action: raw[action], cost: float(raw.get(cost, 0)), ts: raw[timestamp], } await self.emit(StreamMessage(keymsg.key, dataparsed, timestampmsg.timestamp)) class FilterUnit(BaseUnit): 过滤掉 action 为 ping 的探活日志。 async def process(self, msg: StreamMessage): if msg.data[action] ! ping: await self.emit(msg) class SumWindowUnit(BaseUnit): 按 user_id 累计 cost收到足够数量后一次性发射。 def __init__(self, window_size: int 10, parallelism: int 1): super().__init__(parallelismparallelism) self.window_size window_size self.buffers {} def _buffer_for(self, key: str) - list: if key not in self.buffers: self.buffers[key] [] return self.buffers[key] async def process(self, msg: StreamMessage): buf self._buffer_for(msg.key) buf.append(msg.data[cost]) if len(buf) self.window_size: total sum(buf) self.buffers[msg.key] [] await self.emit(StreamMessage( keymsg.key, data{user_id: msg.key, total_cost: total}, timestampmsg.timestamp, ))这里SumWindowUnit是有状态单元状态就是self.buffers。为了保证安全同一个 key 的消息必须始终进入同一个实例这个问题靠 Pipeline 的分区路由解决。Pipeline 的实现如下class Pipeline: def __init__(self, loop: asyncio.AbstractEventLoop): self.loop loop self.source_queue: Optional[asyncio.Queue] None self.units: list[BaseUnit] [] self.sink: Optional[BaseUnit] None def set_source(self, source: BaseUnit): self.source source self.source_queue asyncio.Queue(maxsize1024) def add_unit(self, unit: BaseUnit): self.units.append(unit) def set_sink(self, sink: BaseUnit): self.sink sink def wire_up(self): 把 source 到 sink 的队列全部连接起来。 prev_queue self.source_queue for unit in self.units: for _ in range(unit.parallelism): unit.bind(prev_queue, self.loop) prev_queue asyncio.Queue(maxsize1024) self.sink.bind(prev_queue, self.loop) async def start(self): self.wire_up() tasks [] # 启动每个单元的消费者任务每个并行实例一个任务 for unit in self.units: for i in range(unit.parallelism): tasks.append(self._run_unit(unit, i)) tasks.append(self._run_sink(self.sink)) tasks.append(self._run_source(self.source)) await asyncio.gather(*tasks) async def _run_unit(self, unit: BaseUnit, idx: int): await unit.init() while True: msg await unit._queue.get() try: await unit.process(msg) except Exception as exc: print(funit {type(unit).__name__} instance {idx} error: {exc}) finally: unit._queue.task_done()这个版本故意做了简化为了让你看清核心逻辑而不是被并发细节淹没。实际工程里_run_unit还需要处理结束信号比如收到None消息就退出循环以及把异常上报到统一监控而不是打个日志就完事。3.3 用 asyncio.Queue 实现背压的细节说明前面提到emit里用了await queue.put(msg)这个操作在队列满的时候会挂起当前协程直到下游消费出一个空位。这正是背压的核心机制。我实测过一个 50000 条数据的测试上游每 1 毫秒发射一条消息下游每 5 毫秒处理一条。如果不加有界队列上游会瞬间把所有数据都塞进下游缓冲区内存峰值飙到 300MB 以上加了maxsize1024的有界队列后内存峰值稳定在几 MB上游的执行速率会被自动拉到和下游差不多的水平。这个方案的代价是如果上游是外部数据源比如 Kafka那么背压信号最终要靠“暂停拉取”来实现asyncio.Queue 阻塞的不只是上游执行单元而是整个生产链路。所以在接入真实数据源时需要给 Source 也加上背压感知逻辑队列满时Source 暂时不调用await queue.put而是在队列有空位后再继续从外部读取。3.4 生命周期、错误处理与单元测试生命周期是执行单元框架里容易被忽略的部分。我见过不少人在单元里直接__init__里连数据库结果并行实例一多连接数直接打满。正确做法是资源密集型操作放在init里做并且让框架只在进程启动时调用一次而不是实例化时初始化。错误处理上也踩过坑。最开始的版本是某个实例处理数据抛异常整个任务崩掉所有数据都停了。后来改成实例内捕获异常记录错误消息的 key 和失败原因同时让框架层面的监控能看到每个单元的成功数和失败数。对于可重试的异常比如下游数据库超时我加了一个简单重试装饰器最多重试 3 次每次等待时间翻倍。单元测试这块我强烈建议不要等到整个 Pipeline 搭完再测。每个单元应该独立测喂几条构造数据断言输出。我用的是pytest-asyncio插件测试代码长这样import pytest pytest.mark.asyncio async def test_filter_unit_drops_ping(): unit FilterUnit() # 构造一个只有输出队列的测试环境 output asyncio.Queue() unit.bind(output, asyncio.get_event_loop()) await unit.init() await unit.process(StreamMessage(keyu1, data{ action: ping, user_id: u1})) assert output.empty() await unit.process(StreamMessage(keyu1, data{ action: click, user_id: u1})) assert not output.empty()这种测试写多了之后重构单元内部逻辑几乎没有心理负担因为回归成本很低。这也是我把执行单元拆解得很细的根本原因——每个单元越小越独立可测试性和可维护性就越好。4. 实战中的常见问题与排查实录4.1 背压失效内存还是涨上去了有界队列看起来解决了背压问题但有一次我跑长任务时内存依然缓慢上涨。排查发现问题出在一个执行单元里它处理每条消息时都会往self.buffers写入数据但清理逻辑写在process的某个分支里。上游一旦遇到异常数据处理流程提前 return清理逻辑无法执行buffer 就只增不减。这给我一个教训背压只解决队列层面的积压单元内部的状态积累才是更难发现的内存泄漏。排查手段很简单在监控面板里画出每个单元状态对象的大小变化如果某个单元的状态单调增长且没有回落基本就是状态清理时机不对。修复方法是把状态清理放到finally块里确保无论正常还是异常都会执行。4.2 状态被两个线程同时改坏有一回我在 FilterUnit 里加了并行度同时处理 200 个并发结果某个计数器偶尔多算或者少算。查了半天才发现过滤逻辑是无状态的但全局变量里藏着一个统计用的可变对象被所有并行实例共享。修复方案有两种要么把统计逻辑也拆成独立的有状态单元并设置好 key 分区要么用threading.local隔离每个实例的临时变量。我最终选择前者因为统计本身就是数据流的一部分不该藏在某个单元的私有角落。4.3 循环依赖导致整个 Pipeline 卡死数据流图理论上是有向无环的但实际业务中经常出现“需要把处理结果反馈回上游”的场景比如实时推荐系统的特征回灌。我曾试图直接把输出连接到输入端口结果队列互相等待两个单元谁也没法继续执行。处理这类问题通常有两种方式。一是引入缓存中间层反馈数据不直接写回上游队列而是写入一个外部存储上游通过定时轮询或订阅感知新数据。二是把环路打破让反馈数据走一条异步通道不占用主线队列资源。我建议优先选第一种因为外部存储能自然隔离反馈链路的故障不至于让主流程被拖垮。4.4 明明分区了下游却还是出现了乱序用hash(key) % parallelism做路由能保证同一个 key 的数据进入同一个实例。但我在实现时犯了个低级错误hash()函数在 Python 进程重启后随机化PYTHONHASHSEED导致同一 key 在重启后路由到不同实例。如果框架支持动态扩容或重启恢复这个随机化会把整个分区规则打乱。解决方法是改用稳定的哈希算法比如 MurmurHash 或 CRC32而不是内置hash()。尤其在做增量计算、累计状态时稳定路由是局部有序的前提这一点绝对马虎不得。几个常见问题整理成一张速查表方便以后排查症状根因定位方法解决方案内存持续上涨有界队列设置过大或状态清理缺失观察队列长度指标、状态对象大小调小队列 maxsize修复状态清理逻辑统计偶发不准共享可变状态被并发修改检查实例间是否有共享对象拆独立有状态单元强分区约束任务卡死无日志循环依赖导致队列互相等待画出执行图找环引入外部存储或异步反馈通道重启后结果对不上默认 hash 随机化改变路由打印路由结果对比改用稳定哈希算法上游消息丢失队列满时直接丢弃看是否有 try-put 失败分支改为 await put 阻塞式背压4.5 数据倾斜分区不均会拖慢整个链路分区路由还有一个隐藏问题数据倾斜。虽然 key 设计得不错但真实业务中某些 hot key比如一个大主播的直播间数据会产生远超平均值的数据量导致某个实例忙死其他实例空闲。应对倾斜有几个常见套路。一是局部聚合再全局聚合比如统计 UV先在每个实例内部做去重再把结果汇总到下游减少热节点的传输量。二是加盐把 hot key 拆成多个虚拟 key让数据分散到不同实例计算完成后再按真实 key 做合并。三是动态识别热 key单独路由当某个 key 流量超过阈值时把它拆分或分配给专门实例。我在日志清洗场景里遇到过一次某个用户疯狂触发日志单实例 CPU 50%其他实例只有 5%。最终采用了拆虚拟 key 的方案先按虚拟 key 做字段提取再在下游聚合时合并原 key。效果很明显整体吞吐翻了接近一倍。我个人在实际项目里的体会是执行单元的设计再精巧也扛不住对数据分布情况的一无所知。无论框架怎么封装、调度器怎么优化你始终要对自己数据的 key 分布、流量波动有数。很多跑线上才发现的问题如果在上线前用一个小时做统计分析和压测其实都能提前暴露。真等到任务上线之后与其在手忙脚乱中调并行度、改路由不如回到执行单元本身看看哪一步假设不成立往往才是根因所在。
我是一个平时喜欢折腾数据管线的工程师今天想好好聊聊数据流式编程里最基础也最关键的一个概念——执行单元。这个词听起来有点学院派但说白了它就是数据流里那个真正干活的节点收一条数据进来处理一下再把结果吐给下一个环节。不管你是接触过 Flink、Kafka Streams还是用过 Node-RED、RxJS甚至只是写过一条简单的异步 Promise 链背后都离不开执行单元的这套逻辑。这篇文章我会从设计原理讲到手写实现再聊一聊我在实际项目中踩过的坑。适合正在学习流式计算、想自己搭建轻量级数据处理框架、或者被异步并发搞得头疼的同学。概念我会讲得尽量通俗代码也给了可直接跑的版本保证不是那种看完就忘的科普。1. 数据流式编程的骨架执行单元到底解决了什么问题1.1 先从一个让我头疼的场景说起之前我维护过一个日志清洗服务逻辑不算复杂从消息队列拿原始日志解析成结构化字段过滤掉无用的调试信息做字段补全最后写到存储里。听起来很直接吧但用传统命令式写法没过多久代码就成了一团乱麻。核心问题不是逻辑本身难而是每一步之间耦合得太紧。解析方法要等拿到输入才执行过滤逻辑散落在各处如果某一环节需要并行处理还得手动管理线程和队列。后来我把这段逻辑用数据流的方式重写每个处理阶段拆成一个执行单元单元之间只通过消息传递数据。改动之后代码结构清晰了很多每个单元可以独立测试并且随时能在中间插入新功能比如加一个脱敏节点完全不用动上下游逻辑。这个经历其实反映了执行单元的核心价值它将一段连续的处理过程拆成了离散、可组合、可独立伸缩的步骤。每一步接收输入、处理、发射输出至于上游是谁、下游是谁单元本身不关心只要数据格式匹配就行。1.2 执行单元到底长什么样从抽象角度看一个执行单元通常具备这三样东西输入端口接收上游传来的数据。简单场景只有一个输入端口复杂场景可能有多个比如双流 join 就需要两个输入。输出端口把处理结果发给下游。大多数单元是单输出的但也存在条件分支单元按规则把数据路由到不同输出。生命周期方法负责初始化资源、处理单条数据、释放资源。我把它们统称为init、process、dispose。很多流处理框架里概念叫法不同但本质上是一致的Flink 管它叫算子OperatorAkka Streams 里叫 StageNode-RED 里叫 NodeJava Stream 里叫中间操作。换个马甲而已。有一个关键点值得强调执行单元的理想状态是无副作用的也就是它不修改外部共享变量只依赖输入和自身保存的状态。这样做的好处是单元可以随意移动、复制、并行扩展而不会产生奇怪的竞态问题。1.3 它和 Actor 模型、流水线、响应式编程的关系很多初学者搞不清数据流、Actor、响应式编程的区别我整理过一张熟悉的关系网概念侧重点和执行单元的关系Actor 模型并发、消息传递、状态隔离每个 Actor 可以看作一个带邮箱的执行单元单元间完全通过消息通信流水线阶段性加工、并行处理执行单元的有序组合就是一条数据流水线响应式编程异步数据流、声明式组合map/filter 等操作符就是现成的执行单元数据流编程图结构、数据驱动执行执行单元是图的节点边是数据通道这是最上层的抽象打个比方执行单元就像一条自动生产线上的工位。工人只管把自己工位上的活干完然后放到传送带上给下一个工位。传送带怎么调度、哪个工位多安排几个人、哪个工位需要缓冲库存这是框架层面要解决的问题。2. 执行单元的核心机制拆解2.1 输入输出契约消息与端口怎么定义才不容易踩坑执行单元之间的接口设计决定了整个数据流框架的灵活性和健壮性。我在这次实践里采用的是结构化消息而非裸对象每条消息包含三部分dataclass class StreamMessage: key: str # 分区键用于确定路由到哪个实例处理 data: Any # 实际负载 timestamp: float # 事件发生时间用于乱序处理和延迟统计为什么要带key和timestamp因为分布式场景下仅靠消息内容无法决定“这条数据该交给哪个执行单元实例”和“这条数据是不是来晚了”。很多初学者在设计接口时只留一个payload字段后面做并行分区和时间窗口时才发现缺信息被迫改接口。输入端口个数也有讲究。常规单元是单输入单输出map、filter 都是这种但多输入场景并不罕见。我在自己的框架里允许单元声明多个输入端口每个端口绑定一个上游输出内部用select多路复用接收数据。这样做能处理类似“订单流和支付流做关联”的需求不需要额外引入 join 算子。输出这块我建议至少给出emit和emit_many两个方法。前者发一条后者批量发。批量发在吞吐量要求高时很有用单条发射的方式调用开销太大。实测在 Python 环境中批量发射能减少大概 20%30% 的调度开销。2.2 状态管理执行单元里有状态怎么保证不出乱子执行单元分无状态和有状态两种。无状态单元很容易理解输入什么就输出什么比如格式转换、字段过滤。有状态单元就复杂了它需要在处理消息的同时维护内部数据例如滚动窗口计算、去重、计数。状态存放有三种常见策略纯内存状态状态存在对象字段里比如self.counter。实现最简单但一旦进程重启状态全丢只适合允许丢失的场景。外部存储状态把状态写到 Redis、数据库里。可靠性高但每次读写都有 IO 延迟性能会下降不少。本地持久化状态像 Flink 的 RocksDB 后端那样把状态保存在本地嵌入式数据库。可靠性和性能平衡得比较好但实现复杂。我在轻量框架里用的是纯内存状态但特意加了一个约束只有带相同 key 的数据才会被路由到同一个执行单元实例。这就把并发问题缩小到单实例内部避免两个线程同时修改同一个计数器的尴尬。这里给新手一个经验设计有状态执行单元时务必想清楚两个问题——数据按什么维度分组、状态生命周期从哪开始到哪结束。比如做一个“过去 5 分钟每个用户下单总金额”的计算器分组维度是用户 ID状态生命周期从第一次收到该用户消息开始到窗口结束释放。没有这两个答案代码怎么写都会别扭。2.3 调度模型推式、拉式和背压到底选哪种执行单元之间怎么传递数据是数据流框架最核心的决策之一。我遇到过三种模型推式Push上游处理完直接调用下游的process。实现简单响应快但问题是一旦下游处理速度跟不上数据会堆积在内存里最终 OOM。拉式Pull下游主动向上游要数据。下游消费多少上游才生产多少天然带背压。但实现起来要处理“上游没数据时下游等多久”的问题延迟偏高。动态推拉混合正常情况下上游推当下游缓冲队列达到阈值时触发反压信号通知上游停下来等一会儿。这是目前主流流处理引擎的通用方案。我的实践中背压不是理论问题而是真实血泪。最早版本用的是纯推式每条消息都立刻传给下游结果下游一个慢速的磁盘写入把内存直接打爆。后来我把连接单元之间的通道设计成有界队列当队列满时上游写入会被阻塞自然形成背压。这里用餐厅传菜做类比厨师做好菜放到出餐口传菜员取走去上菜。如果出餐口只有一个位置厨师就得放慢节奏等传菜员如果出餐口无限大菜就会堆满整个厨房。有界队列就是那个“只有一个位置”的出餐口用阻塞换取系统的稳定性。2.4 并行执行与数据分区为什么不能简单多开几个线程单个执行单元处理速度有上限于是自然想到并行让多个单元实例同时干活。但并行不是简单地把一个单元复制两份而是要处理数据该怎么分配的问题。我的做法是引入分区键就是消息里的key字段。框架内部维护多个单元实例按照hash(key) % 实例数把消息路由到不同实例。这样做的意义是同一条 key 的数据始终进入同一个实例局部有序性得到保证每个实例内部的状态也不会被别的实例破坏。刚开始并行时我犯过一个错误为了让某个无状态单元跑快点我把它复制成 4 个实例数据随机分配到各个实例。结果下游刚好有一个去重单元依赖上游“同一用户记录只经过一个实例”的假设导致同一个用户的两条记录被不同实例处理去重失效统计数字直接翻倍。修复方法就是在无状态单元上也保留分区路由保证相同 key 的消息走同一条路径。并行度也并非越大越好。每个实例都有调度和内存开销而且如果单元里访问了外部系统并行度过高还会压垮下游数据库。我一般遵循一个经验法则先看单个实例的吞吐瓶颈在 CPU 还是 IO瓶颈在 CPU 就按核数设置并行度瓶颈在 IO 就适当调高但给下游留 30% 余量。3. 实操手写一个轻量级执行单元框架3.1 框架整体设计思路这部分是全文的重头戏。目标不是把它做成生产级引擎而是让你看完后能理解流处理框架的核心环节是怎么串起来的。我选 Python 语言原因有三一是大家读起来没门槛二是 Python 的asyncio.Queue天然支持有界队列和背压正好和前面的调度模型对得上三是写单元测试方便。框架包含四个核心类StreamMessage消息载体包含 key、data、timestamp。BaseUnit执行单元基类声明生命周期钩子和输入输出逻辑。Pipeline管道类负责连接单元、调度执行、分配消息。Source特殊执行单元没有输入只有输出是数据流的起点。整体结构是Source 产生数据经过一系列 BaseUnit最终到达 Sink一个只收不发的特殊单元。Pipeline 内部维护一条有向无环的执行链数据从头流到尾。3.2 骨架代码执行单元和平管道的实现首先定义消息和执行单元基类import asyncio from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Optional dataclass class StreamMessage: 数据流中传递的消息体。key 用于分区路由data 是实际内容。 key: str data: Any timestamp: float 0.0 class BaseUnit(ABC): def __init__(self, parallelism: int 1): self.parallelism parallelism self._outputs: list[asyncio.Queue] [] self._event_loop: Optional[asyncio.AbstractEventLoop] None def bind(self, downstream_queue: asyncio.Queue, loop: asyncio.AbstractEventLoop): 把当前单元的输出绑定到一个下游队列。 self._outputs.append(downstream_queue) self._event_loop loop async def emit(self, msg: StreamMessage): 向下游发射一条消息。队列满时会自动阻塞形成背压。 for queue in self._outputs: await queue.put(msg) async def init(self): 初始化钩子比如建立数据库连接、加载模型。 pass abstractmethod async def process(self, msg: StreamMessage): 处理单条消息这是执行单元的核心。 pass async def dispose(self): 释放资源钩子。 pass这段代码里最关键的是emit方法。它不像传统做法那样直接调用下游的process而是把消息放进一个有界队列。如果下游处理慢队列空间不足await queue.put(msg)就会挂起整个链路自然减速而不会导致内存无限增长。接着是单元实现示例。我写一个字段提取、一个过滤、一个聚合窗口class ExtractUnit(BaseUnit): 从原始日志中解析出关键字段。 async def process(self, msg: StreamMessage): raw msg.data parsed { user_id: raw[user_id], action: raw[action], cost: float(raw.get(cost, 0)), ts: raw[timestamp], } await self.emit(StreamMessage(keymsg.key, dataparsed, timestampmsg.timestamp)) class FilterUnit(BaseUnit): 过滤掉 action 为 ping 的探活日志。 async def process(self, msg: StreamMessage): if msg.data[action] ! ping: await self.emit(msg) class SumWindowUnit(BaseUnit): 按 user_id 累计 cost收到足够数量后一次性发射。 def __init__(self, window_size: int 10, parallelism: int 1): super().__init__(parallelismparallelism) self.window_size window_size self.buffers {} def _buffer_for(self, key: str) - list: if key not in self.buffers: self.buffers[key] [] return self.buffers[key] async def process(self, msg: StreamMessage): buf self._buffer_for(msg.key) buf.append(msg.data[cost]) if len(buf) self.window_size: total sum(buf) self.buffers[msg.key] [] await self.emit(StreamMessage( keymsg.key, data{user_id: msg.key, total_cost: total}, timestampmsg.timestamp, ))这里SumWindowUnit是有状态单元状态就是self.buffers。为了保证安全同一个 key 的消息必须始终进入同一个实例这个问题靠 Pipeline 的分区路由解决。Pipeline 的实现如下class Pipeline: def __init__(self, loop: asyncio.AbstractEventLoop): self.loop loop self.source_queue: Optional[asyncio.Queue] None self.units: list[BaseUnit] [] self.sink: Optional[BaseUnit] None def set_source(self, source: BaseUnit): self.source source self.source_queue asyncio.Queue(maxsize1024) def add_unit(self, unit: BaseUnit): self.units.append(unit) def set_sink(self, sink: BaseUnit): self.sink sink def wire_up(self): 把 source 到 sink 的队列全部连接起来。 prev_queue self.source_queue for unit in self.units: for _ in range(unit.parallelism): unit.bind(prev_queue, self.loop) prev_queue asyncio.Queue(maxsize1024) self.sink.bind(prev_queue, self.loop) async def start(self): self.wire_up() tasks [] # 启动每个单元的消费者任务每个并行实例一个任务 for unit in self.units: for i in range(unit.parallelism): tasks.append(self._run_unit(unit, i)) tasks.append(self._run_sink(self.sink)) tasks.append(self._run_source(self.source)) await asyncio.gather(*tasks) async def _run_unit(self, unit: BaseUnit, idx: int): await unit.init() while True: msg await unit._queue.get() try: await unit.process(msg) except Exception as exc: print(funit {type(unit).__name__} instance {idx} error: {exc}) finally: unit._queue.task_done()这个版本故意做了简化为了让你看清核心逻辑而不是被并发细节淹没。实际工程里_run_unit还需要处理结束信号比如收到None消息就退出循环以及把异常上报到统一监控而不是打个日志就完事。3.3 用 asyncio.Queue 实现背压的细节说明前面提到emit里用了await queue.put(msg)这个操作在队列满的时候会挂起当前协程直到下游消费出一个空位。这正是背压的核心机制。我实测过一个 50000 条数据的测试上游每 1 毫秒发射一条消息下游每 5 毫秒处理一条。如果不加有界队列上游会瞬间把所有数据都塞进下游缓冲区内存峰值飙到 300MB 以上加了maxsize1024的有界队列后内存峰值稳定在几 MB上游的执行速率会被自动拉到和下游差不多的水平。这个方案的代价是如果上游是外部数据源比如 Kafka那么背压信号最终要靠“暂停拉取”来实现asyncio.Queue 阻塞的不只是上游执行单元而是整个生产链路。所以在接入真实数据源时需要给 Source 也加上背压感知逻辑队列满时Source 暂时不调用await queue.put而是在队列有空位后再继续从外部读取。3.4 生命周期、错误处理与单元测试生命周期是执行单元框架里容易被忽略的部分。我见过不少人在单元里直接__init__里连数据库结果并行实例一多连接数直接打满。正确做法是资源密集型操作放在init里做并且让框架只在进程启动时调用一次而不是实例化时初始化。错误处理上也踩过坑。最开始的版本是某个实例处理数据抛异常整个任务崩掉所有数据都停了。后来改成实例内捕获异常记录错误消息的 key 和失败原因同时让框架层面的监控能看到每个单元的成功数和失败数。对于可重试的异常比如下游数据库超时我加了一个简单重试装饰器最多重试 3 次每次等待时间翻倍。单元测试这块我强烈建议不要等到整个 Pipeline 搭完再测。每个单元应该独立测喂几条构造数据断言输出。我用的是pytest-asyncio插件测试代码长这样import pytest pytest.mark.asyncio async def test_filter_unit_drops_ping(): unit FilterUnit() # 构造一个只有输出队列的测试环境 output asyncio.Queue() unit.bind(output, asyncio.get_event_loop()) await unit.init() await unit.process(StreamMessage(keyu1, data{ action: ping, user_id: u1})) assert output.empty() await unit.process(StreamMessage(keyu1, data{ action: click, user_id: u1})) assert not output.empty()这种测试写多了之后重构单元内部逻辑几乎没有心理负担因为回归成本很低。这也是我把执行单元拆解得很细的根本原因——每个单元越小越独立可测试性和可维护性就越好。4. 实战中的常见问题与排查实录4.1 背压失效内存还是涨上去了有界队列看起来解决了背压问题但有一次我跑长任务时内存依然缓慢上涨。排查发现问题出在一个执行单元里它处理每条消息时都会往self.buffers写入数据但清理逻辑写在process的某个分支里。上游一旦遇到异常数据处理流程提前 return清理逻辑无法执行buffer 就只增不减。这给我一个教训背压只解决队列层面的积压单元内部的状态积累才是更难发现的内存泄漏。排查手段很简单在监控面板里画出每个单元状态对象的大小变化如果某个单元的状态单调增长且没有回落基本就是状态清理时机不对。修复方法是把状态清理放到finally块里确保无论正常还是异常都会执行。4.2 状态被两个线程同时改坏有一回我在 FilterUnit 里加了并行度同时处理 200 个并发结果某个计数器偶尔多算或者少算。查了半天才发现过滤逻辑是无状态的但全局变量里藏着一个统计用的可变对象被所有并行实例共享。修复方案有两种要么把统计逻辑也拆成独立的有状态单元并设置好 key 分区要么用threading.local隔离每个实例的临时变量。我最终选择前者因为统计本身就是数据流的一部分不该藏在某个单元的私有角落。4.3 循环依赖导致整个 Pipeline 卡死数据流图理论上是有向无环的但实际业务中经常出现“需要把处理结果反馈回上游”的场景比如实时推荐系统的特征回灌。我曾试图直接把输出连接到输入端口结果队列互相等待两个单元谁也没法继续执行。处理这类问题通常有两种方式。一是引入缓存中间层反馈数据不直接写回上游队列而是写入一个外部存储上游通过定时轮询或订阅感知新数据。二是把环路打破让反馈数据走一条异步通道不占用主线队列资源。我建议优先选第一种因为外部存储能自然隔离反馈链路的故障不至于让主流程被拖垮。4.4 明明分区了下游却还是出现了乱序用hash(key) % parallelism做路由能保证同一个 key 的数据进入同一个实例。但我在实现时犯了个低级错误hash()函数在 Python 进程重启后随机化PYTHONHASHSEED导致同一 key 在重启后路由到不同实例。如果框架支持动态扩容或重启恢复这个随机化会把整个分区规则打乱。解决方法是改用稳定的哈希算法比如 MurmurHash 或 CRC32而不是内置hash()。尤其在做增量计算、累计状态时稳定路由是局部有序的前提这一点绝对马虎不得。几个常见问题整理成一张速查表方便以后排查症状根因定位方法解决方案内存持续上涨有界队列设置过大或状态清理缺失观察队列长度指标、状态对象大小调小队列 maxsize修复状态清理逻辑统计偶发不准共享可变状态被并发修改检查实例间是否有共享对象拆独立有状态单元强分区约束任务卡死无日志循环依赖导致队列互相等待画出执行图找环引入外部存储或异步反馈通道重启后结果对不上默认 hash 随机化改变路由打印路由结果对比改用稳定哈希算法上游消息丢失队列满时直接丢弃看是否有 try-put 失败分支改为 await put 阻塞式背压4.5 数据倾斜分区不均会拖慢整个链路分区路由还有一个隐藏问题数据倾斜。虽然 key 设计得不错但真实业务中某些 hot key比如一个大主播的直播间数据会产生远超平均值的数据量导致某个实例忙死其他实例空闲。应对倾斜有几个常见套路。一是局部聚合再全局聚合比如统计 UV先在每个实例内部做去重再把结果汇总到下游减少热节点的传输量。二是加盐把 hot key 拆成多个虚拟 key让数据分散到不同实例计算完成后再按真实 key 做合并。三是动态识别热 key单独路由当某个 key 流量超过阈值时把它拆分或分配给专门实例。我在日志清洗场景里遇到过一次某个用户疯狂触发日志单实例 CPU 50%其他实例只有 5%。最终采用了拆虚拟 key 的方案先按虚拟 key 做字段提取再在下游聚合时合并原 key。效果很明显整体吞吐翻了接近一倍。我个人在实际项目里的体会是执行单元的设计再精巧也扛不住对数据分布情况的一无所知。无论框架怎么封装、调度器怎么优化你始终要对自己数据的 key 分布、流量波动有数。很多跑线上才发现的问题如果在上线前用一个小时做统计分析和压测其实都能提前暴露。真等到任务上线之后与其在手忙脚乱中调并行度、改路由不如回到执行单元本身看看哪一步假设不成立往往才是根因所在。
