当前位置: 首页 > news >正文

EDA - Spring Boot构建基于事件驱动的消息系统

文章目录

  • 概述
  • 事件驱动架构的基本概念
  • 工程结构
  • Code
    • 创建事件和事件处理器
    • 创建事件总线
    • 创建消息通道和发送逻辑
    • 创建事件处理器
    • 消息持久化
    • 创建消息发送事件
    • 配置 Spring Boot 启动类
    • 测试
    • 运行项目

在这里插入图片描述


概述

在微服务架构和大规模分布式系统中,事件驱动架构(EDA)成为了非常重要的设计模式。通过事件驱动,我们可以解耦系统的各个组件,提高系统的可扩展性、可维护性和响应能力。

接下来,我们将演示一下如何在 Spring Boot 中实现一个基于事件驱动的消息发送和接收流程,从消息的发送、事件的发布到事件的监听。


事件驱动架构的基本概念

在事件驱动架构中,系统的各个组件通过事件进行通信。每个事件代表一个特定的行为或状态变化,当事件发布时,系统的其他部分可以响应这些事件并做出相应的处理。消息发送和接收的流程正是通过发布和监听事件来实现的。

接下来我们使用 Spring Boot 来实现一个基于事件驱动的消息系统。、

系统包含以下几个部分:

  • 消息发送: 消息将通过一个 MessageEventProcessor 进行处理,并且在处理完成后会发布一个事件。
  • 事件发布: 消息成功发送后,通过 ApplicationEventPublisher 发布一个 MessageSentEvent
  • 事件监听: 一个监听器会接收到发布的事件并进行相应的处理(比如记录日志、通知其他组件等)

工程结构

在这里插入图片描述

  • EventBus:事件总线,负责发布事件。
  • MessageEventProcessor:处理消息事件的处理器。
  • EventMessageEventMessageSentEvent:事件类,MessageEventMessageSentEvent继承自Event
  • MessageChannel:消息通道接口,EmailMessageChannel是其具体实现。
  • MessageRepository:消息存储库,用于保存消息事件。
  • MessageChannelConfig:消息通道配置,配置了消息通道的Bean。
  • MessageController:消息控制器,处理发送消息的请求。
  • MessageSentEventListener:监听消息发送事件的监听器。

Code

创建事件和事件处理器

Event.java - 定义基础事件

package com.artisan.booteventbus.domain;public abstract class Event {// 事件的基本字段
}

MessageEvent.java - 定义具体的消息事件

package com.artisan.booteventbus.domain;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;import java.util.Map;@EqualsAndHashCode(callSuper = true)
@Data
@Slf4j
@NoArgsConstructor
@AllArgsConstructor
public class MessageEvent extends Event {private String message;private String channel;private Map<String, Object> metadata;}

EventHandler.java - 定义事件处理器接口

package com.artisan.booteventbus.bus;import com.artisan.booteventbus.domain.Event;public interface EventHandler<T extends Event> {void handle(T event);
}

创建事件总线

EventBus.java - 用于发布事件

package com.artisan.booteventbus.bus;import com.artisan.booteventbus.domain.Event;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;@Component
public class EventBus {private final ApplicationEventPublisher publisher;public EventBus(ApplicationEventPublisher publisher) {this.publisher = publisher;}public void publish(Event event) {publisher.publishEvent(event);}
}

创建消息通道和发送逻辑

MessageChannel.java - 定义消息通道接口

package com.artisan.booteventbus.service;import com.artisan.booteventbus.domain.MessageEvent;import java.util.concurrent.CompletableFuture;public interface MessageChannel {boolean supports(MessageEvent event);CompletableFuture<Void> sendAsync(MessageEvent event);
}

MessageChannelConfig.java - 初始化channel

package com.artisan.booteventbus.config;import com.artisan.booteventbus.service.MessageChannel;
import com.artisan.booteventbus.service.impl.EmailMessageChannel;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.ArrayList;
import java.util.List;@Configuration
public class MessageChannelConfig {@Beanpublic List<MessageChannel> messageChannels() {List<MessageChannel> channels = new ArrayList<>();channels.add(new EmailMessageChannel());// 可以继续添加其他类型的通道return channels;}
}

EmailMessageChannel.java - 实现邮件发送通道

package com.artisan.booteventbus.service.impl;import com.artisan.booteventbus.domain.MessageEvent;
import com.artisan.booteventbus.service.MessageChannel;
import lombok.extern.slf4j.Slf4j;import java.util.concurrent.CompletableFuture;@Slf4j
public class EmailMessageChannel implements MessageChannel {@Overridepublic boolean supports(MessageEvent event) {return "email".equals(event.getChannel());}@Overridepublic CompletableFuture<Void> sendAsync(MessageEvent event) {return CompletableFuture.runAsync(() -> {// 模拟邮件发送System.out.println(Thread.currentThread().getName() + "- Sending email: " + event.getMessage());log.info("Sending email: {}", event.getMessage());});}
}

创建事件处理器

MessageEventProcessor.java - 处理消息事件,保存事件并发送

package com.artisan.booteventbus.bus;import com.artisan.booteventbus.dao.MessageRepository;
import com.artisan.booteventbus.domain.MessageEvent;
import com.artisan.booteventbus.domain.MessageSentEvent;import com.artisan.booteventbus.service.MessageChannel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.scheduling.annotation.Async;import java.util.List;@Component
public class MessageEventProcessor implements EventHandler<MessageEvent> {private final EventBus eventBus;private final MessageRepository messageRepository;private final List<MessageChannel> channels;@Autowiredpublic MessageEventProcessor(EventBus eventBus, MessageRepository messageRepository, List<MessageChannel> channels) {this.eventBus = eventBus;this.messageRepository = messageRepository;this.channels = channels;}/*** @param event* Asyn 请使用自定义线程池,这里仅仅是 为了演示异步*/@Async@Overridepublic void handle(MessageEvent event) {// 1. 消息持久化messageRepository.save(event);// 2. 通道路由MessageChannel channel = channels.stream().filter(ch -> ch.supports(event)).findFirst().orElseThrow();// 3. 异步发送channel.sendAsync(event).thenRun(() -> eventBus.publish(new MessageSentEvent(event)));}}

消息持久化

MessageRepository.java - 用于消息的持久化(可以使用内存或数据库)

package com.artisan.booteventbus.dao;import com.artisan.booteventbus.domain.MessageEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Repository;import java.util.ArrayList;
import java.util.List;@Slf4j
@Repository
public class MessageRepository {private final List<MessageEvent> messageStore = new ArrayList<>();public void save(MessageEvent event) {// 模拟存储messageStore.add(event);System.out.println(Thread.currentThread().getName() + " - Message saved: " + event.getMessage());log.info("Message saved {}", event.getMessage());}
}

创建消息发送事件

MessageSentEvent.java - 定义发送后的事件

package com.artisan.booteventbus.domain;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;@EqualsAndHashCode(callSuper = true)
@Data
@NoArgsConstructor
@AllArgsConstructor
public class MessageSentEvent extends Event {private MessageEvent originalEvent;}

配置 Spring Boot 启动类

package com.artisan;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;@EnableAsync(proxyTargetClass=true)
@SpringBootApplication
public class BootEventBusApplication {public static void main(String[] args) {SpringApplication.run(BootEventBusApplication.class, args);}}

测试

为了测试整个架构,创建一个控制器来模拟发送消息。

package com.artisan.booteventbus.controller;import com.artisan.booteventbus.domain.MessageEvent;
import com.artisan.booteventbus.bus.EventBus;
import com.artisan.booteventbus.bus.MessageEventProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;import java.util.HashMap;@RestController
@RequestMapping("/messages")
public class MessageController {private final EventBus eventBus;private final MessageEventProcessor eventProcessor;@Autowiredpublic MessageController(EventBus eventBus, MessageEventProcessor eventProcessor) {this.eventBus = eventBus;this.eventProcessor = eventProcessor;}@RequestMapping("/send")public String sendMessage(@RequestParam String message, @RequestParam String channel) {MessageEvent event = new MessageEvent(message, channel, new HashMap<>());eventProcessor.handle(event); // 异步处理消息return "Message is being processed";}
}

运行项目

http://localhost:8080/messages/send?message=artisan&channel=email

在这里插入图片描述
在这里插入图片描述

在这里插入图片描述

相关文章:

EDA - Spring Boot构建基于事件驱动的消息系统

文章目录 概述事件驱动架构的基本概念工程结构Code创建事件和事件处理器创建事件总线创建消息通道和发送逻辑创建事件处理器消息持久化创建消息发送事件配置 Spring Boot 启动类测试运行项目 概述 在微服务架构和大规模分布式系统中&#xff0c;事件驱动架构&#xff08;EDA&a…...

使用vue-seamless-scroll实现echarts图表大屏滚动,出现空白间隔的解决方案

一、背景介绍 最近的业务开发需求&#xff0c;想要实现echarts图表大屏滚动&#xff0c;小编首先采用vue-seamless-scroll进行实现&#xff0c;结果发现第二屏出现空白间隔&#xff0c;尝试了多种解决方案均不生效&#xff0c;最终选择换一个方案。 二、封装的ScrollList组件…...

ios使用UIScrollView和PageControl创建图片轮播

1.创建cocoa touch class 2.同时创建xib页面 3.SceneDelegate设置根视图控制器 // // SceneDelegate.m // iosstudy2024 // // Created by figo on 2024/8/5. //#import "SceneDelegate.h" #import "WidgetViewController.h"interface SceneDelegate …...

3D 生成重建024-LGM第一个开源的3D生成大模型!

3D 生成重建024-LGM第一个开源的3D生成大模型 文章目录 0 论文工作1 论文方法2 实验效果 0 论文工作 这篇论文介绍了一种名为LGM&#xff08;大型多视角高斯模型&#xff09;的新方法&#xff0c;用于从单视角图像或文本提示生成高分辨率的三维内容。该方法的核心思想是双重的…...

linux目录权限

一、目录权限的基本概念 Linux中的每个文件和目录都有与之关联的权限&#xff0c;这些权限决定了谁可以读取、写入或执行它们。权限分为三组&#xff1a; 所有者&#xff08;Owner&#xff09;权限&#xff1a;目录所有者的权限群组&#xff08;Group&#xff09;权限&#x…...

语言模型使用心得

使用像文心一言这样的语言模型&#xff0c;在撰写文章时确实能提供极大的帮助。然而&#xff0c;重要的是我们要明确主次关系&#xff1a;自己的创意和内容应当是文章的核心&#xff0c;而语言模型则扮演着一个辅助角色&#xff0c;帮助我们梳理思路&#xff0c;使文章条理更加…...

ChatGPT客户端安装教程(附下载链接)

用惯了各类AI的我们发现每天打开网页还挺不习惯和麻烦&#xff0c;突然发现客户端上架了&#xff0c;懂摸鱼的人都知道这里面的道行有多深&#xff0c;话不多说&#xff0c;开整&#xff01; 以下是ChatGPT客户端的详细安装教程&#xff0c;适用于Windows和Mac系统&#xff1a…...

Electron 基础+传值+引用+安全

文章目录 概要elctron 生命周期及窗口应用主进程与渲染进程交互技术细节electron 中需要注意的安全问题 概要 一、Electron简介 Electron是一个开源框架&#xff0c;它允许开发者使用JavaScript、HTML和CSS构建跨平台的桌面应用程序。它基于Chromium&#xff08;谷歌浏览器的…...

手机租赁系统全面解析与开发指南

内容概要 手机租赁系统已经成为现代商业中不可或缺的一部分&#xff0c;尤其是在智能手机普及的时代。随着消费者对新机型兴趣的不断增加&#xff0c;大家纷纷走上了“试一试再买”的道路&#xff0c;手机租赁这条路因此越走越宽。这部分的市场需求让创业者们看到了机会。不仅…...

mongoDb的读session和写session权限报错问题

go在使用mongoDb时用到了全局会话&#xff0c;发现在创建的session的逻辑相同&#xff0c;首先会进行数据的查询&#xff0c;此时获取了全局session执行读操作&#xff0c;查询所有文档&#xff0c;则当前会话为读会话&#xff0c;当再去插入时发现会报错&#xff0c;此时sessi…...

Centos在2024年6月30日停止维护后如何换yum源安装组件

现象&#xff1a; 在centos7里使用yum安装报错&#xff1a; Loading mirror speeds from cached hostfile Could not retrieve mirrorlist http://mirrorlist.centos.org/?release7&archx86_64&repoos&infrastock error was 14: curl#6 - “Could not resolve…...

阿里云ACP云计算模拟试题(附答案解析)

1、将基础设施作为服务的云计算服务类型是_____服务。 A.laas B.Paas C.SaaS D.Daas 答案&#xff1a;A 解析&#xff1a;基础设施即服务有时缩写为 IaaS&#xff0c;包含云 IT 的基本构建块&#xff0c;通常提供对联网功能、计算机&#xff08;虚拟或专用硬件&#x…...

简单的爬虫脚本编写

一、数据来源分析 想爬取一个网站的数据&#xff0c;我们首先要进行数据分析。通过浏览器F12开发者工具栏进行抓包&#xff0c;可以分析我们想要的数据来源。 通过关键字搜索&#xff0c;可以找到相对应的数据包 二、爬虫实现 需要用到的模块为&#xff1a;request&#xf…...

[MySQL基础](三)SQL--图形化界面+DML

本专栏内容为&#xff1a;MySQL学习专栏 &#x1f493;博主csdn个人主页&#xff1a;小小unicorn ⏩专栏分类&#xff1a;MySql &#x1f69a;代码仓库&#xff1a;小小unicorn的代码仓库&#x1f69a; &#x1f339;&#x1f339;&#x1f339;关注我带你学习编程知识 目录 图…...

11.23[大数据]

PRO1:LSTM模型预测输出都是同一个值&#xff1f; 画出来的图像就是一条横线 这个搜了搜&#xff0c;原因可能有很多&#xff0c;但感觉最主要的原因极可能是激活函数选择不当&#xff0c;以及层的搭建不合适 原模型是 REF https://zhuanlan.zhihu.com/p/654325094 https:/…...

C++ 游戏开发进阶:打造更精彩的游戏世界

在之前的 C 游戏开发入门教程中&#xff0c;我们已经了解了游戏开发的基本概念和一些简单的实现方法。现在&#xff0c;让我们进一步深入探讨 C 游戏开发中的进阶技术&#xff0c;为玩家打造更精彩、更具沉浸感的游戏体验。 一、游戏物理引擎的集成 物理引擎是现代游戏开发中…...

想在iPad上远程操作安卓手机的APP,怎样实现iPad远程控制安卓?

学生党或互联网行业的打工人&#xff0c;人手连三台电子设备也很常见&#xff0c;手机、平板还有笔记本电脑一大堆&#xff0c;如果出门要全带上&#xff0c;背包压力也变大。 有没有想过用远程控制功能&#xff0c;让iPad远程控制安卓手机&#xff1f;这样做&#xff0c;出门就…...

GPS北斗卫星授时服务器功能是什么?应用是什么?

GPS北斗卫星授时服务器功能是什么&#xff1f;应用是什么&#xff1f; GPS北斗卫星授时服务器功能是什么&#xff1f;应用是什么&#xff1f; 摘 要:首先对计算机网络时间同步相关技术进行了介绍,然后阐述了时间同步技术在现代计算机网络中的应用与发展,最后指出时间同步网络…...

利用Java爬虫获取商品数据的完整指南

在当今数字化时代&#xff0c;数据已成为企业和个人决策的关键资源。特别是在电商领域&#xff0c;获取商品数据对于市场分析、价格监控和竞争对手分析至关重要。Java作为一种强大且广泛使用的编程语言&#xff0c;非常适合开发复杂的爬虫系统。本文将详细介绍如何利用Java编写…...

mysql 迁移达梦数据库出现的 sql 语法问题 以及迁移方案

迁移方案&#xff1a; 1.下载官方DM8开发版 产品下载-达梦数据 2.会下载到win系统下的左下角的开始 1.1.2 创建工程 右击空白处&#xff0c;新建 1.1.3 新建迁移 1.1.3.1 选择迁移方式 MySql迁移DM 1.1.3.2 配置数据源 输入你的mysql配置后&#xff0c;刷新&#xff0c;选择…...

从零实现富文本编辑器#5-编辑器选区模型的状态结构表达

先前我们总结了浏览器选区模型的交互策略&#xff0c;并且实现了基本的选区操作&#xff0c;还调研了自绘选区的实现。那么相对的&#xff0c;我们还需要设计编辑器的选区表达&#xff0c;也可以称为模型选区。编辑器中应用变更时的操作范围&#xff0c;就是以模型选区为基准来…...

Redis相关知识总结(缓存雪崩,缓存穿透,缓存击穿,Redis实现分布式锁,如何保持数据库和缓存一致)

文章目录 1.什么是Redis&#xff1f;2.为什么要使用redis作为mysql的缓存&#xff1f;3.什么是缓存雪崩、缓存穿透、缓存击穿&#xff1f;3.1缓存雪崩3.1.1 大量缓存同时过期3.1.2 Redis宕机 3.2 缓存击穿3.3 缓存穿透3.4 总结 4. 数据库和缓存如何保持一致性5. Redis实现分布式…...

2021-03-15 iview一些问题

1.iview 在使用tree组件时&#xff0c;发现没有set类的方法&#xff0c;只有get&#xff0c;那么要改变tree值&#xff0c;只能遍历treeData&#xff0c;递归修改treeData的checked&#xff0c;发现无法更改&#xff0c;原因在于check模式下&#xff0c;子元素的勾选状态跟父节…...

ServerTrust 并非唯一

NSURLAuthenticationMethodServerTrust 只是 authenticationMethod 的冰山一角 要理解 NSURLAuthenticationMethodServerTrust, 首先要明白它只是 authenticationMethod 的选项之一, 并非唯一 1 先厘清概念 点说明authenticationMethodURLAuthenticationChallenge.protectionS…...

Python基于历史模拟方法实现投资组合风险管理的VaR与ES模型项目实战

说明&#xff1a;这是一个机器学习实战项目&#xff08;附带数据代码文档&#xff09;&#xff0c;如需数据代码文档可以直接到文章最后关注获取。 1.项目背景 在金融市场日益复杂和波动加剧的背景下&#xff0c;风险管理成为金融机构和个人投资者关注的核心议题之一。VaR&…...

Unity中的transform.up

2025年6月8日&#xff0c;周日下午 在Unity中&#xff0c;transform.up是Transform组件的一个属性&#xff0c;表示游戏对象在世界空间中的“上”方向&#xff08;Y轴正方向&#xff09;&#xff0c;且会随对象旋转动态变化。以下是关键点解析&#xff1a; 基本定义 transfor…...

MySQL体系架构解析(三):MySQL目录与启动配置全解析

MySQL中的目录和文件 bin目录 在 MySQL 的安装目录下有一个特别重要的 bin 目录&#xff0c;这个目录下存放着许多可执行文件。与其他系统的可执行文件类似&#xff0c;这些可执行文件都是与服务器和客户端程序相关的。 启动MySQL服务器程序 在 UNIX 系统中&#xff0c;用…...

Qwen系列之Qwen3解读:最强开源模型的细节拆解

文章目录 1.1分钟快览2.模型架构2.1.Dense模型2.2.MoE模型 3.预训练阶段3.1.数据3.2.训练3.3.评估 4.后训练阶段S1: 长链思维冷启动S2: 推理强化学习S3: 思考模式融合S4: 通用强化学习 5.全家桶中的小模型训练评估评估数据集评估细节评估效果弱智评估和民间Arena 分析展望 如果…...

手动给中文分词和 直接用神经网络RNN做有什么区别

手动分词和基于神经网络&#xff08;如 RNN&#xff09;的自动分词在原理、实现方式和效果上有显著差异&#xff0c;以下是核心对比&#xff1a; 1. 实现原理对比 对比维度手动分词&#xff08;规则 / 词典驱动&#xff09;神经网络 RNN 分词&#xff08;数据驱动&#xff09…...

【QT控件】显示类控件

目录 一、Label 二、LCD Number 三、ProgressBar 四、Calendar Widget QT专栏&#xff1a;QT_uyeonashi的博客-CSDN博客 一、Label QLabel 可以用来显示文本和图片. 核心属性如下 代码示例: 显示不同格式的文本 1) 在界面上创建三个 QLabel 尺寸放大一些. objectName 分别…...