0
0

SpringCloud 集成 RocketMQ 示例及配置详解

文章目录

前言

  1. 定义 Spring Cloud Stream 是一个用来为微服务应用构建消息驱动能力的框架。它可以基于 Spring Boot 来创建独立的、可用于生产的 Spring 应用程序。Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，并引入了发布-订阅、消费组、分区这三个核心概念。简单的说，Spring Cloud Stream本质上就是整合了Spring Boot和Spring Integration，实现了一套轻量级的消息驱动的微服务框架。
  2. 抽象模型 我们都知道市面上有很多消息中间件，Sping Cloud Stream 为了可以集成各种各样的中间件，它抽象出了 Binder 的概念,每个消息中间件都需要有对应自己的 Binder。这样它就可以根据不同的 Binder 集成不同的中间件。下图的input和output是channel，Binder则是消息中间件和通道之间的桥梁。在这里插入图片描述
  3. 绑定器 通过使用 Spring Cloud Stream，可以有效简化开发人员对消息中间件的使用复杂度，让系统开发人员可以有更多的精力关注于核心业务逻辑的处理。但是目前 Spring Cloud Stream 只支持 RabbitMQ 和 Kafka 的自动化配置。 Spring Cloud Stream 提供了 Binder (负责与消息中间件进行交互)，我们则通过 inputs 或者 outputs 这样的消息通道与 Binder 进行交互。

Binder 绑定器是 Spring cloud Stream 中一个非常重要的概念，实现了应用程序和消息中间件之间的隔离，同时我们也可以通过应用程序实现，消息中间件之间的通信。在我们的项目的可以继承多种绑定器，我们可以根据不同特性的消息使用不同的消息中间件。Spring Cloud Stream 为我们实现了 RabbitMQ 和Kafka 的绑定器。如果你想使用其他的消息中间件需要自己去实现绑定器接口。

一、SpringCloud 集成 RocketMQ

1. pom 依赖

  1. <!-- rocketmq --><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-stream-rocketmq</artifactId></dependency>

2. yml 配置

  1. spring:cloud:stream:function:definition: producer1;consumer1 # 方法定义（用于定义发送者或消费者方法，多个分号隔开）# 配置消息通道通用属性（适用于所有消息中间件）bindings:producer1-out-0:destination: producer_topic # topic消息主题content-type: application/json # 内容格式# 配置channel消息通道consumer1-in-0:destination: consumer_topic # topic消息主题content-type: application/json # 内容格式group: consumer-group # 消费者组rocketmq:binder:name-server: 127.0.0.1:9876# rocketmq服务地址vipChannelEnabled:true# 是否开启vip通道（兼容老版本使用。多监听一个端口用于接受处理消息，防止端口占用。）# 配置消息通道独特属性（仅适用于rocketmq）bindings:# 配置channel消息通道（生产者：[functionName]-out-[index]，消费者：[functionName]-in-[index]）producer1-out-0:producer:group: consumer-group
  2.               sync:true# 是否开启同步发送consumer1-in-0:consumer:subscription: myTag  # 消费tagdelayLevelWhenNextConsume:-1suspendCurrentQueueTimeMillis:99999999broadcasting:false# 是否使用广播消费，默认为false使用集群消费

3. 操作实体

  1. packagecom.demo.model;importlombok.AllArgsConstructor;importlombok.Data;/**
  2.  * 消息model
  3.  */@Data@AllArgsConstructorpublicclassMsgModel{/**
  4.      * 消息id
  5.      */privateString msgId;/**
  6.      * 消息内容
  7.      */privateString message;}

4. 生产消息

4.1. 自动发送消息

通过 MessageBuilder 自动发送消息。

  1. packagecom.demo;importcom.demo.model.MsgModel;importlombok.extern.slf4j.Slf4j;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.messaging.Message;importorg.springframework.messaging.support.MessageBuilder;importjava.util.function.Supplier;/**
  2.  * 消息生产者类
  3.  */@Configuration@Slf4jpublicclassMyProducer{/**
  4.      * 消息生产者1
  5.      */@BeanpublicSupplier<Message<MsgModel>>producer1(){return()->{MsgModel msgModel =newMsgModel(System.currentTimeMillis(),"测试消息");
  6.             log.info("producer1发送消息："+ msgModel);returnMessageBuilder.withPayload(entity).build();};}}

这种方式定义 suppelier 会 默认1000ms 发送一次记录。可以修改 spring.cloud.stream.poller.fixedDelay 设置延迟毫秒值。

4.2. 手动发送消息

通过 StreamBridge 手动发送消息。

  1. packagecom.demo.controller;importcom.demo.model.MsgModel;importlombok.RequiredArgsConstructor;importlombok.extern.slf4j.Slf4j;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.cloud.stream.function.StreamBridge;importorg.springframework.messaging.support.MessageBuilder;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RestController;/**
  2.  * 消息controller
  3.  */@RestController@RequiredArgsConstructor@RequestMapping("/msg")@Slf4jpublicclassMsgController{privatefinalStreamBridge streamBridge;/**
  4.      * 发送消息
  5.      */@GetMapping("/send")publicvoidsendMsg(){MsgModel msgModel =newMsgModel(System.currentTimeMillis(),"测试消息");
  6.         log.info("producer1发送消息："+ msgModel);
  7.         streamBridge.send("producer1-out-0",MessageBuilder.withPayload(entity).setHeader("MyHearder","这是一个请求头").build());}}

5. 消费消息

  1. packagecom.demo;importcom.demo.model.MsgModel;importlombok.extern.slf4j.Slf4j;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.messaging.Message;importorg.springframework.messaging.MessageHeaders;importjava.util.function.Consumer;/**
  2.  * 消息消费者类
  3.  */@Configuration@Slf4jpublicclassReceiveMQ{/**
  4.      * 消息消费者1
  5.      */@BeanpublicConsumer<Message<MsgModel>>consumer1(){return(message)->{MessageHeaders headers = message.getHeaders();MsgModel msgModel = message.getPayload();
  6.             log.info("consumer1接收消息，消息头："+ headers.get("MyHeader"));
  7.             log.info("consumer1接收消息，消息内容："+ msgModel);};}}

二、配置解析

1. spring.cloud.function.definition 方法定义

进行生产者或消费者方法定义，在 rocketmq 初始时会加载这些方法以创建生产者或消费者列表。

不管是创建 Consumer 还是 Supplier 或者是 Function Stream 都会将其方法名称进行一个 topic 拆封和绑定。假设创建了一个 Consumer< String > myTopic 的方法，Stream 会将其 拆分成 In 和 out 两个通道：

  • 输入通道（消费者）： [functionName]-in-[index] consumer1-in-0
  • 输出通道（生产者）： [functionName]-out-[index] producer1-out-0

注意：这里的 functionName 需要和生产者或消费者方法名称以及 spring.cloud.function.definition 下的名称保持一致。

  1. spring:cloud:stream:function:definition: producer1;consumer1 # 方法定义（用于定义发送者或消费者方法，多个分号隔开）

2. spring.cloud.stream.binders 公共服务配置

配置消息中间件公共的连接信息。

配置消息中间件公共的主机名、端口、用户名和密码等信息。可以配置多个binder，每个binder对应一个消息中间件。

当需要定义多个消息中间件时，使用 stream.binders 定义。
当只定义一个中间件时，如 RocketMQ，请直接使用 stream.rocketmq.binder 定义。

  1. spring:cloud:stream:# 公共服务配置# 当需要定义多个消息中间件时，使用 stream.binders 定义。# 当只定义一个中间件时，如 RocketMQ，请直接使用 stream.rocketmq.binder 定义。binders:rocket1:type: rocket # 消息中间件类型environment:spring:rocketmq:host: 127.0.0.1 # IP地址port:9876# 端口username: guest    # 账号password: guest # 密码

3. spring.cloud.stream.bindings 公共消息通道配置

配置消息通道公共的配置信息。

配置消息通道公共的名称、目标（对应消息中间件中的队列或主题名）、分区、消息内容类型等信息。

当需要定义多个消息中间件时，使用 stream.bindings 定义。
当只定义一个中间件时，如 RocketMQ，请直接使用 stream.rocketmq.bindings 定义。

  1. spring:cloud:stream:# 公共消息通道配置# 当需要定义多个消息中间件时，使用 stream.bindings 定义。# 当只定义一个中间件时，如 RocketMQ，请直接使用 stream.rocketmq.bindings 定义。# 定义生产者通道：[functionName]-out-[index]、定义消费者通道：[functionName]-in-[index]。bindings:# 生产者通道producer1-out-0:destination: producer1_topic # 消息队列（topic主题）content-type: application/json # 消息内容格式binder: rocket1 # 绑定的消息中间件# 消费者通道consumer1-in-0:destination: consumer1_topic # 消息队列（topic主题）content-type: application/json # 消息内容格式group: consumer1-group # 消费组（在一个组内，一条消息只能被消费一次）binder: rocket1 # 绑定的消息中间件

2. spring.cloud.stream.rocketmq.binder RocketMQ 服务配置

配置 RocketMQ 消息中间件的连接信息。

配置 RocketMQ 消息中间件的主机名、端口、用户名和密码等信息。

  1. spring:cloud:stream:rocketmq:# 服务配置binder:name-server: 127.0.0.1:9876# 服务地址vipChannelEnabled:true# 是否开启vip通道（兼容老版本使用。多监听一个端口用于接受处理消息，防止端口占用）

3. spring.cloud.stream.rocketmq.bindings RocketMQ 消息通道配置

配置 RocketMQ 消息通道的配置信息。

配置 RocketMQ 通道的名称、目标（对应消息中间件中的队列或主题名）、分区、消息内容类型等信息。

  1. spring:cloud:stream:rocketmq:bindings:# 生产者通道（可定义通道的生产者producer）producer1-out-0:producer:group: producer1-group # 生产组（多个生产者实例，当一个实例挂掉时，使用另一个实例执行消息的提交或回滚）sync:true# 是否同步发送（发送消息后同步等待，直至发送结果返回）# 消费者通道（可定义通道的消费者consumer）consumer1-in-0:consumer:group:# 消费组（多个消费者实例，用于负载均衡和容错。每个实例订阅的主题必须相同）subscription: myTag  # 消费tag（多个tag用||隔开）broadcasting:false# 是否使用广播消费，默认为false使用集群消费maxReconsumeTimes:16# 最大重试次数（-1 – 16，默认16）delayLevelWhenNextConsume:-1# 重试消费的延迟等级（-1不重试；默认0，默认等级3）suspendCurrentQueueTimeMillis:99999999# 重试时间间隔ms（默认1000）

三、RocketMQ 特性

1. 消息回滚

当消费消息发生异常时，会自动进行消息回滚，并且再进行最多两次消息重发，当第二次消息重发依旧发生异常时，才会丢弃消息，并打印异常日志，否则异常被 RocketMQ 拦截处理，不会打印异常日志。

  1. packagecom.demo;importcom.demo.model.MsgModel;importlombok.extern.slf4j.Slf4j;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.messaging.Message;importorg.springframework.messaging.MessageHeaders;importjava.util.function.Consumer;/**
  2.  * 消息消费者类
  3.  */@Configuration@Slf4jpublicclassReceiveMQ{/**
  4.      * 消息消费者1
  5.      */@BeanpublicConsumer<Message<MsgModel>>consumer1(){return(message)->{
  6.             log.info("consumer1接收消息，消息内容："+ msgModel);thrownewRuntimeException("消费消息异常");};}}

运行结果

  1. 2024-07-1717:12:20[ConsumeMessageThread_1]INFO[com.demo.consumer.ReceiveMQ:19]- 消息内容："测试消息"2024-07-1717:12:21[ConsumeMessageThread_1]INFO[com.demo.consumer.ReceiveMQ:19]- 消息内容："测试消息"2024-07-1717:12:23[ConsumeMessageThread_1]INFO[com.demo.consumer.ReceiveMQ:19]- 消息内容："测试消息"2024-07-1717:12:23[ConsumeMessageThread_1]ERROR[com.demo.consumer.ReceiveMQ:20]- 消息消费异常
标签： rocketmq spring cloud
本文转载自: https://blog.csdn.net/demo_yo/article/details/138617904
版权归原作者 Whitemeen太白 所有， 如有侵权，请联系我们删除。
登录后发布评论

“SpringCloud 集成 RocketMQ 示例及配置详解”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？