Eventbridge学习

简介: Eventbridge学习

Eventbridge是一个云原生事件驱动引擎,基于DDD领域驱动模型实现,其中有相关概念和运行的流程是需要我们了解的。以下内容基于rocketmq官网和自己的学习。

一、相关概念和使用场景

1.eventsource

事件源,用于管理发送到eventbridge的事件,对应CNCF中cloudEvent事件体的source字段。

代表事件发送的源头,通常与微服务系统相对应,也即属于系统层级的一个概念,类似于监控中应用服务级别的概念。

2.eventbus

事件总线,用于存储发送到eventbridge的事件。可以是Local、RocketMQ、Kafka等。

事件生产者发送事件时,需指定事件总线。

3.eventrule

事件规则,当消费者需要订阅事件时,可以通过规则配置过滤和转换信息,将事件推送到指定的目标端。

事件规则包含三部分:

事件过滤、事件转换、事件目标

4.filterPattern

过滤规则,用于在规则中配置过滤出目标端需要的事件,可以对事件总线上的事件进行过滤。

目前支持:指定配置、前缀配置、后缀匹配、除外配置、数值匹配、数组匹配以及复杂的组合逻辑匹配。

5.transform

事件转换,将事件格式转换成目标端需要的数据格式

目前eventbridage支持的事件转换能力:

完整事件:不做转换,直接投递原生cloudEvent

部分事件:通过jsonpath语法从cloudEvent中提取出需要投递到事件目标的内容

常量:事件只起到触发器的作用,投递内容为常量

模板转换器:通过定义目标,灵活地渲染投递处理地事件格式

6.eventTarget

事件目标端,真正地事件消费者,消费者只需要按照自己地业务领域模型设计,提供一个公共地API,eventbridge就会按照Api定义需要地数据格式,将事件安全可靠地推送给target消费者。

二、事件特性

已发生、无期望、天然有序且唯一、具象化

三、RocketMQ Eventbridge是如何工作的

下图来源于Apache RocketMQ的官网:

img

1.确认事件标准:基于CNCF的cloudEvent标准

2.建立事件中心:不同应用属于不同的事件,建立一个事件中心管理不同的事件

3.定义事件格式:明确对应的格式

4.订阅规则:订阅的时候,会存在对应的规则,而这些规则是执行过滤、转换的基础

5.事件总线:建立source和target之间的联系

也即:

source->事件中心->事件总线->filter(event)->transform->target

四、源码学习

根据前面我们提到的概念,在代码Adapter中,我们可以看到对应的控制层的相关的信息:


而这些controller中涉及的操作都是基于Spring WebFlux响应式编程实现。也即Flux和Mono实现返回数据格式对象。

1.Flux和Mono的联系和区别

其中:

Flux 是一个发出(emit)0-N个元素组成的异步序列的Publisher<T>,可以被onComplete信号或者onError信号所终止。在响应流规范中存在三种给下游消费者调用的方法 onNext, onComplete, 和onError。

Mono 是一个发出(emit)0-1个元素的Publisher<T>,可以被onComplete信号或者onError信号所终止。

两者的区别在于返回的数据:

一个是集合,一个是对象或者空,同时Flux和Mono之间可以转化。

2.putEvents操作和发送消息

其中EventDataController中会执行放入事件,发送消息的过程,比如putEvents和putHttpEvents这两个方法:

这里涉及到具体的转换器,可以看的作者的思路是基于EventConverterAdapter进行适配,这个过程中借助了map初始化,然后进行匹配请求头中的类型,然后拿到具体的转换器进行转换。

发送消息:

上面的这些controller都涉及到和数据库交互。

3.refresh刷新的场景

这个过程其实和消费是栖息相关的,因为消费者的增加和减少都会影响。类似重平衡中,消费者发生变化的时候,就会发生重平衡,因此这个操作也非常重要。

4.在runtime中会涉及到的相关操作

1)初始化信息


这个是后续操作的基础。

三种都继承了ServiceThread,可以看到最终实现了Runnable接口。


2)同时可以看到EventTargetTrigger的run方法里面:


执行记录放入到sinkTask中,同时提交偏移量,调用rocketmq。

3)EventRuleTransfer相关信息的处理


此时会调用转换引擎,通过openMessaging的Api接口,从而进入到转换引擎TransformEngine,进行EventBridgeTransform、EventBridgeFilterTransform、CloudEventTransform相关操作,也即相关的转换过滤这些操作都是在这里进行的。此时基于Transform中的process方法进行操作:


4)EventBusListener监听事件拉取消息记录放入上下文


5.消息拉取消费

storage模块中,我们可以看到我们想看的类:


可以看的eventbridge基于LitePullConsumerImpl进行poll操作,poll的过程基于本地消息缓存实现。


eventbirdge的学习,我们就到这里。整体上看,作者的思路是非常清晰的。

相关实践学习
快速体验阿里云云消息队列RocketMQ版
本实验将带您快速体验使用云消息队列RocketMQ版Serverless系列实例进行获取接入点、创建Topic、创建订阅组、收发消息、查看消息轨迹和仪表盘。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
人工智能 安全 Cloud Native
阿里云事件总线 EventBridge 正式商业化,构建智能化时代的企业级云上事件枢纽
阿里云事件总线EventBridge自2020年发布以来,致力于构建统一的事件枢纽,支持微服务架构演进。其核心特性包括稳定安全、高性能低成本、开放集成及统一事件标准,适用于EDA、流式ETL、AI数据集成等多种场景。EventBridge于2025年6月3日正式商业化,提供灵活计费模式,包括事件量和CU配额计费,帮助企业高效实现松耦合、分布式的事件驱动架构。
|
9月前
|
人工智能 监控 安全
2026年阿里云+本地部署OpenClaw终极进化指南:Agent-Reach skill一条命令实时读取B站、小红书等全平台内容
2026年,AI智能体的核心竞争力已从“模型大小”转向“视野广度”。OpenClaw(原Clawdbot、Moltbot)作为开源AI助手的标杆,虽能实现7×24小时自动化任务执行,但传统使用模式下仍受限于“信息孤岛”——需手动投喂内容、依赖付费API、实时信息获取能力薄弱。而刚爆火的开源工具Agent-Reach(900+ GitHub Star)彻底改变这一现状,它就像给OpenClaw装上“全网眼睛”,零API费、零配置,一条命令即可让AI实时读取Twitter、B站、小红书等全平台内容,真正实现“自主上网、主动感知”。
3256 2
|
10月前
|
Kubernetes Linux Docker
蓝易云:Rocky Linux 8.9配置Kubernetes集群详解,适用于CentOS环境
这是在Rocky Linux 8.9上配置Kubernetes集群的高级概述。确保遵循每个步骤,不要跳过环境设置部分,因为这是集群稳定性的关键。
374 2
|
消息中间件 Cloud Native Serverless
RocketMQ 事件驱动:云时代的事件驱动的不同
本文深入探讨了云时代 EDA 的新内涵及它在云时代再次流行的主要驱动力,包括技术驱动力和商业驱动力,随后重点介绍了 RocketMQ 5.0 推出的子产品 EventBridge,并通过几个云时代事件驱动的典型案例,进一步叙述了云时代事件驱动的常见场景和最佳实践。
248330 156
RocketMQ 事件驱动:云时代的事件驱动的不同
|
存储 安全 Go
Go语言中的map数据结构是如何实现的?
Go 语言中的 `map` 是基于哈希表实现的键值对数据结构,支持快速查找、插入和删除操作。其原理涉及哈希函数、桶(Bucket)、动态扩容和哈希冲突处理等关键机制,平均时间复杂度为 O(1)。为了确保线程安全,Go 提供了 `sync.Map` 类型,通过分段锁实现并发访问的安全性。示例代码展示了如何使用自定义结构体和切片模拟 `map` 功能,以及如何使用 `sync.Map` 进行线程安全的操作。
713 9
|
Kubernetes 监控 负载均衡
10个最佳的Kubernetes工具
Kubernetes生态工具整理
1922 0
10个最佳的Kubernetes工具
|
微服务
成功解决:java.lang.NoSuchMethodError: reactor.netty.http.client.HttpClient.chunkedTransfer(Z)Lreactor/ne
这篇文章讲述了在微服务架构中整合gateway网关时遇到的`java.lang.NoSuchMethodError`错误的解决方法。问题主要是由于`spring-boot-starter-parent`的版本和`spring-cloud-starter-gateway`的版本不匹配所导致。文章提供了具体的版本不一致的错误配置,并给出了匹配的版本配置方案,以及成功测试的截图。
成功解决:java.lang.NoSuchMethodError: reactor.netty.http.client.HttpClient.chunkedTransfer(Z)Lreactor/ne
|
敏捷开发 存储 数据可视化
aPaaS是什么(aPaaS与iPaaS的区别)
依题:aPaaS是什么?aPaaS与iPaaS二者之间的区别在哪?要想了解区别,首先得搞清概念,不然就是在耍流氓!下面本人就从概念到区别用大白话给你一次性讲清楚。
aPaaS是什么(aPaaS与iPaaS的区别)
|
消息中间件 存储 城市大脑
云原生事件驱动引擎(RocketMQ-EventBridge)应用场景与技术解析
RocketMQ 给人最大的印象一直是一个消息引擎。那什么是事件驱动引擎?为什么我们这次要推出事件驱动引擎这个产品?他有哪些应用场景,以及对应的技术方案是什么?本文我们就一起来看下。
1432 0
云原生事件驱动引擎(RocketMQ-EventBridge)应用场景与技术解析