基于 EventBridge 轻松搭建消息集成应用

作者：昶风

前言

本篇文章主要介绍基于阿里云 EventBridge 的消息集成能力，结合目前消息产品的需求热点，从能力范围到场景实战，对 EventBridge 的消息集成解决方案进行了概要的介绍。

从消息现状谈起

消息队列作为应用解耦，流量削峰填谷的有效工具，早已被开发者广泛应用于各类分布式服务中。随着业务的不断发展迭代，越来越多的服务面临着稳定性和可维护方面的挑战，对消息队列的需求也不再局限于普通的订阅-消费，这里列举 3 个场景的能力需求：

消息路由能力

这里主要的需求场景是数据同步和异地灾备。对于大多数消息队列用户而言，消息队列中的业务数据一般由一类服务来消费和处理，当用户有多个应用，且需要一个全局视野来汇总分析这些不同应用的数据时，消息路由就成了一个避不开的能力需求。同样，出于数据灾备的考虑，云上大用户通常也有在不同 region 同步消息数据的需求。

但在云上实现消息路由可能并不是一件容易的事情。众所周知，云上个 region 之间网络是严格隔离的，用户想要实现数据的跨 region 传输，要么使用 CEN 等网络打通方案，要么使用公网进行数据传输。前者需要业务在设立之初就有完备的网络规划，同时 CEN 的费用成本也是一笔不小的开销；后者则需要考虑公网带宽成本，和将数据暴露在公网所面临的安全问题。

消息加工能力

设想这样一个场景，电商平台上，用户的订单信息是由 RocketMQ 来负责承载的，下游的支付、库存、营销部门都需要消费订单信息，由于历史和技术栈原因，这几个部门使用的消息产品类型各不相同，分别是 RocketMQ、RabbitMQ 和 Kafka，而且这几个部门所需要的并不是全量的原始消息，而是部分自己感兴趣的。同时，自身的接口也对数据 schema 做出了一定的要求，并不期望原始消息投递过来。

此时，单纯的消息路由似乎也不能满足需求。如果这项任务交给一款产品来处理，那么这款产品需要满足 2 点能力，第一就是将消息路由至其他种类消息产品的能力，另外一点就是必须具备数据的过滤加工能力。

生态扩展能力

传统消息用户，其消息的生命周期仅在自身业务的发送与消费阶段。但在万物上云的今天，如果可以利用种类繁多，功能强大的各类云产品来丰富消息的上下游，则将极大地发掘消息数据的潜在价值。例如用户可能需要将 SLS 的日志数据导入到消息队列，以实现离线的数据加工能力；又或者可能需要将消息队列中的数据用于触发函数计算，以实现整个应用的 Serverless 化。

但现状是，用户如果自身去实现各个生态产品的对接，则面临着开发工作量大，后续维护繁琐的问题。最理想的方式是有一款开箱即用的工具，能够对接足够多的生态，而且本身使用也尽量简单。那么是否有一款云产品可以解决以上问题，帮助消息用户实现便捷、可靠的消息集成能力呢？这里就需要介绍本期的主角：阿里云事件总线 EventBridge。

事件总线与消息集成

事件总线 EventBridge 是阿里云提供的一款无服务器事件总线服务，支持阿里云服务、自定义应用、SaaS 应用以标准化、中心化的方式接入，并能够以标准化的 CloudEvents 1.0 协议在这些应用之间路由事件。

EventBridge 的核心能力可以归纳为 3 点：统一事件服务、数据管道与开放与集成。

• 统一事件服务

  在 EventBridge 出现之前，各个云产品的事件相互独立，用户无法在一个地方地方完成对全部云产品事件的收集和处理。EventBridge 的出现打破了这一局面，其收纳了云上的绝大多数云产品事件，并在用户侧提供统一、体系的展示交互，让云上用户可以更加系统的梳理、处理云产品事件。

• 数据管道

  EventBridge 并不简简单单仅覆盖云上事件，也提供数据管道的能力，实现数据的联通交互。例如用户可以使用 EventBridge 创建消息路由任务，将消息队列中的数据导入到其他消息队列，也可以创建任务将 SLS 中的日志数据导入到消息队列中来。目前 EventBridge 在源端和目标端均有丰富的产品种类支持。

• 开放与集成

  EventBridge 开放了多种语言的 SDK，方便不同技术栈的开发者可以快速方便的使用 EventBridge。同时 EventBridge 也兼容了云上标准 Cloudevents 的 sdk，即用户使用业界开源 sdk 也可以方便的将事件投递至 EventBridge。在事件 schema 定义上，EventBridge 采用了目前云上事件的事实标准 cloudevents 1.0，保证了使用其他事件服务的客户可以无缝迁移到阿里云 EventBridge，其在阿里云 EventBridge 上的事件也可以快速接入到其他主流 EDA 引擎，避免了厂商锁定的情况。

针对第一部分谈及的消息用户面临的问题，EventBridge 在消息集成领域提供了完整的解决方案，具体包括消息流转（也称为消息路由）能力解决方案，消息处理能力解决方案与生态支持能力解决方案。

消息流转解决方案

目前 EventBridge 已经支持了云上大部分消息产品的接入，如 RocketMQ、RabbitMQ、Kafka、MNS 和 MQTT。使用这些消息产品的用户，可以在对应消息产品的控制台，消息流入流出页面，或者是 EventBridge 的控制台来创建对应的路由任务。

针对部分云上用户的需求，EventBridge 还支持跨 region 和跨账号的消息路由。

针对跨 region 场景，当用户需要创建北京到上海的消息路由时，在消息队列控制台消息流入流出页面，选择好对应的源实例信息和目标实例信息，一键创建启动即可。

当用户有跨账号消息路由的需求（例如一家公司不同部门各自拥有独立账号，需要对各部门进行数据汇总的场景）时，使用 EventBridge 的事件总线可以完成不同账号间数据的同步。具体详细创建任务步骤可以参考 EventBridge 的官方文档。

当用户自身的消息源不是云上产品时，目前 EventBridge 也提供了多种便捷的方式帮助用户快速对接。如果用户是在云上自建的消息服务，如自建 RocketMQ，EventBridge 已经支持用户通过简单的配置快速将消息数据接入进来。如果用户服务是部署在自建机房，使用的消息产品并非云上已支持的产品，这个时候用户也可以使用 EventBridge 提供的 sdk 主动将事件投递上来，进而写入到云上各类丰富的下游云产品，此类场景适合迁移上云客户。

消息处理解决方案

EventBridge 目前提供了丰富的事件提取和加工能力，用户可以配置多种多样的过滤与加工规则来满足其业务需要。

举个例子，用户的事件有一个字段是订单类型，内容有在线订单和线下订单两类。用户期望过滤订单类型为在线订单的数据，投递到处理在线业务的函数计算服务。这里就可以在过滤规则中，参考指定值匹配，设置过滤规则中对应字段的值为在线订单。

除了指定值匹配，EventBridge 还支持前缀匹配、后缀匹配、除外匹配、数值匹配、数组匹配、复杂组合匹配等逻辑，满足客户绝大部分场景诉求。

有时用户并不期望原样的事件内容传递至目标端，而是加工成一定格式后再进行投递，对此 EventBridge 也提供了事件转换能力。目前支持的转换器类型包括完整事件、部分事件、常量、模板、和函数计算转换器。这些转换器分工各有差异，搭配使用可以帮助用户从事件体中提取出部分有价值字段，并按照要求拼接成目标端需要的数据格式。

消息生态解决方案

针对消息生态闭环的问题，EventBridge 的介入可以有效地扩充其上下游生态丰富度。

在消息流入侧，EventBridge 支持消息类产品如 MNS、Kafka，数据库类产品如 DTS、云上管控、报警事件以及自定义事件的接入，使用户可以便捷地获取各类事件数据。

在消息流出侧，EventBridge 支持计算类产品如 FC、SAE，数据库类产品 RDS、通知类产品如钉钉、微信，通用类 API 如 http target 以及第三方 Saas 应用的触发。消息产品的用户通过简单的配置就可以将消息导入到下游服务以产生业务价值。

场景与实战

下面，我们列举几个场景并看下在消息产品控制台上创建对应任务的步骤。

消息路由场景

第一个场景就是跨可用区消息路由场景，用户如果有灾备或者数据汇总的需求时，可能会面临此问题。

如图所示，用户的业务部署在杭州、北京、青岛 3 个 region。每个 region 都独立处理本 region 的业务数据。但为了搭建一个全局范围的数据大盘服务，需要用户汇总各个 region 的消息产品数据。

在这里，用户可以使用 EventBridge 的消息集成能力，分别创建杭州到北京和青岛到北京的消息路由任务，再使用一个独立的消费组去消费这些消息，即可完成跨 region 的消息数据汇总。

但有一点需要注意，在异地容灾场景下，用户需要配置双向路由时，一定要配置合适的过滤规则以避免数据成环。

现在我们看下创建此跨 region 消息路由任务的步骤。由于这些服务是由 EventBridge 来提供的，所以需要用户提前开通 EventBridge 服务，并在概览页授予 EventBridge 访问相关云产品服务的权限。

这里以创建一个北京到上海的 RocketMQ 消息路由任务为例，首先登陆 RocketMQ 控制台。

1. 在控制台左侧页面，选择消息流出，点击创建任务。
2. 在创建任务栏中，流出类型选择消息队列 RocketMQ。
3. 随后在弹出的任务配置下拉参数中，源端的地域选择为北京，RocketMQ 实例信息按照实际情况填写。

4. 在目标端，选择流入的上海 RocketMQ 实例，填入相关信息。

5. 消息过滤和消息转换按照需要填写，点击确定。
6. 等待 1-2 分钟，就可以看到消息路由任务已经处于运行中状态，这表明消息路由任务已经成功创建并运行。

离线加工场景

第二个场景是一个消息离线数据分析的场景。

用户除了对 MQ 中的消息进行即时业务消费，也需要将其导入到 SLS 中，以用作数据分析和报警配置。同时用户还对写入到 SLS 的内容格式有一定要求。

这里用户可以在消息流入流出页面创建一个任务，将数据经由 EventBridge 导入到 SLS。在配置任务的过程中，按照图中示例的变量和模板配置，就可以将原始数据内容中的 name 和 state 字段提取出来，按照需求生成新的内容格式。

还是以 RocketMQ 举例，在 RocketMQ 控制台消息流出页面，点击创建任务。

1. 在流出类型中选择日志服务 SLS。
2. 在下方的任务参数配置中，在源端选择对应的 RocketMQ 实例信息，在目标端选择对应的 SLS project 和 logStore。

3. 在日志内容格式部分，按照图中提示的变量和模板配置，将对应内容填入，点击确定。

4. 等待 1-2 分钟，就可以看到消息写入 SLS 的任务已经处于运行中状态，这表明任务已经成功创建并运行。

以上就是阿里云 EventBridge 的消息集成能力的介绍。后续 EventBridge 还会加大对消息领域能力的支持，帮助用户低成本的实现消息数据价值。感兴趣的朋友可以关注 EventBridge 官网与官方文档，第一时间获取最新产品消息。