spring boot 集成rocketMq + 基本使用

news/2024/5/18 22:02:47

1. RocketMq基本概念

1. NameServer
每个NameServer结点之间是相互独立,彼此没有任何信息交互
启动NameServer。NameServer启动后监听端口,等待Broker、Producer、Consumer连接,
相当于一个路由控制中心。主要是用来保存topic路由信息,管理Broker
2. Broker
消息存储和中转角色,负责存储和转发消息
在启动时会向NameServer进行注册并且定时发送心跳包。心跳包中包含当前 Broker 信息
以及存储所有 Topic 信息。注册成功后,NameServer 集群中就有 Topic跟Broker 的映射关系。
3. topic : 一个消息的集合的名字
创建 Topic 时需要指定该 Topic 要存储在哪些 Broker 上,也可以在发送消息时自动创建Topic。
4. 生产者
生产者发送消息。启动时先从 NameServer 集群中的其中一台拉取到路由表,缓存到本地,
并从 NameServer 中获取当前发送的 Topic存在于哪些 Broker 上,
轮询从队列列表中选择一个队列(默认轮询)
5. 消费者
消费者跟其中一台NameServer建立连接,获取当前订阅Topic存在哪些Broker上,
然后直接跟Broker建立连接通道,然后开始消费消息

2. maven 引入starter

<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.2</version>
</dependency>

3.yml配置

3.1 生产者yml 配置

rocketmq:name-server: 127.0.0.1:9876producer:group: my-group# 发送消息超时时间send-message-timeout: 5000# 发送消息失败重试次数retry-times-when-send-failed: 2retry-times-when-send-async-failed: 3 # 异步发送消息失败重试次数,默认2

3.2 消费者yml 配置

rocketmq:name-server: 127.0.0.1:9876consumer:topic: topic_testgroup: consumer_my-group

4.生产者发送消息

4.1 一般消息

@Resourceprivate RocketMQTemplate rocketMQTemplate;/***  一般消息* Topic 与 Tag 都是业务上用来归类的标识,区别在于 Topic 是一级分类,而 Tag 可以理解为是二级分类。* 使用 Tag 可以实现对 Topic 中的消息进行过滤。* **/@GetMapping("/send")public String send(){rocketMQTemplate.convertAndSend("topic_test", "Hello, World!");rocketMQTemplate.convertAndSend("topic_test:tagB","Hello, World222--tagB");return "rocketMq普通消息发送完成";}

4.2 顺序消息

/** 支持消费者按照发送消息的先后顺序获取消息 */@GetMapping("/send/orderly")public String sendOrder(){//发送顺序消息,参数:topic,消息,hashkey,相同hashkey发送至同一个队列rocketMQTemplate.syncSendOrderly("topic_test:tagA", MessageBuilder.withPayload("消息编号" + 1).build(),"queue");rocketMQTemplate.syncSendOrderly("topic_test:tagA", MessageBuilder.withPayload("消息编号" + 2).build(),"queue");return "rocketMq顺序-消息发送成功";}

4.3 同步消息

@GetMapping("/send/sync")public String sendMsg() {String message = "我是同步消息:" + LocalDateTime.now();SendResult result = rocketMQTemplate.syncSend("topic_test:tagA", MessageBuilder.withPayload(message).build());log.info("同步-消息发送成功:" + LocalDateTime.now());return "rocketMq 同步-消息发送成功:" + result.getSendStatus();}

4.4 异步消息

/** 发送异步消息 */@GetMapping("/send/async")public String asyncSendMsg(){String message = "我是异步消息:" + LocalDateTime.now();rocketMQTemplate.asyncSend("topic_test:tagA",message,new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {log.info("发送成功 (后执行),SendStatus = {}",sendResult.getSendStatus());}@Overridepublic void onException(Throwable throwable) {log.info("发送失败 (后执行)");}});return "rocketMq 异步-消息发送成功:" + LocalDateTime.now();}

 4.5 单向消息:一般用来发送日志等不重要的消息

@GetMapping("/send/oneWay")public String sendOneWayMessage() {String message =  "我是单向消息:"+LocalDateTime.now();this.rocketMQTemplate.sendOneWay("topic_test:tagA", message);log.info("单向发送消息完成:message = {}", message);return "rocketMq 单向-消息发送成功:" + LocalDateTime.now();}

 

4.6 延时消息

/** 延时消息 */@GetMapping("/sendDelay")public String sendDelay(){String message = "我是延时消息:" + LocalDateTime.now();// 第四个参数为延时级别,分为1-18:1、5、10、30、1m、2m、3m、...10m、20m、30m、1h、2hrocketMQTemplate.syncSend("topic_test:tagC", MessageBuilder.withPayload(message).build(), 3000, 2);return "rocketMq延时-消息发送成功";}

4.7 事务消息

4.7.1 事务消息发送代码

/** 事务消息 */@GetMapping("/send/transaction/{id}")public void sendTransactionMessage(@PathVariable("id") Integer id){//发送事务消息:采用的是sendMessageInTransaction方法,返回结果为TransactionSendResult对象,该对象中包含了事务发送的状态、本地事务执行的状态等//参数一:topic;参数二:消息// 事务idString[] tags = {"tagA", "tagB", "tagC"};int i = id%3;String transactionId = UUID.randomUUID().toString();String message = "我是事务消息:" + LocalDateTime.now();TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction("topic_test:" + tags[i], MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.TRANSACTION_ID,transactionId).build(),// 给本地事务的参数2);//发送状态String sendStatus = result.getSendStatus().name();//本地事务执行状态String localState = result.getLocalTransactionState().name();log.info("发送状态:"+sendStatus+";本地事务执行状态"+localState);}

4.7.2 继承 RocketMQLocalTransactionListener

@Slf4j
@RocketMQTransactionListener
public class MyTransactionListener implements RocketMQLocalTransactionListener {@Overridepublic RocketMQLocalTransactionState executeLocalTransaction(org.springframework.messaging.Message message, Object o) {MessageHeaders headers = message.getHeaders();//获取事务IDString transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);log.info("执行本地事务 ,transactionId is {}, orderId is {}",transactionId, message.getHeaders().get("rocketmq_TAGS"));try{//模拟网络波动Thread.sleep(3000);/**** 首先发送一个半消息(half message),这个消息不会立即投递给消费者;然后执行本地事务(比如数据库操作)。* 根据本地事务的执行结果,决定是提交(commit)还是回滚(rollback)这个消息。* 如果本地事务成功,消息会被提交并发送给消费者;* 如果失败,消息会被回滚,消费者不会接收到这个消息*/}catch (Exception e){return RocketMQLocalTransactionState.ROLLBACK;}// 执行本地事务String tag = String.valueOf(message.getHeaders().get("rocketmq_TAGS"));if (StringUtils.equals("tagA", tag)){//这里只讲TAGA消息提交,状态为可执行return RocketMQLocalTransactionState.COMMIT;}else if (StringUtils.equals("tagB", tag)) {return RocketMQLocalTransactionState.ROLLBACK;} else if (StringUtils.equals("tagC",tag)) {return RocketMQLocalTransactionState.UNKNOWN;}log.info("事务提交,消息正常处理: " + LocalDateTime.now());//执行成功,可以提交事务return RocketMQLocalTransactionState.COMMIT;}@Overridepublic RocketMQLocalTransactionState checkLocalTransaction(org.springframework.messaging.Message message) {MessageHeaders headers = message.getHeaders();//获取事务IDString transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);log.info(transactionId + ",消息回查"+ LocalDateTime.now());return RocketMQLocalTransactionState.ROLLBACK;}
}

tagA、tagB、tagC 三种事务消息,只有Commit的才能发送到broker 

 

 5. 消费端

/*** topic指定消费的主题,consumerGroup指定消费组,* 一个主题可以有多个消费者组,一个消息可以被多个不同的组的消费者都消费*  2.实现RocketMQListener接口*  如果想拿到消息的其他参数可以写成MessageExt*  selectorExpression = "tagA || tagB" 指定tag 的消费*/
@Service
@Slf4j
@RocketMQMessageListener(topic = "${rocketmq.consumer.topic}", consumerGroup = "${rocketmq.consumer.group}")
public class RocketMqConsumer implements RocketMQListener<String>{@Overridepublic void onMessage(String s) {log.info("topic_test: 所有的收到消息:"+s);}}

6.广播消费模式

生产端是一样的,但是消费端需要增加一个参数

messageModel:设置消费模式,取值范围CLUSTERING(集群消费)、BROADCASTING(广播消费)
@Service
@Slf4j
@RocketMQMessageListener(topic = "${rocketmq.consumer.topic}", consumerGroup = "${rocketmq.consumer.group}", messageModel = MessageModel.BROADCASTING)
public class RocketMqConsumer implements RocketMQListener<String>{@Overridepublic void onMessage(String s) {log.info("consumer2---topic_test: 所有的收到消息:"+s);}}// 第2个消费者类,他们都是一样的代码,
//为了表示广播,就是一个消息,会被这两个消费者消费@Service
@Slf4j
@RocketMQMessageListener(topic = "${rocketmq.consumer.topic}", consumerGroup = "${rocketmq.consumer.group}", messageModel = MessageModel.BROADCASTING)
public class RocketMqConsumer implements RocketMQListener<String>{@Overridepublic void onMessage(String s) {log.info("consumer1--topic_test: 所有的收到消息:"+s);}}

7.其他

RocketMQ 通过消费者组(Consumer Group)来维护不同消费者的消费进度。每个消费者组都有一个消费进度(offset),用于标记该组下的消费者在某个主题(Topic)和队列(Queue)上已经消费到的位置。所以:不同的消费者组会被视为不同的消费者


https://dhexx.cn/news/show-5318120.html

相关文章

2024华中杯A题完整1-3问py代码+完整思路16页+后续参考论文

A题太阳能路灯光伏板朝向问题 &#xff08;完整版资料文末获取&#xff09; 第1小问&#xff1a;计算每月15日的太阳直射强度和总能量 1. 理解太阳直射辐射和光伏板的关系**&#xff1a;光伏板接收太阳辐射并转化为电能&#xff0c;直射辐射对光伏板的效率影响最大。 2. 收集…

openEuler-23.03下载

下载地址&#xff1a;openEuler下载 | 欧拉系统ISO镜像 | openEuler社区官网 下载版本&#xff1a;openEuler-23.03-x86_64-dvd.iso

物联网网络传输,STM32和ESP32

2024.4.17 今天主要是记录一下我这个STM32F407和esp8266组成的互联网板子的一些例程运行心得&#xff0c; 下面这个例程是STM32发AT指令给esp8266&#xff0c;使得esp8266作为tcp client连接到我的手机热点&#xff0c;然后我用电脑连接wifi&#xff0c;打开tcp server&#x…

【UE 材质】序列图动画、波动效果、颜色切换效果

目录 一、序列图动画 方法1 方法2 二、波动效果 三、颜色切换效果 一、序列图动画 方法1 新建一个材质&#xff0c;在材质中添加一个纹理采样节点&#xff0c;使用纹理“T_Explosion_SubUV” 由于该序列图是6*6的&#xff0c;因此我们可以通过将纹理坐标节点乘1/6得到…

AWB学习记录

主要参考食鱼者博客&#xff1a;https://blog.csdn.net/wtzhu_13/article/details/119301096&#xff0c;以及相关的论文&#xff0c;感谢食鱼者老师整理分享。 灰度世界和完全反射 灰度世界法和完全反射法分别是基于(Rmean, Gmean, Bmean)和(Rmax, Gmax, Bmax)来进行白平衡校…

2024 MathorCup C 题 物流网络分拣中心货量预测及人员排班

一、问题重述 电商物流网络在订单履约中由多个环节组成&#xff0c;图1是一个简化的物流网络示意图。其中&#xff0c;分拣中心作为网络的中间环节&#xff0c;需要将包裹按照不同流向进行分拣并发往下一个场地&#xff0c;最终使包裹到达消费者手中。分拣中心管理效率的提升&…

1024同步写作助手操作说明文档

开发这个浏览器插件的主要目的是为了解决多个平台发布文章不同步问题&#xff0c;比如我在csdn上写了一篇文章发布了&#xff0c;但是我还有掘金/知乎/博客园/今日头条/微信公众号等平台啊&#xff0c;我也想把这篇文章同步到其他几个平台&#xff0c;怎么办&#xff1f;难道让…

邮箱群组是什么?怎么创建邮箱群组?

在我们群发邮件时&#xff0c;可能会遇到这样的状况&#xff0c;一个个输入邮箱地址效率很低&#xff0c;而且很容易就漏发。而对于一个企业来说&#xff0c;如果出现这样的问题&#xff0c;很有可能会影响公司的业务进展和团队协作。这个时候我们就需要邮箱群组这个功能&#…

SQL Server Management Studio 显示行号

前言 在使用 SQL Server Management Studio (SSMS) 进行数据库管理和查询时&#xff0c;能够看到代码的行号是非常有用的。这可以帮助您更容易地定位代码错误、讨论特定的代码行&#xff0c;或者在执行长查询时快速找到特定行。在本文中&#xff0c;我将向您展示如何在 SSMS 中…

基础算法之二分算法

前言 本次博客&#xff0c;将要介绍二分算法的基本原理以及如何使用&#xff0c;深入浅出 二分可以针对整型以及浮点型接下来对其讲解希望对小白有所帮助吧 整型的二分法 一般要在一个数组中猜出一个数是否存在我们可以遍历一遍整个数组&#xff0c;判断是否存在&#xff0…