使用 Spring Cloud Bus 和 Spring Cloud Stream 集成实现基于消息的事件驱动

2023-04-23 00:54:40 来源:腾讯云

基于消息的事件驱动是一种常见的微服务架构设计模式,它将不同的微服务之间通过消息进行通信,实现松耦合、高可伸缩性和高可靠性。在这种架构下,每个微服务都是独立的,它们可以在消息传递的过程中进行异步操作,这使得整个系统的性能得到了很大的提升。

在 Spring Cloud 中,我们可以使用 Spring Cloud Bus 和 Spring Cloud Stream 集成来实现基于消息的事件驱动。Spring Cloud Bus 是一个消息总线,它可以在微服务之间传递消息,可以将所有微服务视为一个整体,向所有微服务广播消息或向指定的微服务发送消息。Spring Cloud Stream 是一个消息驱动的微服务框架,它可以轻松地将消息通道与微服务进行集成。

准备工作

首先,我们需要在 pom.xml 文件中添加以下依赖:


(资料图片)

    org.springframework.cloud    spring-cloud-starter-bus-amqp    org.springframework.cloud    spring-cloud-stream-binder-rabbit

这些依赖将会引入 Spring Cloud Bus 和 Spring Cloud Stream 的相关库,并且使用 RabbitMQ 作为消息代理。如果你想使用其他消息代理,可以根据实际需求进行修改。

创建消息通道

在这个例子中,我们将创建一个名为 myChannel 的消息通道,用于在微服务之间传递消息。在创建消息通道之前,我们需要在 application.yml 文件中添加以下配置:

spring:  cloud:    stream:      bindings:        myChannel:          destination: myChannel

这个配置将创建一个名为 myChannel 的消息通道,并将它绑定到 RabbitMQ 的 myChannel 队列上。现在,我们可以在代码中使用 @Input 和 @Output 注解来定义输入和输出消息通道了。

public interface MyChannel {    String INPUT = "myInput";    String OUTPUT = "myOutput";    @Input(INPUT)    SubscribableChannel input();    @Output(OUTPUT)    MessageChannel output();}

这个接口定义了一个名为 MyChannel 的消息通道,其中包括一个名为 myInput 的输入消息通道和一个名为 myOutput 的输出消息通道。

发布消息

在这个例子中,我们将创建一个名为 MyController 的控制器类,该类将发布一个名为 MyMessage 的消息到 myOutput 消息通道上。

@RestControllerpublic class MyController {    @Autowired    private MessageChannel output;    @PostMapping("/send")    public void sendMessage(@RequestBody MyMessage message) {        output.send(MessageBuilder.withPayload(message).build());    }}

这个控制器类注入了名为 output 的 MessageChannel,用于向 myOutput 消息通道发送消息。在 sendMessage 方法中,我们通过 MessageBuilder 创建一个名为 message 的 MyMessage 消息,然后通过 output.send 方法将这个消息发送到 myOutput 消息通道上。

处理消息

在这个例子中,我们将创建一个名为 MyListener 的监听器类,该类将监听 myInput 消息通道上的消息,并将消息打印到控制台上。

@EnableBinding(MyChannel.class)public class MyListener {    @StreamListener(MyChannel.INPUT)    public void handleMessage(MyMessage message) {        System.out.println("Received message: " + message);    }}

这个监听器类使用 @EnableBinding 注解将 MyChannel 消息通道绑定到 Spring Cloud Stream 上。在 handleMessage 方法中,我们使用 @StreamListener 注解监听 myInput 消息通道上的消息,当有消息到来时,Spring Cloud Stream 将自动将消息转换为 MyMessage 类型,并将其传递给 handleMessage 方法进行处理。在这个例子中,我们只是简单地将消息打印到控制台上,你可以根据实际需求进行修改。

运行测试

现在,我们已经创建了消息通道、发布了消息和处理了消息,我们可以启动应用程序并测试它了。首先,我们需要在终端窗口中启动 RabbitMQ,执行以下命令:

cssCopy codedocker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

这个命令将启动一个名为 rabbitmq 的容器,并将其映射到本地主机的 5672 和 15672 端口上。现在,我们可以启动应用程序并访问 http://localhost:8080/send发送消息了。在控制台中,你应该可以看到类似下面的输出:

Received message: MyMessage{id=1, content="Hello, world!"}

这表明消息已经成功传递到了 myInput 消息通道,并被 MyListener 监听到并处理了。

标签:

国美与阿里云达成战略合作

8月12日,据新华网报道,国美零售与阿里云签订战略合作协议。双方将基于云计算、物联网、数据技术,在零...

2022-08-15 15:07:54

唐山加快验收时间 为建筑工地开复工保驾护航

唐山市住建局坚持两手抓、两手硬,统筹疫情防控和经济发展,想尽一切办法,简化一切手续,实现一路绿灯...

2022-03-19 15:15:42

消费者“身材焦虑” 减肥市场疯狂吸金

三月不减肥,四月徒伤悲。冬去春来之际,为了能穿上各种漂亮轻薄的服装,变得更瘦成为了许多人的追求,...

2022-03-19 15:14:49

2月份秦皇岛新建商品住宅销售价格环比下降0.2% 同比下降4%

3月16日,国家统计局公布2月份70个大中城市商品住宅销售价格变动情况,河北省入统的石家庄、唐山、秦皇...

2022-03-19 15:13:01

深圳技术进出口全年合同数量共1347项 同比增长2.51%

科技部火炬中心近日公布了2021年度全国技术合同交易数据。截至2021年12月31日,全国共登记技术合同67050...

2022-03-19 15:10:37

邢台柏乡:打造羊肚菌产业示范带 引领村级集体经济发展

我们通过打造食用菌产业高标准试验示范园,不仅盘活了闲置土地,还进一步增加了群众和村集体经济收入。...

2022-03-19 15:09:40

胡金秋32分17板 浙江广厦男篮“双杀”稠州金租

CBA浙江德比二番战昨晚开打,再度狭路相逢的浙江广厦男篮与浙江稠州金租男篮表现难分伯仲,比赛直到最后...

2022-03-19 15:07:37

黑龙江新增本土核酸检测初筛阳性人员5例 均在讷河市

  中新网12月13日电 据黑龙江省卫生健康委员会网站消息,2021年12月12日0-24时,黑龙江省无新增确诊...

2021-12-13 08:35:21

“恋爱盲盒”抽的不是爱情,是急功近利的心

  越是急功近利,越是焦虑,对恋爱关系的处理就越可能出现问题。  “玩法”简单,最低只需花费0 99...

2021-12-13 08:35:20

北京12月12日新增2例境外输入确诊病例

  中新网12月13日电 据北京市卫生健康委员会官方微博消息,北京12月12日新增2例境外输入确诊病例。 ...

2021-12-13 08:35:20
x 广告
x 广告

Copyright ©  2015-2022 时代经营网版权所有  备案号:   联系邮箱: 514 676 113@qq.com