4、事件驱动架构核心组件：事件生产者、事件路由器、事件总线、事件存储、事件消费者

好，咱们今天来聊聊事件驱动架构里的几个核心零件。说白了，这套架构就像一条流水线，每个环节都有自己的职责。我这些年做 Serverless 项目，踩过不少坑，也总结了一些经验。今天就把这五个组件掰开揉碎了讲给你听。

4.1 事件生产者：谁在“搞事情”？

事件生产者，就是那个发出事件的角色。它可以是任何东西——一个用户点击按钮、一台 IoT 设备上报数据、一个定时器触发，甚至另一个系统发来的消息。

我个人习惯把生产者分成两类：

  • 主动型生产者：比如用户注册、下单，这些是人主动触发的。
  • 被动型生产者：比如数据库变更捕获（CDC）、日志采集，这些是系统自动产生的。

我在项目中遇到过一个问题：某个团队把所有业务逻辑都塞进了生产者里，结果生产者变得又重又慢。你想想看，生产者本该只管“发出事件”，结果它还要去查数据库、做校验、甚至调用外部 API。这其实是个坏味道。

我的建议：生产者只负责“发生了什么”，不负责“怎么处理”。事件里带上必要的上下文就够了，别把整个业务对象都塞进去。

4.2 事件路由器：把消息送到该去的地方

事件路由器，顾名思义，就是决定“这个事件该发给谁”。它不像消息队列那样做存储，它只做转发。

常见的路由策略有几种：

路由策略 说明 适用场景
基于内容的路由 根据事件内容中的字段决定目标 订单类型不同，走不同处理流程
基于主题的路由 根据事件的主题（Topic）分发 日志分类、监控告警
基于标签的路由 给事件打标签，消费者按标签订阅 多租户场景、灰度发布

嗯，这里要注意：路由器的设计一定要轻量。我曾经见过一个项目，路由器里写了上百行 if-else，每次加一个新事件类型都要改路由器代码。这其实违背了事件驱动架构的初衷——松耦合。

避坑指南：我曾经因为路由器配置写得太死，导致新业务上线时不得不重启整个路由服务。后来我改用基于规则的动态路由，才彻底解决了这个问题。

4.3 事件总线：系统的“高速公路”

事件总线，你可以把它理解成一条高速公路。所有的事件都在上面跑，消费者在路边等着接货。

在 Serverless 架构里，事件总线的作用尤其重要。它不仅仅是传输，还承担了这些职责：

  • 解耦：生产者和消费者互不知道对方的存在
  • 缓冲：当消费者处理不过来时，事件可以在总线上排队
  • 过滤：只把符合条件的事件发给对应的消费者

我常用的几个事件总线产品：AWS EventBridge、阿里云 EventBridge、以及开源的 Apache Kafka（虽然它更偏消息队列，但也能当总线用）。

举个例子，在 AWS 上配置一个简单的事件总线规则：

{
  "EventBusName": "my-custom-bus",
  "EventPattern": {
    "source": ["myapp.orders"],
    "detail-type": ["OrderCreated"],
    "detail": {
      "status": ["pending"]
    }
  },
  "Targets": [
    {
      "Arn": "arn:aws:lambda:us-east-1:123456789:function:processOrder",
      "Id": "order-processor"
    }
  ]
}

这段配置的意思是：当总线上出现来源是 myapp.orders、类型是 OrderCreated、且状态是 pending 的事件时，就自动触发一个 Lambda 函数去处理。

关键点：事件总线是事件驱动架构的“中枢神经”。选型时一定要考虑吞吐量、延迟、以及是否支持事件重放。

4.4 事件存储：给事件“留个底”

事件存储，说白了就是把事件存下来。为什么要存？因为消费者可能不是实时处理的，或者我们需要回溯历史事件。

我见过两种常见的事件存储方式：

  1. 消息队列的持久化：比如 Kafka 的日志文件，事件消费完还可以保留一段时间
  2. 专门的事件存储库：比如 EventStoreDB、或者用 DynamoDB 自己实现一个

我个人更推荐第二种，尤其是做事件溯源（Event Sourcing）的时候。事件溯源的核心思想就是：不存当前状态，只存所有发生过的事件。想恢复状态？把事件从头到尾重放一遍就行。

举个例子，一个订单的状态机：

// 事件存储中的记录
[
  { "eventType": "OrderCreated", "timestamp": 1700000000, "data": { "orderId": "123", "amount": 100 } },
  { "eventType": "OrderPaid",    "timestamp": 1700000100, "data": { "orderId": "123", "paymentId": "p456" } },
  { "eventType": "OrderShipped", "timestamp": 1700000200, "data": { "orderId": "123", "trackingId": "t789" } }
]

你看，只要这些事件还在，任何时候我都能知道这个订单经历了什么。这在审计、调试、甚至数据恢复时特别有用。

注意：事件存储会占用大量空间。我曾经因为没设过期策略，导致存储成本飙升。建议根据业务需求设置合理的保留周期，比如 7 天、30 天或永久保留。

4.5 事件消费者：最终干活的“人”

事件消费者，就是那个真正处理事件的组件。它可以是 Lambda 函数、微服务、甚至是另一个事件总线。

消费者的设计有几个要点：

  • 幂等性：同一个事件被消费两次，结果应该一样。这是最基本的要求。
  • 异步处理：消费者不应该阻塞事件总线的流转。处理慢？那就加个队列缓冲一下。
  • 错误处理：消费失败怎么办？重试？死信队列？还是直接忽略？

我在项目中遇到过一个问题：某个消费者处理事件时调用了外部 API，结果 API 超时了，消费者一直重试，最后把事件总线堵死了。后来我加了一个重试次数限制和死信队列，才解决了这个问题。

一个典型的消费者代码（Node.js + Lambda）：

exports.handler = async (event) => {
  for (const record of event.Records) {
    try {
      const order = JSON.parse(record.body);
      // 处理订单
      await processOrder(order);
    } catch (error) {
      console.error('处理失败:', error);
      // 如果重试次数超过 3 次，就丢到死信队列
      if (record.attributes.ApproximateReceiveCount > 3) {
        await sendToDLQ(record);
      } else {
        throw error; // 让 Lambda 自动重试
      }
    }
  }
};
小技巧：消费者最好设计成“无状态”的。这样即使某个消费者挂了，另一个消费者可以立刻顶上，不会丢失事件。

4.6 五个组件如何协同工作？

好了，五个组件都讲完了。它们是怎么配合的呢？我画个简单的流程：

  1. 用户下单 → 事件生产者 发出 OrderCreated 事件
  2. 事件进入 事件总线，事件路由器 根据规则判断该发给谁
  3. 事件被写入 事件存储，留个底
  4. 事件消费者 从总线拉取事件，开始处理
  5. 处理完成后，消费者可能又变成生产者，发出下一个事件（比如 OrderPaid）

你看，这就是一个完整的事件驱动链路。每个组件各司其职，互不干扰。这也是为什么我特别喜欢这种架构——它让系统变得特别“干净”。

总结一下：事件驱动架构的核心，就是把这五个组件设计好、搭配好。生产者只管发，路由器只管转，总线只管传，存储只管存，消费者只管处理。谁也别越界，谁也别偷懒。

下一章，我会带你看看这些组件在 Serverless 环境下的具体落地实践。到时候咱们聊聊 AWS Lambda + EventBridge 的黄金组合，以及一些我踩过的坑。嗯，敬请期待。

← 上一章 📖 返回目录 下一章 →