Skip to content

39 Serverless：如何基于MQ和Serverless设计事件驱动架构？

你好，我是文强。

上节课我们讲了如何基于 Serverless 架构实现流式数据处理。这节课我们来看一下如何基于消息队列和 Serverless 来设计实现事件驱动架构（EDA）。

想必你对事件驱动架构这个词会有点陌生，为了让你更好地理解它，我们先来回顾两个知识点。

  1. 在 第08讲 讲消费者的时候，我们讲过消费数据时 Push 模型的实时性是最高的。
  2. 在 第38讲 讲到 Serverless Function 的数据源事件时，事件源数据的触发方式有一种是事件触发。

从技术上来看，这两个知识点都是事件驱动架构的一种实现。接下来我们就详细了解一下什么是事件驱动架构，它有什么用，以及它的架构和底层运行原理。

什么是事件驱动架构

事件驱动架构，英文是 Event Driven Architecture，简称 EDA。我们通过下面这张图来认识一下它。

如上图所示，这是从事件驱动架构抽象出来的架构图，图中包含了 事件源、 事件处理平台（事件总线）、 事件目标 三个部分。

  1. 事件源就是数据源，一个事件可以理解为一个数据。比如业务定义了一个事件，然后投递到了事件处理中心，本质上就是投递了一个数据。
  2. 事件处理平台（在公有云产品化后一般叫做事件总线）负责接收和持久化存储不同事件源的事件，然后根据设置好的事件规则触发执行不同的业务逻辑。
  3. 事件目标包含事件目标和目标调用两部分。事件目标一般是实体，比如HTTP/TCP Server、某个存储引擎等等。目标调用是一个动作，比如 HTTP API 调用、调用引擎客户端 SDK 写入数据等等。

所以说，事件驱动架构是指 主动拉取或被动接收上游不同事件源的数据，然后根据配置好的事件规则，触发执行不同的业务逻辑。

此时就有一个问题：既然事件是数据，那这个数据是什么格式的呢？从本质上来看，数据分为格式和内容两部分。比如下面是一个 JSON 格式的事件数据，该数据包含事件类型和事件值两个内容。

{
  "EventType"："UploadImage"，
  "EventValue":"{}"
}

从技术上来看，格式和内容是由事件处理平台规定的。此时就会出现不同事件处理平台设计的格式和内容都不一样，从而会导致下面两个问题：

  1. 从用户的角度看，不同事件平台事件定义不一样，当用户需要切换平台时就需要修改代码以匹配新的事件平台。因此就会造成用户和平台发生绑定，无法低成本地切换到其他事件平台。
  2. 从平台角度来看，自定义规定格式和内容会增加重复开发和设计的成本，长期来看也不利于行业的发展。

因此为了解决事件定义和描述的规范化，CNCF Serverless 工作组提出了 CloudEvents 的概念，用来制定统一的事件标准，比如请求方式、内容格式、内容组成等等。接下来我们来看看什么是 CloudEvents。

CNCF Serverless 工作组由 CNCF 的技术监管委员会成立，用于研究 Serverless 相关技术，并为 CNCF 推荐相关领域的未来发展计划。

什么是 CloudEvents

CloudEvents 是一个开源项目，它的目的是 定义通用的、标准化的事件格式。 用来简化事件驱动架构中的事件发布、订阅和处理流程。它规范定义了事件的结构和元数据，使得不同的事件源、中间件和消费者之间可以更容易地进行相互操作。

它的主要目标是：

  1. 提供一种通用的事件格式，以便在不同的云服务和应用程序之间传输事件。
  2. 使事件驱动的系统更加可扩展、可维护和可重用。
  3. 降低事件生产者和消费者之间的耦合度，简化事件处理流程。

如下图所示，CloudEvents 规范包括事件上下文、事件数据、传输协议三个部分。

  1. 事件上下文：包含事件的元数据，如事件类型、事件源、事件ID等。
  2. 事件数据：包含事件的有效负载，即与事件相关的具体信息。
  3. 传输协议：定义了如何在不同的系统和服务之间传输事件，例如 HTTP、MQTT、AMQP 等。

从整体来看，如果事件处理平台适配了 CloudEvents 标准，用户就可以更容易地构建事件驱动的应用程序，同时确保它们在不同的云平台和服务之间具有良好的互操作性。如果需要了解更多技术的细节，可以去看一下 官方文档 和 官方 Github 项目。

接下来我们通过几个事件驱动架构中的典型应用场景，来认识一下它在业务中的价值，也让你对事件驱动架构有一个更深的理解。

业务中的典型应用场景

这里我们主要讲自动化运维、应用连接和集成、商品订单中台三个比较典型的场景。

自动化运维

事件总线的一个典型场景就是自动化运维。

如下图所示，在业务架构中，系统异常时肯定会有一些指标异常。正常处理逻辑是：收集基础指标并上报后，根据一定的规则对源数据进行过滤或聚合，然后触发告警推送后续的自动化处理流程。

而为了达到这个效果，我们就需要进行一些代码开发和系统对接的工作。如果引入事件总线，就可以省去这部分的工作量。

如上图所示，业务采集异常指标后，可以通过事件处理平台开放的 API（比如 CloudEvent API ）进行上报。事件处理平台接收、存储事件数据，将这些数据聚合、过滤、处理，然后根据预先设置的规则，完成消息的推送与自动化处理。

从使用者的角度上看，满足这个需求我们只有“根据事件平台定义的规范上报数据”和“配置数据的处理规则”两个工作，开发实现的成本会降低很多。

接下来我们来看看应用连接和集成的场景。

应用连接和集成

我们在工作中会使用邮件进行交流，那么在邮件安全审核场景，当收到某些特殊的邮件后，就需要触发下游的安全审核、深度木马分析等等动作。

正常流程是：开发一段代码对接安全审核系统，收到邮件后进行初步的审核。当发现可能有风险的邮件后，就根据安全系统定义的 API 将邮件数据投递过去。如果还需要对接其他业务系统，也需要重复整个过程，因此重复开发成本和维护成本就很高。

为了解决这个问题，我们可以引入事件处理平台。

如上图所示，邮件系统对接事件中心，将有风险的邮件数据上报到事件处理平台。事件处理平台会根据事先设定的规则，自动化调用对应系统的接口进行投递。

从邮件业务开发的角度，工作量只剩对接事件处理平台。而且当需要接入其他业务时，只需要在事件处理平台添加新的事件处理规则即可，没有重复的开发工作量。

最后来看一个比较复杂的商品订单中台场景。

商品订单中台

当前不少企业都会通过ERP、CRM等内部系统来实现企业数字化。此时就会出现多项系统彼此闭环，数据难以统一管理的问题。为了解决它，我们就需要拥有一套数据连接聚合系统，把数据汇总起来。如果全部自定义开发，成本就很高，此时我们就可以引入事件驱动架构。

如上图所示，事件处理平台提供了统一的事件投递规范。业务方产生的不同类型事件（如用户下单、商品入库、订单更新等），通过 CloudEvent API 以相同规范进行投递。由事件处理平台进行事件的过滤、提取后，根据配置的不同路由规则，将对应事件投递给相应的处理目标，完成事件的自动化处理。

在这个场景下，事件处理平台完成了类似业务中台的基础能力。企业也可以基于事件处理平台提供的接口规范以及路由原则，将事件处理平台作为底层架构，完成更复杂的业务中台搭建，从而简化开发成本。

好，在了解了事件处理平台的典型应用场景之后，我们来看一下如何构建事件处理平台。

如何构建事件处理平台

先来看一张整体的架构图。事件处理平台分为接入层、缓存层、运行层、分发层四个部分。

接入层

接入层顾名思义就是用来接收事件数据的。从功能来看，有被动接收和主动拉取两种形态。

被动接收 是指开发部署维护支持 HTTP、CloudEvent 等协议的 Server，并设计上报协议。客户端会根据 Server 规定的协议组织数据并完成上报。

主动拉取 是指客户端没有上报的能力，需要事件处理平台通过一定的方式去事件源拉取事件数据。比如，数据库类的事件源（MySQL Binlog、Mongo ChangeStream等），就需要事件处理平台去主动订阅。

从技术上看，被动接收一般通过维护接收数据的 Server 的集群来实现，需要关注集群的容量和稳定性。主动拉取一般通过一些开源的方案，如 Kafka Connector、Flink CDC、Debezium 等来订阅数据源的数据。或者走纯自研路线，从数据源订阅数据，自研路线的好处就是代码可控性高，底层原理和开源方案差不多。

缓存层

从技术上看，缓存层一般使用的是消息队列集群，比如 Kafka、Pulsar、RocketMQ 等。需要关注以下 3 个问题：

  1. 消息队列集群的性能。即集群的容量，这个一般是业务根据数据源的数据量进行评估。
  2. 数据的可靠性。即接收保存数据后，需要保证数据不会丢失。这个一般需要关注集群的副本数量和一致性协议的选择，更多细节可以回顾一下 第17讲。
  3. 事件数据存储方式 。 即事件源肯定是有归属的，比如事件是归属某一个客户或某一个业务，此时如果有多个事件源，底层如何保存数据。

前两点很好理解，关于第三点，你可能有点模糊，我们重点讲讲。

如下图所示，假设我们有7个事件源（以不同颜色区分）。此时如果所有的事件都存储在一个分区或 Topic 里面，当某个事件源数据量太大导致消费堆积时，同一个分区里面其他事件源的事件处理就会受到影响。

此时就得考虑事件数据存储方式。 所有的事件源存储在一起，还是每个事件源都进行独立的分区或Topic存储。使用哪种方案更合理呢？它们各自存在什么问题？又如何解决？你可以先想想，我们留在思考题中解决。

运行层

运行层主要用来执行事件处理、过滤、分发的逻辑，相当于事件总线的内核。它主要有三个功能。

  1. 提供接口给用户配置事件规则相关的信息，提供增删改的接口，并持久化存储这些数据。
  2. 从缓存中拉取数据，根据预先配置好的规则对数据进行处理、过滤、聚合。
  3. 对处理完成的数据，匹配对应的规则触发调用分发层的接口。

运行层需要关注它底层的运行时是什么？如下图所示，从技术上来看，有内置固定规则运行时和Serverless 运行时两种形态。

内置固定规则运行时，是指我们通过编码实现固定的业务处理、分发逻辑。然后用户通过一些配置项来配置事件处理规则或分发策略，从而触发固定的逻辑处理。

从技术实现来看，这种运行时底层一般是基于自定义编码、Flink/Spark 等来构建的。因为本质上运行时是一个计算层，即拉取数据然后处理。而大数据框架 Flink/Spark 擅长处理的就是这个场景。这种方案的好处是用户开箱即用，使用成本低。缺点是平台开发成本较高、不够灵活，无法满足客户复杂的定制需求。

Serverless 运行时，是指底层的部署形态是 Serverless Funciton。对Serverless Funciton 不熟悉的可以回顾一下 第38讲。即用户可以通过自定义编写、修改函数来实现自定义的逻辑。

比如事件数据处理时，需要从某个第三方系统拉取一些数据进行汇总和计算，这种就算是特殊的自定义逻辑。这种场景就很适合使用 Serverless 运行时。这种方案的好处就是非常灵活，基本可以满足所有场景。缺点是用户需要付出一定的编码成本。

从业界落地来看，事件处理平台都会同时支持这两种运行时，以满足更多用户的需求。最后我们来看看分发层。

分发层

分发层的核心功能就是对接各种下游系统。如下图所示，即集成各种下游系统的 SDK，比如 HTTP SDK、ES SDK、JDBC SDK等，然后供运行层调用。每种下游系统，在运行层只需要对接一次即可。

从使用上看，运行层执行完计算、处理、过滤的逻辑后，就会调用分发层触发具体的操作，比如调用 HTTP API、将数据写入到 DB 等等。

分发层的挑战在于，如何快速低成本地对接各种业务系统。这一点先不展开，你可以思考下如何实现。有需要的话，我们可以在留言区讨论。

除了架构层面四个部分的实现外，事件处理平台还需要保证事件数据不会丢失。

保证数据不丢失

事件处理平台保证数据不丢失包含下面两层含义：

  1. 接入层接收成功的事件数据不能丢失。
  2. 对事件数据的处理必须有结果。

第一点比较好理解，即接入层收到数据后，需要写入缓存层成功后，才能返回成功。还得在缓存系统异常时，保证数据不能丢失。这块依赖的就是编码技巧，以及缓存引擎的选择和使用方式。

第二点需要重点关注，即 一个事件必须有一个执行结果， 结果可以成功也可以失败。成功，可以是写入或者触发事件目标成功。失败的话就需要保存相关的日志，或投递到死信队列。

注意，这里不能出现用户上报事件，但是这个事件却没有任何执行结果的情况。即如果只接收到事件，却没有执行，那这个事件就相当于丢失了。为了跟踪事件的处理过程，就需要记录这个事件的运行轨迹。此时就可以结合 第22讲 的消息轨迹，以及一些开源的可观测性方案（比如日志 + 监控 + Prometheus + Grafana）来观测事件的处理轨迹。

最后来看看事件处理过程中，事件数据的一致性问题。

数据一致性语义

先来看下图，事件数据在事件处理平台可能是链式的处理，此时一个事件就有可能被传递多次。那如何保证事件不会被重复投递呢？

这个问题就是事件处理平台中数据一致性语义，它分为最多投递一次、最少投递一次、精确投递一次三种情况。

  1. 最多一次指消息不会被重复发送，最多被传输一次，但也有可能一次也不传输。
  2. 最少一次指消息不会被漏发送，最少被传输一次，但也有可能被重复传输。
  3. 精确一次指不会漏传输也不会重复传输，每个消息都只会被传输一次。

理想情况下，在每个环节的传递、处理过程中，肯定是做到精确一次（Exactly Once）的语义是最好的。但是我们从 第30讲 知道，Exactly Once 语义需要基于事务实现。而事务的底层需要基于幂等、协调者等机制，性能会降低较多。并且支持 Exactly Once 会让系统复杂度提升很多。所以在实际实现中，我们一般实现的是 最少一次 的语义，即允许少量的重复，让业务侧来处理重复的情况。

当然，ExactlyOnce 语义也是总线的一个技术优化点，如果能支持 ExactlyOnce 语义，将是这款产品的核心竞争力。

前面提到过，事件处理平台在公有云产品化后一般叫做事件总线（EventBridge）。所以接下来我们就来设计实现一个事件总线。

设计实现事件总线架构

先来看一下整体的架构图。

如上图所示，整个架构分为数据源、事件总线、数据目标三个部分。这里我们主要需要完成的是事件总线部分，由接入层、缓存层、运行层、分发层四部分组成。

核心流程拆解

在接入层我们分别提供了支持 HTTP 协议集群、支持 TCP 协议的集群、支持 Connector 集群三种形态。

  • HTTP/TCP 协议的集群是用来满足被动接收的场景。服务端既支持自定义的上报协议，也支持标准的 CloudEvent 协议，用来满足不同场景上报的需求。
  • Connector 集群用来满足主动拉取的场景。从技术上看，Connector 集群有自研、开源两种方案。开源则包含比如各个 MQ Connector（如 Kakfa/RocketMQ Connector、Pulsar IO等）、Flink CDC 等多种方案。

缓存层，我们选择了 Kafka 来作为中间层。因为它具有高性能、稳定性强等特性，能够很好地扛住流量，长期保存数据。

因为我们选择了 Kafka 当缓存层，所以我们接入层的 Connector 集群，底层是基于 Kafka Connector 实现的。Connector 的作用是去主动订阅数据源的数据，比如 MySQL Binlog、Mongo ChangeStream 等等。Kafka Connector 的技术细节我们会在下节课详细展开。

在运行层，我们提供了事件相关的元数据的管理，如事件源/目标、事件规则的增删改查等等。同时提供了固定规则运行时和 Serverless 运行时，来执行固定和灵活的数据处理、过滤、聚合等操作。

我们选择了 Flink 来作为固定规则运行时的底层框架，因为 Flink 的生态丰富，有很多现成的能力可以复用。 Serverless 运行时，则是选择公有云上的 Serverless Function 产品，比如腾讯云的云函数。

在分发层它主要是一段代码库，通过集成不同引擎的SDK，对接不同的下游引擎。这块就比较简单，不再赘述。

讲完了架构图的核心处理流程，从技术上来看，还是有很多技术细节需要考虑，我们来简单看一下。

技术细节

  1. 引入Schema，即事件数据在事件总线中传输，也有上下游的概念。需要保证事件数据在传递过程中，是具备一定的规范格式的。需要注意的是，这里的 Schema 和消息队列的 Schema 概念有点不一样。不过事件总线 Schema 的实现底层有一部分是可以基于消息队列 Schema 来实现的，你可以去回顾一下 第33讲。

  2. 接入层/分发层如何快速集成更多的数据源和目标。即在实际开发过程中，我们需要对接各种各样的数据源。此时就需要思考如何能够低成本地对接各种新的数据源。从技术上看，主要是插件化、代码复用、直接利用开源的插件等几个思路。其中在开源的插件上二次开发是一个推荐的方案。

  3. 接入层集群的拆分、扩缩容、安全控制如何设计。因为接入层集群是暴露在公网的，流量也是最大的，所以我们需要考虑是否给大客户部署独立的接入层集群，如何快速地扩缩容（比如容器化），如何给接口增加鉴权，支持 TLS/SSL 协议，防范DoS攻击等等。

  4. 缓存层的容量拆分问题。从技术上来看，缓存层主要是运维层面需要关注的，主要需要关注集群的运行水位。另外研发层面需要关注事件在缓存层的存储模型。

  5. 运行时固定规则的设计。这块属于功能层面的设计，即我们的事件总线能支持怎样的事件处理规则，比如数据清洗、数据过滤、数据聚合、数据投递等等。这块考验的是产品经理的设计，技术上实现就还好。

  6. 运行时如何集成 Serverless Function。即如何考虑快速集成 Serverlss 运行时，并做到Serverless 运行时的可视化，保证出问题时可以有告警，能快速定位到问题等等。

上面这几个问题，我们稍微提供了几个思路，没有特意展开，细节很多，如果需要我们可以在留言区讨论。

总结

事件驱动架构（EDA）也称为事件总线，它的作用是主动拉取或被动接收上游不同事件源的数据，然后根据配置好的事件规则，触发执行不同的业务逻辑。

在事件驱动架构中传递的数据称为事件，事件由格式和内容两部分组成。为了规范数据的格式和内容，业界推出了 CloudEvents 规范，目的是定义通用的、标准化的事件格式。CloudEvents 规范由事件上下文、事件数据、传输协议三部分组成。

事件驱动架构在自动化运维、数据集成、业务数据中台等场景中有着广泛的应用。

从技术上来看，事件驱动架构由数据源、事件处理平台、数据目标三部分组成。其中时间处理平台是架构的核心，它由接入层、缓存层、运行层、分发层四部分组成。事件处理平台在处理事件的过程中，需要注意保证数据不丢失，并保证数据具有一定的一致性语义。运行时来满足多种业务场景。

流的事件驱动架构来看，接入层一般会适配多种协议（比如基于HTTP或TCP的自定义协议或标准协议）来满足业务主动上报的场景，同时也会提供主动订阅的组件（比如Connector集群、Flink CDC 集群等等）来订阅数据源的数据。在缓存层，一般选择能承载大流量的流场景的组件，比如选择Kafka、Pulsar来作为业务管道。运行层会同时支持固定规则的运行时和Serverless 运行时。分发层就比较简单，定义接口，调用下游的 SDK 写入数据即可。

思考题

事件源底层的存储模型，是所有事件源存储在一起，还是每个事件源进行独立的分区或Topic存储，使用哪种方案更合理呢？它们各自存在什么问题？又如何解决？

欢迎分享你的想法，如果觉得有收获，也欢迎你把这节课分享给感兴趣的朋友。我们下节课再见！

上节课思考闭环

在你当前的业务中，还有哪些场景能够基于 Serverless 架构实现数据的流式处理？

分享一个我的，数据库订阅场景，举例说明。

如下图所示，我们的业务需要将 MongoDB、MySQL、上游 MQ 等异构数据源同步到ClickHouse里面。数据源的数据首先同步到消息队列（如Kafka），然后进行格式化和数据转换。

这里的数据转换不是简单的数据格式化和清洗，还需要和第三方系统交互，渲染数据，并且还需要多个库、多个表之间的数据进行关联，聚合成一个大列，另外还有一些业务的特殊逻辑。

我们的解决思路依旧是通过 Serverless Function来实现。首先编写一段函数逻辑，逻辑如下：

1. 消费Kafka的数据。

2. 进行格式转换、多系统交互、多行合并、数据裁剪等操作。虽然看起来很复杂，但是在函数式编程里面，其实就是100多行代码的实现量。

3. 最后将处理完成的数据写入到下游的ClickHouse。

通过这三步，我们简单地实现了一段函数逻辑，进而实现了一个实时处理场景，它和Spark的运行效果是一样的，只是我们写一个函数的成本和写一个Spark Jar包的成本完全不是一个量级。