自研轻量级事件编排引擎rea：解决微服务异步协作与资源感知路由

1. 项目概述rea到底是什么1.1 从一次服务间状态同步的痛点说起先说个背景。之前我在做某跨平台系统的时候遇到一个特别别扭的问题订单服务、库存服务、支付服务是三个独立部署的模块但它们之间必须实时同步状态。订单创建成功之后要通知库存扣减支付回调之后要通知订单更新还有一堆类似“用户操作后触发邮件告警”的异步流程。最早我图省事直接让订单服务在业务代码里同步调用库存服务的 HTTP 接口结果一次库存服务重启订单流程直接卡死线上告警炸了一晚上。后来改成轮询数据库每隔几秒扫一次未处理的状态字段看似稳定了但延迟高、代码丑而且每个服务都要写一套轮询逻辑维护成本直接起飞。正是在这个背景下我才决定动手写一个轻量级的事件编排工具代号就叫reaResource Event Assistant资源事件编排助手。它解决的问题一句话说清楚让不同服务之间通过事件驱动的方式异步协作同时还能感知每个服务节点的资源状态按状态做路由和调度。不是要取代那些重量级消息中间件而是给中小团队、单体应用拆分初期、以及个人项目提供一个“开箱即用、看得懂、改得动”的轻量方案。1.2 rea的核心能力与适用边界rea 不只是一个消息队列它在消息队列的基础上多做了两件事第一维护一份动态的资源注册表每个接入的服务节点都要上报自己的存活状态、负载情况和自定义标签第二把事件路由规则和资源状态绑定比如“事件类型是 order.created 且目标资源组处于健康状态时才投递”“如果目标服务当前不可用事件进入重试队列而不是直接丢失”。这套能力非常适合这几类场景微服务拆分初期服务数量不大10 到 30 个但已经需要用异步事件解耦业务逻辑。现有的消息中间件要么太重维护成本高要么太弱只有简单的发布订阅没有路由策略。团队需要快速看到事件流转情况出问题能定位到具体事件实例而不是对着黑盒日志猜。当然它也有明显的适用边界。rea 是纯 Python 实现、默认用 SQLite 做持久化单机吞吐量肯定比不上那些专职消息队列。如果你的业务已经到每秒几万条事件、需要海量分区和流处理能力那还是老老实实上专业组件rea 更适合控制在每秒几千条以内的中小规模场景。2. 整体设计与技术选型2.1 为什么没有直接选现成的消息中间件在动手之前我把主流方案都过了一遍也做了简单的压测对比这里直接放结论。方案部署成本接入成本单机吞吐量运维心智Kafka高依赖 KRaft/ZooKeeper磁盘规划麻烦中客户端较重概念多极高十万级高分区、副本、 offset 管理都要理解RabbitMQ中依赖 Erlang 运行时低AMQP 协议成熟高万级中交换器、绑定、队列策略需要学习Redis Stream低低高万级中持久化弱、内存要盯着rea自研极低单进程 SQLite极低HTTP/JSON 简单客户端库3500 条/秒实测单机低代码量小逻辑透明选型逻辑很直白我当时面对的问题核心不是高吞吐而是架构解耦 可观测性 轻量接入。Kafka 再牛我团队里没人愿意为了每天几万条状态消息去维护一套三节点集群。RabbitMQ 虽然好但部署一套 Erlang 环境、设计 exchange/binding 的规则对一个小团队来说仍然是额外的心智负担。Redis Stream 没选是因为它的持久化机制在异常断电场景下有丢消息风险而我这个场景里“订单已创建但库存没扣到”是绝对不允许发生的事故。所以最终决定自己写一个足够简单、足够透明的事件编排引擎。核心代码控制在三千行以内任何新人花一天时间就能读完整个项目源码。这就是 rea 的设计初衷——事情不大但要做透。2.2 核心架构拆解rea 由三个核心模块组成事件总线Event Bus负责接收生产者发布的事件、做本地持久化、根据路由规则投递给消费者维护每个事件的投递状态。资源注册中心Resource Registry收集各个服务节点的注册信息、心跳数据、资源健康状态。每个节点启动时调用注册接口上报身份信息之后每 30 秒续一次心跳。路由引擎Router将“事件类型 资源状态 自定义规则”三者结合决定事件到底该进入哪个队列、投递给哪些消费者节点。这三个模块在 rea 内部是互相独立的既可以用一个进程全部跑起来单机模式也可以拆开部署资源注册中心单独实例事件总线单独实例。我在第一版实现里先把它们放在同一个进程内通过 Python 的 threading queue 做内部通信优先保证逻辑正确后面再考虑拆进程。事件的数据流是这样的业务服务生产者调用 rea 的 HTTP 接口提交一个事件对象。rea 校验事件格式生成全局事件 ID写入 SQLite 日志表返回“接收成功”。后台调度器扫描日志表中状态为 pending 的事件交给路由引擎匹配规则。路由引擎从资源注册中心拉取当前可用的消费者节点列表将事件放入对应的投递队列。投递线程把事件发送给消费者消费者处理完成后调用 ack 接口确认。已经 ack 的事件不再投递投递多次失败的事件进入死信队列。2.3 几个关键的技术决策为什么用 SQLite 存事件日志因为在中小规模场景下SQLite 的可靠性足够好而且事务能力让我可以很方便地实现“先落盘、再投递”的语义。我们把事件写入和投递状态更新放在同一个事务里就能做到崩溃恢复之后不丢已确认消息。SQLite 加上 WAL 模式之后读写在多线程下也能接受。很多开发者看不起 SQLite但它在嵌入式场景下的稳定性真的被低估了。内存队列与持久化日志的分工。rea 内部有两层结构内存队列负责加速投递SQLite 表负责记录所有状态。事件先写日志再进内存队列消费者 ack 之后再更新日志状态。如果进程在投递过程中崩溃重启后就从 SQLite 里捞状态为 pending 的事件重新投递这就保证了at-least-once语义。为什么选择 Python因为 rea 的核心不是追求极高性能而是追求极低的理解门槛。Python 的代码结构清晰、类型标注完善做 HTTP 接口和线程管理都非常顺手。如果后续某个模块需要更高性能我会单独把资源注册中心用 Go 重写其他部分保持 Python 不动。3. 核心功能实现与实操过程3.1 环境准备与工程骨架rea 的运行环境非常简单Python 3.9依赖库为 FastAPI、uvicorn、requests 和标准库 json/hashlib/sqlite3。用 pip 安装完依赖之后整个项目按下面的目录结构组织rea/ ├── main.py # 服务入口启动后同时拉起总线、注册中心、路由引擎 ├── event_bus/ │ ├── transport.py # HTTP 收发与 ack 接口 │ ├── storage.py # SQLite 持久化逻辑 │ └── dispatcher.py # 投递线程池与重试调度 ├── registry/ │ ├── models.py # 资源节点、心跳、健康状态的数据模型 │ ├── api.py # 注册、心跳、查询接口 │ └── health.py # 资源状态评估逻辑 ├── router/ │ ├── rules.py # 路由规则解析与匹配 │ ├── matcher.py # 基于前缀树/索引的匹配器 │ └── policy.py # 重试、熔断、死信策略 └── config.yaml # 端口、DB路径、超时、重试参数集成方在各自服务内部调用 rea 的客户端 SDK其实就是一个 requests 封装做两件事启动时调用register_node()注册自身业务处理中使用publish_event()提交事件。3.2 实现事件总线核心存储事件日志表是 rea 所有可靠性的基石。建表 SQL 如下CREATE TABLE IF NOT EXISTS event_log ( event_id TEXT PRIMARY KEY, event_type TEXT NOT NULL, trace_id TEXT, resource_key TEXT, source TEXT, payload TEXT, occurred_at TEXT NOT NULL, status TEXT NOT NULL DEFAULT pending, retry_count INTEGER NOT NULL DEFAULT 0, next_retry_at TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_event_status ON event_log(status, next_retry_at); CREATE INDEX IF NOT EXISTS idx_event_type ON event_log(event_type);存储模块把“写入事件”和“更新状态”都包在事务里。发布事件时body 里的 payload 会被 JSON 序列化成字符串存入 SQLite事件 ID 是自定义算法生成的其实就是时间前缀 随机序列 业务 KEY 的哈希这样在分布式环境里也能保持全局唯一。# event_bus/storage.py 核心片段 def append_event(self, event: dict) - str: event_id generate_event_id(event[source], event[event_type]) with self._get_connection() as conn: conn.execute( INSERT INTO event_log( event_id, event_type, trace_id, resource_key, source, payload, occurred_at, status, retry_count, next_retry_at, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, pending, 0, ?, ?, ?) , (...) ) conn.commit() return event_id为什么把 retry_count 和 next_retry_at 放在主表而不是单独建一张重试表因为这样可以保证重试逻辑只通过一个 UPDATE 语句完成不需要跨表联查简单且不容易出错。我第一次设计时确实单独建了 retry_task 表后来发现状态同步复杂了很多主表更新了状态、重试表却没删掉导致一个事件被投递两次。合并成一张表后这种问题从根上消失了。3.3 实现资源注册中心与路由匹配资源注册中心的模型设计如下每个节点必须上报三个维度的信息。身份维度节点 ID全局唯一、服务名、实例 IP、端口。资源维度当前 CPU 负载、内存占用率、自定义标签如环境、机房、灰度组。状态维度上一次心跳时间、持续健康标记、熔断计数。路由规则我设计成 JSON 格式存在配置文件中支持热加载。规则从两个维度做匹配{ rules: [ { id: rule_order_stock, event_type: order.created, resource_key_pattern: order:${order_id}, targets: [ {service: stock-service, tag_selector: {env: prod}}, {service: notify-service, tag_selector: {env: prod}} ], when: resource_group_is_healthy(stock-group), retry_policy: {max_retry: 5, backoff_sec: [1, 5, 30, 120, 300]} } ] }这里有个非常关键的细节为什么不直接把目标服务的地址写死而是写“服务名 标签选择器”因为服务地址是会变的。特别是在云环境里实例重启之后 IP 换了、扩容之后节点多了写死地址会导致投递失败或漏投。rea 的做法是通过标签选择器在资源注册中心里“查询当前应该把事件交给谁”把“路由”和“服务发现”结合起来。路由匹配的实现逻辑先根据事件中的 resource_key 计算前缀索引快速过滤出可能命中的规则候选集再对每条规则做完整字段匹配。如果规则中的when条件涉及资源健康状态路由引擎会去资源注册中心实时查询对应资源组的健康度。投递线程在发送事件之前会再确认一次目标节点的心跳时间是否在有效窗口内避免事件发给一个已经掉线的节点。# router/matcher.py 核心思路 def match(self, event: dict) - list[RouteTarget]: candidates self._candidate_rules(event[event_type]) if not candidates: return [] results [] for rule in candidates: if not re.match(rule[resource_key_pattern], event.get(resource_key, )): continue if rule.get(when) and not self._evaluate_condition(rule[when]): continue targets self.registry.resolve_targets(rule[targets]) if targets: results.extend(targets) return results这里踩过一个坑最初我用正则表达式直接匹配 resource_key_pattern规则一多、事件量一上来CPU 消耗很高甚至有一条恶意构造的长字符串让正则回溯卡了几秒。后来我改成前缀树预筛候选规则只有候选集里的规则才走正则性能提升非常明显。3.4 重试、幂等与死信处理投递线程发送事件到消费者接口时消费者必须实现一个约定好的回调格式接收 POST 请求解析{event_id, trace_id, event_type, payload}业务处理完成后返回200 OK如果返回非 2XX 状态码或者超时rea 判定本次投递失败。重试策略支持配置指数退避序列。默认序列是第 1、2、3、4、5 次重试分别延迟 1、5、30、120、300 秒。重试次数超过max_retry之后事件状态改为dead写入死信表。另外死信表中的事件不能直接丢弃需要提供后台管理接口支持人工查看、重放、标记完成。实际线上运行中我发现死信事件里相当一部分是消费者代码 bug 导致的比如反序列化 payload 时报错人工修复之后重放比起重新触发整条业务链路要省事得多。幂等处理是 at-least-once 语义必须面对的另一个问题一个事件因为消费者 ack 之前进程崩溃重启后会重投一次。所以消费者在处理事件时一定要基于业务数据的唯一键做幂等判断。举个实际例子订单服务处理“扣减库存”事件业务表里加一个processed_event_id字段处理前先查这个字段是否已经存在存在就直接返回成功不存在就执行扣减并写入处理记录。这个约定在接入 rea 的文档里会做成强制要求因为 rea 只能保证“尽量不丢事件”不能保证“恰好一次”。4. 常见问题与排查技巧实录4.1 丢消息与重复投递问题先说说我遇到次数最多的丢消息场景。第一版 rea 是在消费者 ack 成功之后再去更新 SQLite 状态这逻辑听起来没问题但实际上有个隐患消费者成功处理完业务之后在调 ack 接口的路上网络超时了。ack 请求没有到达 rearea 会认为投递失败稍后重试。这时候消费者已经处理过业务了重试就会造成重复处理。这个不是丢消息是重复投递。真正丢消息的情况也遇到过生产者在调用 rea 发布接口后rea 还没完成 SQLite 持久化就返回了 200。一次服务重启那条还没落盘的事件就丢了。修复办法很简单发布接口必须保证“返回 200 之前事件已经写入 SQLite 并提交事务”不能因为追求响应速度而牺牲落盘。问题现象根因解决方案事件丢失发布接口提前返回、SQLite 未落盘写入完成后再返回 200开启 WAL 模式事件重复投递ack 超时 / 消费者崩溃后重投消费者基于业务唯一键做幂等ack 丢失但业务已处理网络抖动 / ack 接口超时幂等键 定期对账补偿重试风暴消费者接口慢导致超时大量事件进入重试调整投递超时时间消费者接口做快速失败排查这类问题我一般先执行这个 SQL 看整体状态分布SELECT status, count(*) FROM event_log GROUP BY status;如果 dead 状态特别多优先看是不是消费者接口报错如果 pending 数量持续高位但消费者处理速度正常很可能是重试队列的 next_retry_at 距离当前时间太久或者投递线程池满了一直阻塞。4.2 资源注册中心心跳设计中的两个教训资源注册中心最烦的问题是“假死”和“脑裂”。我的第一版心跳设计是每 30 秒上报一次超过 90 秒没上报就标记为离线。实际运行中遇到过一次很尴尬的情况消费者节点因为一次 GC 停顿当时节点上跑着另一个 Java 应用频繁 Full GC心跳线程被卡住90 秒内没有上报rea 直接把节点标记为离线。之后事件被路由到其他正常节点但那个“假死”节点其实业务服务还在运行等 GC 结束之后它恢复处理就出现了同一个业务被两个节点同时处理的局面。这个问题的根源是单一心跳信号无法区分“进程假死”和“网络分区”。后来我改成了“心跳 资源级健康检查”双通道机制节点除了心跳还要每 30 秒执行一次自检任务把 CPU、内存、连接池状态一起上报。只有心跳超时且自检接口无法响应时才判定为离线。另一个教训是“脑裂”问题。两个 rea 实例在分布式部署模式下可能同时认为自己是主节点。我在单主模式下用 SQLite 的事务做选主但如果网络抖动导致两个实例同时发起选主就可能出现两本账。后来我引入了一个简单的租约机制每个实例必须每 5 秒续租一次续租失败就立即降级为从节点不提供写服务。这个方案没有真正解决分布式一致性但在中小规模场景下足够安全。4.3 路由规则匹配性能优化路由规则模块上线初期我直接用正则匹配所有规则。规则数量到了 200 条之后发布接口的 P99 延迟从 15ms 飙升到了 300ms。排查发现时间都消耗正则匹配和规则加载上于是做了三件事把事件类型作为第一层索引规则加载时先按 event_type 分组只有当前事件类型对应的规则组才参与匹配。resource_key_pattern 支持前缀匹配的改用前缀树避免无谓的正则回溯。规则文件每次变更后重新加载但加载完成前继续使用旧版本规则保证热加载过程不丢事件。这几项优化做完之后200 条规则下的 P99 又回到了 20ms 左右效果很明显。如果你也遇到类似性能问题建议先做规则分桶再看匹配算法不要一开始就换语言、换存储。5. 实测性能与使用心得5.1 性能基线测试我专门跑过一轮基线压测机器配置是 8 核 16G 内存rea 单进程部署SQLite WAL 模式事件体量平均 2KB。测试结果如下指标数值发布接口吞吐3500 条事件/秒端到端投递延迟 P5065ms端到端投递延迟 P99180ms进程内存占用~180MB稳定后CPU 占用3 核左右峰值这组数据说明一个问题rea 单机就可以支撑一个中等规模的后端系统。如果你用 Go 重新实现核心存储层吞吐应该还能再涨不少但我没有继续优化因为在这个量级上瓶颈已经不再是 rea 本身而是下游消费者接口的处理能力。5.2 集成到业务时的三个实用建议第一个建议是给每个业务事件设计幂等键。幂等键的最佳实践是业务唯一ID 事件类型的组合比如订单号加order.created不要只放订单号否则一个订单的多个历史事件会被误判为同一个事件。第二个建议是建议业务系统里维护一张本地消息表。在 publishers 侧先写业务表和本地消息表同一个事务再异步调用 rea 发布。进程崩溃导致本地消息表里的事件没有发布成功时启动扫描补投。这个方式本质上是先落库、再发事件的两阶段思路能最大程度避免业务成功但事件丢失的问题。第三个建议是优雅停机一定要做。rea 服务收到 SIGTERM 信号后先停止接收新事件然后进入排空模式等待内存队列里的在途事件全部 ack 或超时再释放 SQLite 连接并退出。不做这一步就会出现启动后大量 pending 事件涌入重试队列给消费者造成瞬时冲击。5.3 代码与经验之外的一些思考现在回看 rea 的整个开发过程最让我意外的不是某个技术难点而是“一个足够简单的方案能解决多少问题”。很多项目一上来就选重型中间件运维成本、学习成本、故障排查成本全都被放大了。rea 这个项目的价值不在于代码写得多好而在于它以极低的门槛提供了一个可靠的异步事件框架同时让使用它的人能一眼看懂全部逻辑。我在后续几个月里又把它接入到几类不同场景包括一些数据处理流水线效果都比较稳定。如果你也在面对类似的微服务解耦问题我的建议是先别急着上重量级组件用一个小而美的工具跑通业务模型等量确实上来了再评估是否要平移到大集群。这个过程不会浪费因为你在 small scale 上积累的路由规则、幂等设计、故障恢复经验到了大集群照样用得上。当然 rea 不是万能药它不会自动帮你做好业务幂等不会替你做资源规划也不保证写磁盘失败时数据 100% 不丢。它的优势在于把“轻量级、透明、够用”这三件事做到了平衡。如果你能接受它的边界那它绝对值得一试。