说到“rea”可能很多人第一时间会想到阅读器或者某个编辑器的缩写但对我来说它是我折腾了大半年的一套轻量级事件驱动执行框架的名字。我把Resource、Event、Action三个词的首字母拼在一起于是就叫了rea。简单说它解决的是服务端异步任务编排的问题一个业务对象发生变化后自动触发后续一系列动作而不需要在业务代码里写出一长串互相嵌套的调用链。我遇到过很多次这种场景文件上传后要自动生成缩略图、更新缓存、写审计日志订单状态变更后要触发库存扣减、短信通知、对账记录。最直白的做法是在接口实现里一个个同步调用过去一开始系统小无所谓调用链一长代码就变成意大利面。新同学接手根本不敢动每次加一个联动环节都要顺着代码把整条链路捋一遍别提多痛苦。rea把这类联动抽象成了“资源-事件-动作”三个概念业务方只负责声明“发生了什么”具体谁响应、怎么响应、怎么处理失败统一交给调度内核处理。如果你也正在被异步任务耦合、回调嵌套、状态散落在这几个问题困扰或者只是想了解一种轻量的事件驱动设计思路这篇文章值得你花十分钟读一读代码可以照着抄。1. 为什么造一个叫“rea”的轮子需求背景与设计取舍1.1 问题从哪来同步调用链的失控与备选方案的尴尬我先说一个真实的背景。我之前维护某个内部系统里面有一个典型的“文件处理”链路接口收到文件后先存OSS对象存储、再生成缩略图、再把原图路径和缩略图路径写入内容表、还要给审核人员发一条待办通知。最初版本就是按顺序同步调用代码长这样def upload_file(file): path save_to_oss(file) thumb make_thumbnail(path) db.insert(file_id, path, thumb) notify_audit(file_id) return path单看四个步骤没什么问题但量一上来就露馅了缩略图生成如果慢整个请求就卡住通知服务偶发超时会导致上传接口直接报错后续每加一个新动作比如加一个病毒扫描这段顺序逻辑就要继续往下续一行。等代码量过了几千行任何一个动作出问题排查时都得把调用栈从头翻到尾。那时候也不是没有现成方案。市面上很多任务队列、编排引擎都很成熟但引入它们意味着要部署额外服务、维护一堆配置、处理吞吐与网络开销。对于这个体量中等、又不想依赖大规模基础设施的项目来说真的有点小题大做。我要的其实很简单进程内就能跑起来接口一层和任务处理层可以灵活解耦部署时不需要额外dependency。于是rea就在这种“现有方案要么太重、要么太糙”的情况下诞生了。1.2 REA的命名与核心思想一切围绕Resource、Event、Action为什么叫rea而不是“任务队列”或者“事件总线”因为我在设计的时候刻意不想把它做成一个“队列工具”而是想用三个概念把业务建模这件事框起来。Resource资源系统中可以被操作的主体。文件、订单、用户、任务单都算资源。Event事件资源发生变化的事实。比如文件已上传、订单已支付、用户已注册。Action动作对某个事件作出的响应也就是真正的业务处理逻辑。这三个概念组合起来业务流就变成了一个很清晰的表达某个资源发生了一个事件调度内核负责找到所有关心这个事件的动作并在合适的时机执行它们。跟传统的“任务链”或者“回调注册”相比rea的差别在于业务代码里不会出现“缩略图服务”或“通知服务”的强引用大家只围绕事件类型打交道。上游只需发布一个“file.uploaded”事件下游谁订阅了它、有几个订阅者上游完全不需要关心。这就是解耦最本质的地方。1.3 关键取舍不引入外部消息队列也能跑通你在网上看到很多事件驱动框架第一句话就是“需要安装Redis”或者“依赖Kafka”。rea最初的定位决定了我不能这么干。我希望它是一个库而不是一个服务所以默认的队列后端就是进程内一个线程安全的双端队列deque配合固定数量的worker线程消费。这个取舍带来的好处非常明显接入成本极低pip install 或者直接把一个 py 文件拷进项目就能用。调试方便不依赖外部服务本地复现问题没有环境门槛。对中小体量业务足够单机每秒处理几千个事件完全没有压力。但我也知道它的边界在哪里进程重启内存里的未消费事件会丢多个服务实例之间没法共享同一个队列。所以我在代码里专门留了一个queue_backend抽象层以后真要上消息中间件只需要替换队列实现业务层的Resource/Event/Action代码完全不用动。这是我在设计时做过的最关键的一个决策后面实践也证明了它值回票价。2. REA框架的核心抽象与运行原理2.1 Resource把一切可操作对象统一成资源先看Resource的设计。我不建议把Resource做成业务对象的完全替身比如直接把ORM模型塞进去那样会让框架层跟业务层强耦合。我更推荐的做法是在rea内部维护一个“资源注册表”只存放必要的业务元数据真正的业务数据仍然留在业务自己的系统里。class Resource: def __init__(self, rid: str, data: dict None): self.rid rid # 全局唯一资源ID self.data data or {} # 业务元数据小的关键字段可以放这 self.version 1 # 乐观锁版本号 self.created time.time() self.updated time.time()所有资源统一用字符串形式的rid来标识而不是直接传对象引用这个设计有两个目的。第一跨进程传输事件时资源ID是可序列化的不会出现“这里拿不到服务实例”的问题第二保持框架对业务对象状态的只读视角资源数据是快照不是唯一数据源避免框架被拖进各种复杂事务。实操里我通常把需要被多个Action共同读取的字段放进data比如上传场景里的文件大小、类型、存储路径。如果字段太多也没关系核心做法是只放“路由判断要用的字段”其他重数据让Action自己去业务服务里查。2.2 Event描述“已经发生了什么”而不是“接下来要做什么”Event的设计是最容易走偏的地方。很多初学者会把事件定义成“我希望系统去做什么”比如“createThumbnail”“sendNotification”。如果你这样定义事件框架马上就变成一个“命令中心”Action之间还是会互相耦合因为谁发起命令、谁听命令全都摆在明面上。rea推荐的事件命名方式在代码里我已经体现出来了file.uploaded、order.paid、user.registered。它们描述的是已经发生的事实而不是即将执行的指令。这样做的核心好处是动作执行失败时事件本身依然是真实的重试、补偿、甚至新增一个订阅者都不会改变历史的含义。dataclass class Event: type: str # 事件类型比如 file.uploaded resource_id: str # 关联的资源ID payload: dict # 附加信息 timestamp: float field(default_factorytime.time) trace_id: str field(default_factorylambda: uuid.uuid4().hex[:12])trace_id是调试的关键。在一个复杂链路里一个文件上传事件可能触发三四个Action每个Action又可能发送新事件这时候如果没有一个贯穿始终的trace_id你很难把整条因果链串起来。我为每个初始事件生成一个12位的十六进制trace_id后续派生事件都会继承这个ID。实际排查问题时靠它就能把日志一次性拉出来效率高很多。2.3 Action动作注册、条件匹配与执行隔离Action是真正干活的代码。每个Action声明自己关心哪些事件类型然后提供一个handler回调函数。内核的事件分发器在拿到一个事件后只会把它投递给事件类型匹配的Action。class Action: def __init__(self, name: str, handler, event_typesNone, conditionNone): self.name name self.handler handler self.event_types set(event_types or []) self.condition conditioncondition是可选的条件函数用于做更细粒度的过滤。举个例子我只想对大文件生成缩略图小于1MB的文件直接跳过。那就可以写一个condition函数内部通过kernel.get_resource(rid)拿到资源快照判断它的size字段返回True/False。条件放在Action侧而不是Event侧是因为同一个事件可能有多个关注者每个关注者根据自己的业务规则决定要不要响应这才叫解耦。Action执行时还有一个隐藏点必须做执行隔离。一个Action抛出异常不能影响其他Action的执行。保证这个原则之后事件分发才具备基本的鲁棒性——某个下游动作挂了上游业务不会被拖垮日志里也不会出现混乱的trace。3. 从零实现一个最小可用的rea执行内核3.1 核心数据结构与初始化这一节我直接给出一个可以跑起来的最小实现。完整的rea内核只有三个核心部分资源表、Action注册表、事件队列。初始化代码我用了线程池worker数量通常取CPU核心数即可I/O密集型的Action可以适当调大。import threading import time import uuid import logging from collections import deque from dataclasses import dataclass, field logger logging.getLogger(rea) class ReaKernel: def __init__(self, worker_num4, queue_backendNone): self.resources {} self.actions {} self.queue queue_backend if queue_backend is not None else deque() self.running False self.worker_num worker_num self.workers [] self._start_workers()注意queue_backend参数的存在。默认传None就是用进程内deque将来换成某个外部队列的实现类类方法签名保持一致即可。这个抽象从第一版就留着了也算我给自己的一个提醒架构设计的前瞻性不在“现在就要支持分布式”而是“别把将来支持分布式的路堵死”。3.2 事件分发主循环每个worker线程都运行一个_loop方法不断从队列里取事件然后调用_dispatch。这里有个小细节当队列为空时我用sleep(0.002)避免CPU空转。很多初次实现线程池的人会忽略这个结果一个空跑循环就把CPU占满了这个问题排查起来很隐蔽。def _loop(self): while self.running: try: event self.queue.popleft() except IndexError: time.sleep(0.002) continue try: self._dispatch(event) except Exception: logger.exception(dispatch failed, trace_id%s, event.trace_id)_dispatch的逻辑很简单遍历该事件类型下注册的Action执行条件过滤然后调用handler。如果handler抛异常我只记日志不影响当前事件的后续Action遍历。def _dispatch(self, event): actions self.actions.get(event.type, []) for action in actions: try: if action.condition and not action.condition(event): continue action.handler(event, self) except Exception: logger.exception( action %s failed on event %s, trace_id%s, action.name, event.type, event.trace_id )3.3 Action注册与回调执行资源注册、事件发布和Action注册是面向使用者的三个核心接口。def register_resource(self, resource): self.resources[resource.rid] resource return resource def get_resource(self, rid): return self.resources.get(rid) def watch(self, action): for et in action.event_types: self.actions.setdefault(et, []).append(action) def emit(self, event_type, resource_id, **payload): event Event(typeevent_type, resource_idresource_id, payloadpayload) self.queue.append(event) return eventwatch是给框架用的业务方一般在初始化阶段把Action批量注册进去。emit是给业务方用的接口层只需要拼一个事件类型和资源ID后面的事情跟它无关。到这里整个rea执行内核大概只有不到80行代码。这也在一定程度上说明了它的设计哲学核心机制不复杂复杂的是业务规则而业务规则就应该放在Action里而不是框架里。3.4 跑通一个完整的调用示例我用“文件上传后自动生成缩略图并通知审核队列”来演示完整用法。kernel ReaKernel(worker_num2) # 第一步注册资源 image Resource(file_1001, data{ path: /tmp/uploads/a.jpg, size: 2048576, type: image/jpeg, }) kernel.register_resource(image) # 第二步定义Action def make_thumbnail(event, kernel): res kernel.get_resource(event.resource_id) if not res: logger.warning(resource not found: %s, event.resource_id) return thumb_path /tmp/thumbs/thumb_ res.rid .webp res.data[thumb_path] thumb_path print([action] 生成缩略图:, thumb_path) def notify_audit(event, kernel): print([action] 通知审核队列:, event.resource_id) # 第三步注册Action kernel.watch(Action(make_thumbnail, make_thumbnail, event_types[file.uploaded])) kernel.watch(Action(notify_audit, notify_audit, event_types[file.uploaded])) # 第四步上传接口发布事件 kernel.emit(file.uploaded, file_1001, sourceapi)运行后你会看到两个Action的输出它们的执行顺序是由worker线程调度决定的不是代码注册顺序。如果想保证顺序可以在watch的时候给Action加一个priority字段然后在_dispatch里按优先级排序。不过我一般情况下不建议依赖Action的执行顺序顺序要求往往意味着这两个Action之间应该有显式的依赖关系而不是简单并行。4. 真实项目中的扩展与落地经验4.1 用rea组装一条业务流水线前面的示例只是单个事件的简单响应。真实业务往往是一条流水线事件A触发Action1Action1执行后产生新事件B新事件B再触发Action2。举个电商场景的例子def on_order_paid(event, kernel): # 扣减库存 order kernel.get_resource(event.resource_id) inventory_service.deduct(order.data[sku_id], order.data[quantity]) # 发布新事件库存已扣减 kernel.emit(inventory.deducted, event.resource_id) def on_inventory_deducted(event, kernel): # 生成发货单 shipment_service.create(event.resource_id) kernel.emit(shipment.created, event.resource_id) def on_shipment_created(event, kernel): # 短信通知用户 sms_service.send(event.resource_id)这样做之后你再看业务代码的阅读体验和之前那一长串同步调用完全不同每一条流水线都由“事件类型”这个隐式契约串联起来。想新增一个环节只需要再新建一个Action订阅对应的事件类型不需要改动任何现有代码。我记得第一次在同事评审里展示这个效果时对方第一反应是以后加需求终于不用从头review整个调用链了。不过也提醒一句事件链不要贪深建议不超过四层。一旦超过四层排查成本又开始上升。如果业务链路真的那么长说明你该考虑把其中一部分抽成独立的领域服务而不是用事件把所有事情串在内存里。4.2 异常处理与重试策略Action执行失败怎么处理这是所有使用这类框架的人最关心的问题。我的做法分三层第一层Action内部自己try/except并处理掉可预期的业务异常比如数据不存在、参数校验失败。这类异常是业务常态不应该让框架重试。第二层框架捕获未预期异常把事件丢进一个“死信队列”同时记录失败信息。死信队列我用的是另一个deque只是用于观察不影响主流程。第三层针对“瞬时故障”比如网络超时、数据库连接闪断在Action内部做指数退避重试。我在代码里封装了一个简单装饰器def retry_on_failure(max_retries3, base_delay0.5): def decorator(handler): def wrapper(event, kernel): for attempt in range(1, max_retries 1): try: return handler(event, kernel) except RetryableError: if attempt max_retries: raise time.sleep(base_delay * (2 ** (attempt - 1))) return wrapper return decorator另一个非常重要的设计是幂等。事件驱动框架下同一个事件很可能因为重试被执行两次比如网络在响应后断了客户端收不到结果服务端重试就会导致重复操作。要避免这个问题最有效的办法是让Action处理本身具备幂等性用资源ID加上事件类型作为唯一键在处理前先检查“是否已经处理过”。比如在业务表里加一个processed_events表记录每个trace_id是否已经消费处理前查表处理成功后写表。4.3 日志、监控与可观测性事件驱动架构最大的短板是“调用链不直观”因为不再有显式的函数调用栈。所以日志和监控必须从一开始就设计进去。我做的第一件事是统一日志格式所有Action的日志都带trace_id和resource_id。这样通过一个trace_id就能在一个日志文件里搜出整条链路的全部处理记录体验跟上单线程调用差不多。第二件事是给每个Action的执行耗时加统计。最简单的办法是写一个带计时功能的基类或者装饰器import time def timed_action(name): def decorator(handler): def wrapper(event, kernel): start time.monotonic() try: return handler(event, kernel) finally: cost_ms (time.monotonic() - start) * 1000 logger.info([metric] action%s trace_id%s cost_ms%.2f, name, event.trace_id, cost_ms) return wrapper return decorator有了这些数据以后我就能回答几个核心问题哪个Action最慢哪个事件类型的消费量最大失败率最高的是哪个环节如果发现队列积压直接看每个Action的平均耗时和消费速度基本能定位瓶颈在何处。5. 常见问题与排查技巧实录5.1 事件为什么“丢了”你会发现只要用进程内队列100%会遇到一次事件神秘消失的场景。查来查去原因通常有几个进程在事件入队后、Action执行前重启内存队列直接清空。某个Action异常但被捕获了你只看到日志没看到业务结果误以为事件没触发。worker线程被非daemon线程阻塞或者线程池满了。第一个问题最凶。解决思路是如果是分发到“外部持久系统”的动作必须在事件发布时同步写一张“待处理事件表”再由worker消费后把状态改成“已处理”。进程重启后扫描这张表做补偿。这个方案等于给内存队列加了一层持久化保险虽然增加了一点开销但对于关键链路是必要的。第二个问题的排查方法是让框架在debug模式下输出所有被分发的事件和Action名称对照日志看某个Action是否有输出。我用过很多次这个办法基本一抓一个准。5.2 并发下资源被改乱多个worker线程同时处理同一个资源相关的事件时并发写Resource.data就可能出现竞争。最常见的结果是缩略图路径被覆盖、某个计数被减两次。我采用的方案是给资源加一个简单版本号字段。任何Action要做“读-改-写”操作时先读取resource.version写完资源数据后对版本号加一。如果提交更新时发现版本号已经被改就重新读取再执行。这其实就是乐观锁。再进一步如果某个资源真的需要严格串行处理比如同一个订单的支付和取消事件我通常会把这类事件的消费扔到一个“单worker队列”里让同一个rid的事件始终被同一个线程处理。方法也很简单在emit之前对resource_id哈希取模选择固定的worker线程。5.3 队列越积越长性能怎么定位rea跑一段时间后如果发现队列长度持续增长说明消费速度赶不上生产速度。我的排查顺序是先看每个Action的平均耗时找最慢的那一个。再看是否有Action在使用time.sleep做同步重试这会严重拖慢worker。最后看worker数量是否足够如果Action大多是I/O密集型可以适当加线程数。记得有次队列积压问题定位下来是一个Action里调用了外部接口平均耗时1.5秒而worker只有2个。当上传高峰来临时每个事件光排队就要几十秒。后来我把这个外部调用改成异步触发同时将worker提高到8个积压很快就消下去了。5.4 调试rea事件链路的三个实用技巧第一把队列切换成同步模式。我在ReaKernel里加过一个sync_modeTrue的开关开启后emit会直接调用_dispatch不走队列。这在单步调试时非常实用你可以像调试普通函数一样观察到每个Action的输入输出。第二利用trace_id全链路追踪。把一个事件从入口日志到出口日志全部搜出来按照时间线排列基本能看清整个链路的分支和耗时分布。第三临时加“事件类型过滤”。我经常在测试环境监听某一个特定事件类型打印所有Payload字段用来确认上游到底发了什么数据。这个办法在对接新业务方时特别有用不用猜字段名直接看数据。5.5 组合Action怎么保证“按顺序”执行有人可能会问我用rea的时候确实有两个Action必须A先执行、B后执行怎么办其实最干净的办法是不要直接依赖执行顺序而是让A执行完以后明确发出一个新事件B订阅新事件。这个新事件就是状态转换的显式宣告。比如文件上传后我希望先做病毒扫描再生成缩略图那就定义两个事件file.uploaded负责触发扫描扫描成功后发出file.scanned事件缩略图Action只订阅file.scanned。这样即使将来扫描环节下线或者换实现缩略图动作也不知道它只认“扫描通过”这个事实。这个模式推广开来你会发现整个系统里的事件类型就相当于一份“业务状态机说明书”每一类事件都代表系统的一次确定性状态变化。团队协作时还起到了文档的作用比看代码调用关系直观得多。结尾写在内核之外的话代码写到这一些读者可能觉得这不过是一个简化版的事件分发器谈不上多少技术含量。我反而觉得像rea这种小框架真正值钱的不是那几十行调度循环而是它逼着我把业务重新整理了一遍Resource、Event、Action三个概念一约束以前含糊不清的调用关系全都显式化了。有一次我在实际开发里想临时加一个“上传后立即清理临时目录”的步骤只写了十分钟代码、注册一个Action然后就下班了。放在以前我得在那条同步调用链的合适位置小心翼翼插入一行还要担心是不是会影响后面几行代码。如果你手头的项目也有这种“联动逻辑越来越多、改了怕炸”的苗头我建议你照着上面的思路先用几十行代码搭一个最小内核跑起来然后把你最痛的一条链路迁过来试试。你会发现事件驱动真正的价值不在性能而在它帮你把耦合的线一根根解开。最后再说一个小技巧给Action命名的时候尽量用“动词业务对象”的格式比如check_virus、send_sms日志里一眼就能看懂它在干什么排查问题能省很多力气。