RocketMQ学习笔记

在这里插入图片描述

一、下载安装

1.1 官网下载

官网地址:

http://rocketmq.apache.org

下载地址：

https://rocketmq.apache.org/download/

控制台下载地址:

https://rocketmq.apache.org/zh/download/#rocketmq-dashboard

1.2 集群搭建（2m-2s-async模式）

设计架构如下：

服务器	部署节点	部署broker
192.168.10.131	namesrv、broker	broker-a-m、broker-b-s
192.168.10.132	namesrv、broker	broker-b-m、broker-a-s
192.168.10.133	namesrv	仅部署namesrv

1.2.1 解压、修改内存、新建文件存储目录

rocketMq我放在/usr/local/rocketmq目录,进入目录执行
解压：unzip rocketmq-all-4.9.0-bin-release.zip

3台服务器都修改内存大小：修改runserver.sh和runbroker.sh的JAVA_OPT的大小
cd rocketmq-all-4.9.0-bin-release/bin/

vim runserver.sh

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=160m"
1

vim runbroker.sh

 JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"
1

在3台服务器的/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release新建目录store
在store下新建master slave两个目录分别存放主、从的相应配置
分别在master slave里面新建abort checkpoint commitlog consumequeue index pathroot目录
mkdir store
cd store/
mkdir master slave
cd master/
mkdir pathroot commitlog consumequeue index checkpoint abort
cd slave/
mkdir pathroot commitlog consumequeue index checkpoint abort

1.2.2 配置192.168.10.131

修改配置broker-a-m
进入192.168.10.131的/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/conf/2m-2s-async
vim broker-a.properties

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，名字可重复,为了管理,每个master起一个名字,他的slave同他,eg:Amaster叫broker-a,他的slave也叫broker-a
brokerName=broker-a-m
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=192.168.10.131:9876;192.168.10.132:9876;192.168.10.133:9876

#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口,
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/master/pathroot
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/master/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/master/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/master/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/master/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/master/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=ASYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

修改配置broker-b-s
进入192.168.10.131的/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/conf/2m-2s-async
vim broker-b-s.properties

#所属集群名字
brokerClusterName=rocketmq-cluster
brokerName=broker-b-s
#broker名字，名字可重复,为了管理,每个master起一个名字,他的slave同他,eg:Amaster叫broker-b,他的slave也叫broker-b
#0 表示 Master，>0 表示 Slave
brokerId=1
#nameServer地址，分号分割
namesrvAddr=192.168.10.131:9876;192.168.10.132:9876;192.168.10.133:9876

#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口,
listenPort=10920
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/slave/pathroot
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/slave/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/slave/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/slave/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/slave/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/slave/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=SLAVE
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

1.2.3 配置192.168.10.132

修改配置broker-b-m
进入192.168.10.132的/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/conf/2m-2s-async
vim broker-b.properties

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，名字可重复,为了管理,每个master起一个名字,他的slave同他,eg:Amaster叫broker-a,他的slave也叫broker-a
brokerName=broker-b-m
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=192.168.10.131:9876;192.168.10.132:9876;192.168.10.133:9876

#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口,
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/master/pathroot
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/master/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/master/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/master/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/master/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/master/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=ASYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

修改配置broker-a-s
进入192.168.10.132的/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/conf/2m-2s-async
vim broker-a-s.properties

#所属集群名字
brokerClusterName=rocketmq-cluster
brokerName=broker-a-s
#broker名字，名字可重复,为了管理,每个master起一个名字,他的slave同他,eg:Amaster叫broker-b,他的slave也叫broker-b
#0 表示 Master，>0 表示 Slave
brokerId=1
#nameServer地址，分号分割
namesrvAddr=192.168.10.131:9876;192.168.10.132:9876;192.168.10.133:9876

#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口,
listenPort=10920
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/slave/pathroot
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/slave/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/slave/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/slave/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/slave/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/store/slave/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=SLAVE
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

1.2.4 配置192.168.10.133

只做namesrv不启动broker，所以无需配置

1.3 启动

1.3.1 启动namesrv

分别在3台服务器的/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release
启动： nohup sh bin/mqnamesrv &
查看日志： tail -f ~/logs/rocketmqlogs/namesrv.log

1.3.2 启动broker

192.168.10.131

启动master： nohup sh bin/mqbroker -c conf/2m-2s-async/broker-a.properties &
启动slave: nohup sh bin/mqbroker -c conf/2m-2s-async/broker-b-s.properties &
查看日志：tail -f ~/logs/rocketmqlogs/broker.log

192.168.10.132

启动master：nohup sh bin/mqbroker -c conf/2m-2s-async/broker-b.properties &
启动slave: nohup sh bin/mqbroker -c conf/2m-2s-async/broker-a-s.properties &
查看日志：tail -f ~/logs/rocketmqlogs/broker.log

1.4 监控qocketmq

下载控制台（2021版本）：https://rocketmq.apache.org/zh/download/#rocketmq-dashboard拖到页面最底部，有个 RocketMQ Dashboard，就可以下载了
使用IDEA打开后：
修改服务端口：server.port=10088
修改连接地址：
rocketmq.config.namesrvAddr=192.168.10.131:9876;192.168.10.132:9876;192.168.10.133:9876

访问即可看到rocketMq集群
启动：记得开防火墙（一台机器上既有master又有slave，master用的10911、slave改成了10920，都要开墙）

9876 是nameserver中的端口,链接nameserver就靠这个端口 (必开墙)
10911 listenPort参数是broker的监听端口号，是remotingServer服务组件使用，作为对Producer和Consumer提供服务的端口号 (必开墙)
10909 fastListenPort参数是fastRemotingServer服务组件使用，默认为listenPort - 2，可以通过配置文件修改。打开broker-x.conf，修改或增加fastListenPort参数
10912 haListenPort参数是HAService服务组件使用，用于Broker的主从同步，默认为listenPort -1，可以通过配置文件修改。打开broker-x.conf，修改或增加haListenPort参数：

二、基本概念

1.Name Server
功能介绍 NameServer是一个Broker与Topic路由的注册中心，支持Broker的动态注册与发现。主要包括两个功能：
① Broker管理：接受Broker集群的注册信息并且保存下来作为路由信息的基本数据；提供心跳检测机制，检查Broker是否还存活。
② 路由信息管理：每个NameServer中都保存着Broker集群的整个路由信息和用于客户端查询的队列信息。Producer和Conumser通过NameServer可以获取整个Broker集群的路由信息，从而进行消息的投递和消费。

2.Broker
Broker充当着消息中转角色，负责存储消息、转发消息。Broker在RocketMQ系统中负责接收并存储从生产者发送来的消息，同时为消费者的拉取请求作准备。Broker同时也存储着消息相关的元数据，包括消费者组消费进度偏移offset、主题、队列等

3. 消息（Message）
消息是指，消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。

4.主题（Topic）
Topic表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是RocketMQ进行消息订阅的基本单位。 topic:message 1:n message:topic 1:1 一个生产者可以同时发送多种Topic的消息；而一个消费者只对某种特定的Topic感兴趣，即只可以订阅和消费一种Topic的消息。 producer:topic 1:n consumer:topic 1:1 3 标签（Tag）为消息设置的标签，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。 Topic是消息的一级分类，Tag是消息的二级分类。

5.队列（Queue）
存储消息的物理实体。一个Topic中可以包含多个Queue，每个Queue中存放的就是该Topic的消息。一个Topic的Queue也被称为一个Topic中消息的分区（Partition）。一个Topic的Queue中的消息只能被一个消费者组中的一个消费者消费。一个Queue中的消息不允许同一个消费者组中的多个消费者同时消费。

6. 消息标识（MessageId/Key）
RocketMQ中每个消息拥有唯一的MessageId，且可以携带具有业务标识的Key，以方便对消息的查询。不过需要注意的是，MessageId有两个：在生产者send()消息时会自动生成一个MessageId（msgId)，当消息到达Broker后，Broker也会自动生成一个MessageId(offsetMsgId)。msgId、offsetMsgId与key都称为消息标识。 msgId：由producer端生成，其生成规则为： producerIp + 进程pid + MessageClientIDSetter类的ClassLoader的hashCode + 当前时间 +AutomicInteger自增计数器 offsetMsgId：由broker端生成，其生成规则为：brokerIp + 物理分区的offset（Queue中的偏移量） key：由用户指定的业务相关的唯一标识

三、JAVA基本使用

投递方式：

1.单向 生产者投递消息到mq中，不需要返回结果。
优点：延迟概率比较低
缺点：丢失消息数据
投递消息过程比较耗时时间5毫秒

 @GetMapping("/sendMsg")
public String sendMsg() {
    MsgEntity msg = new MsgEntity("mayikt" + UUID.randomUUID().toString(), 1234);
    rocketMQTemplate.convertAndSend(RocketMQConfig.TOPIC_NAME, msg);
    return "投递消息 => " + msg.toString() + " => 成功";
}
1
2
3
4
5
6

2.异步(使用最多) 生产者投递消息到mq中，使用回调形式返回。投递消息过程比较耗时时间5毫秒补偿----

  @GetMapping("async")
public void async() {
    MsgEntity msg = new MsgEntity("mayikt" + UUID.randomUUID().toString(), 1234);
    log.info(">msg:<<" + msg);
    rocketMQTemplate.asyncSend(RocketMQConfig.TOPIC_NAME, msg.toString(), new SendCallback() {
        @Override
        public void onSuccess(SendResult var1) {
            log.info("异步发送成功{}", var1);
        }

        @Override
        public void onException(Throwable var1) {
            log.info("异步发送失败{}", var1);
        }
    });
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

3.同步 生产者投递消息到mq中，采用同步的形式获取到返回消息是否有投递成功的结果，导致接口延迟概率比较大。投递消息过程比较耗时时间10毫秒

@GetMapping("/sync")
public void sync() {
    MsgEntity msg = new MsgEntity("mayikt" + UUID.randomUUID().toString(), 1234);
    SendResult sendResult = rocketMQTemplate.syncSend(RocketMQConfig.TOPIC_NAME, msg);
    log.info("同步发送字符串{}, 发送结果{}", msg.toString(), sendResult);
}
1
2
3
4
5
6

使用实例

1.springMVC集成RocketMq

导入包版本与rocketMQ相匹配

 <dependency>
            <groupId>org.apache.rocketmqgroupId>
            <artifactId>rocketmq-clientartifactId>
            <version>4.9.0version>
        dependency>
1
2
3
4
5

1.1 生产者

package com.pingan.esbx.cassandra.config.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.*;

/**
 * @author Admin
 * @version 1.0
 * @date 2023-10-7 21:45
 **/
@Slf4j
@Service
public class RocketMqProServiceImpl implements RocketMqProService {
    @Value("${rockermq.host}")
    private String host;
    @Value("${rockermq.tag}")
    private String tag;
    @Value("${rockermq.group}")
    private String group;
    @Value("${rockermq.topic}")
    private String topic;

    private DefaultMQProducer producer;
    private ExecutorService threadPool = null;

    /**
     * 初始化执行
     */
    @PostConstruct
    public void init() {
        threadPool = new ThreadPoolExecutor(
                2,
                Runtime.getRuntime().availableProcessors(),
                2L,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(3000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy());

        producer = new DefaultMQProducer(group);
        producer.setNamesrvAddr(host);
        producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
        try {
            producer.start();
        } catch (MQClientException e) {
           log.error("RocketMq 初始化失败，"+e.getMessage(),e);
        }
    }

    @Override
    public void sendMsgToRocketMq(String data) {
        threadPool.execute(new RocketMqProducerPool(data));
    }

    /**
     * 多线程执行发送，让消息发送与主线程异步解耦
     */
    private class RocketMqProducerPool implements Runnable {
        private String data;

        public RocketMqProducerPool(String data) {
            this.data = data;
        }

        @Override
        public void run() {
            try {
                byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET);
                String keys = "ESG_WARN_MAIL:" + System.currentTimeMillis();
                Message msg = new Message(topic, tag, keys, messageBody);

                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        log.info("消息发送成功");
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        log.info("消息发送失败" + throwable);
                    }
                });

            } catch (Exception e) {
                log.error("消息发送异常:" + e.getMessage(), e);
            }
        }

        /**
         * 被 @PreDestroy 注解修饰的方法会在服务器卸载 Servlet 的时候运行，并且只会被服务器调用一次，类似于 Servlet 的 destroy() 方法。
         * 被 @PreDestroy 注解修饰的方法会在 destroy() 销毁方法之前执行
         */
        @PreDestroy
        public void stop() {
            if (producer != null) {
                producer.shutdown();
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

说明：①RocketMqProService 仅仅是一个interface，方便别的地方调用。②rockermq.host等rocketMQ链接信息配在配置文件中③多线程发送消息，异步解耦

1.2 消费者

package com.pingan.esbx.cassandra.config.rocketmq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author Admin
 * @version 1.0
 * @date 2023-10-14 16:52
 **/
@Component
public class RocketMqConsumerService implements MessageListenerConcurrently {

    private static final Logger log = LoggerFactory.getLogger(RocketMqConsumerService.class);

    @Value("${rockermq.host}")
    private String host;
    @Value("${rockermq.tag}")
    private String tag;
    @Value("${rockermq.group}")
    private String group;
    @Value("${rockermq.topic}")
    private String topic;

    private ExecutorService threadPool = null;
    private static DefaultMQPushConsumer consumer = null;

    /**
     * 初始化
     */
    @PostConstruct
    public void init() {
        try {
            threadPool = new ThreadPoolExecutor(
                    2,
                    Runtime.getRuntime().availableProcessors(),
                    2L,
                    TimeUnit.SECONDS,
                    new LinkedBlockingDeque<>(3000),
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.CallerRunsPolicy());


            consumer = new DefaultMQPushConsumer(group);
            consumer.setNamesrvAddr(host);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.setMessageModel(MessageModel.CLUSTERING);
            consumer.subscribe(topic, "*");
            consumer.registerMessageListener(this);
            consumer.start();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 多线程处理接收到的消息
     * @param msgs
     * @param context
     * @return
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

        try {
            for (int index = 0; index < msgs.size(); index++) {
                MessageExt msg = msgs.get(index);
                threadPool.execute(new RocketMqConsumerPool(msg));
                log.info(Thread.currentThread().getName() + ":主线程执行了");
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            /**
             * 集群消费方式下，消息消费失败后若希望消费重试，则需要在消息监听器接口的实现中明确进行如下三
             * 种方式之一的配置：
             * 方式1：返回ConsumeConcurrentlyStatus.RECONSUME_LATER（推荐）
             * 方式2：返回Null
             * 方式3：抛出异常
             */
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        //返回状态:消费成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    private class RocketMqConsumerPool implements Runnable {
        private MessageExt msg;

        public RocketMqConsumerPool(MessageExt msg) {
            this.msg = msg;
        }
        @Override
        public void run() {
            try {
                String messageBody = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                if (msg.getTopic().equalsIgnoreCase("test-topic")) {
                    if (msg.getTags().equalsIgnoreCase("ESG_WARN_ALARM")) {
                        dealTestTopic(messageBody);
                    }
                } else if (msg.getTopic().equalsIgnoreCase("ESG_WARN_ALARM_TOPIC")) {
                    if (msg.getTags().equalsIgnoreCase("ESG_WARN_ALARM")) {
                        dealEsgWarn(messageBody);
                    }
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    private void dealTestTopic(String messageBody) {
        log.info(Thread.currentThread().getName() + ":子线程执行了");
        log.info(messageBody);
    }

    private void dealEsgWarn(String messageBody) {
        log.info(Thread.currentThread().getName() + ":子线程执行了");
        JSONObject jsonObject = JSONObject.parseObject(messageBody);

        log.info(messageBody);
    }

    @PreDestroy
    public void stop() {
        if (consumer != null) {
            consumer.shutdown();
            log.error("消费者关闭");
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

2.springboot集成RocketMq

引包：（这个包跟RocketMQ版本没有对应起来）

  <dependency>
        <groupId>org.apache.rocketmqgroupId>
        <artifactId>rocketmq-spring-boot-starterartifactId>
        <version>2.2.3version>
    dependency>
1
2
3
4
5

yml配置：

rocketmq:
    name-server: 192.168.10.131:9876;192.168.10.132:9876;192.168.10.133:9876
    producer:
        group: ESG_WARN_ALARM_GROUP
1
2
3
4

2.1 生产者

package com.atpingan.rocketunion.service.impl;

import com.atpingan.rocketunion.service.ProducerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.*;

/**
 * @author Admin
 * @version 1.0
 * @date 2023-10-14 21:24
 **/
@Service
@Slf4j
public class ProducerImplService implements ProducerService {


    @Resource
    private RocketMQTemplate rocketMQTemplate;
    private ExecutorService threadPool = null;
    /**
     * 初始化执行
     */
    @PostConstruct
    public void init() {
        threadPool = new ThreadPoolExecutor(
                2,
                Runtime.getRuntime().availableProcessors(),
                2L,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(3000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    @Override
    public void sendMessage(String data) {
        //多线程异步发送消息
        threadPool.execute(new RocketMqProducerPool(data));

    }
    private class RocketMqProducerPool implements Runnable {
        private String data;

        public RocketMqProducerPool(String data) {
            this.data = data;
        }
        @Override
        public void run() {
            /**
             * 参数1: TOPIC:TAG
             * 参数2: 数据
             * 参数3: 发送后的回调
             * 注意这里设置tag的方式是在topic后面,即 TOPIC:TAG
             */
            rocketMQTemplate.asyncSend("ESG_WARN_ALARM_TOPIC:ESG_WARN_ALARM", data, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("send success");
                }
                @Override
                public void onException(Throwable throwable) {
                    log.error("send fail " + throwable);
                }
            });
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

2.2 消费者

package com.atpingan.rocketunion.service.impl;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.UnsupportedEncodingException;
import java.util.concurrent.*;

/**
 * @author Admin
 * @version 1.0
 * @date 2023-10-14 21:30
 **/
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "ESG_WARN_ALARM_GROUP", topic = "ESG_WARN_ALARM_TOPIC", consumeMode = ConsumeMode.ORDERLY)
public class ConsumerImplService implements RocketMQListener<MessageExt> {
    private ExecutorService threadPool = null;

    @PostConstruct
    public void init() {
        threadPool = new ThreadPoolExecutor(
                2,
                Runtime.getRuntime().availableProcessors(),
                2L,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(3000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    @Override
    public void onMessage(MessageExt messageExt) {
        threadPool.execute(new RocketMqConsumerPool(messageExt));
    }

    private class RocketMqConsumerPool implements Runnable {
        private MessageExt msg;

        public RocketMqConsumerPool(MessageExt msg) {
            this.msg = msg;
        }

        @Override
        public void run() {
            try {
                String messageBody = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                log.info("topic:"+msg.getTopic());
                log.info("tags:"+msg.getTags());
                if (msg.getTopic().equalsIgnoreCase("test-topic")) {
                    if (msg.getTags().equalsIgnoreCase("ESG_WARN_ALARM")) {
                        dealTestTopic(messageBody);
                    }
                } else if (msg.getTopic().equalsIgnoreCase("ESG_WARN_ALARM_TOPIC")) {
                    if (msg.getTags().equalsIgnoreCase("ESG_WARN_ALARM")) {
                        dealEsgWarn(messageBody);
                    }
                }

            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }
    private void dealTestTopic(String messageBody) {
        log.info(Thread.currentThread().getName() + ":子线程执行了");
        log.info(messageBody);
    }
    private void dealEsgWarn(String messageBody) {
        log.info(Thread.currentThread().getName() + ":子线程执行了");
        JSONObject jsonObject = JSONObject.parseObject(messageBody);
        log.info("消费消息:"+messageBody);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

四、存储文件

在这里插入图片描述

RocketMQ将所有主题的消息存储在Commitlog同一文件中，且按发送顺序写入文件，这样尽最大的能力确保消息发送的高性能与高吞吐量。但是按照消息主题检索消息带来了极大的不便，RocketMQ引入了ConsumeQueue消息队列文件，
Commitlog文件目录下每个文件默认大小1G，文件命名方式是：以文件中第一个消息偏移量命名。其中，每条消息的长度不相同。消息存储目录，按消息发送时顺序写入
ConsumeQueue每个消息主题包含多个消息消费队列（内容相同），每一个消息队列有一个消息文件，其专门为消息订阅构建的索引文件，提高根据主题与消息队列检索消息的速度。包含了minOffset、consumerOffset、maxOffset
IndexFile索引文件，其主要设计理念为了加速消息的检索性能，根据消息的属性快速从Commitlog文件中检索消息。存储了Header、Slot
Table 、IndexLinkedList等文件
checkpoint文件，是commitlog、consumequeue、index文件最后一次刷盘的时间戳。该文件固定长度为4k，其中只用该文件的前面24个字节

RocketMQ是高性能的消息中间件，存储部分的设计是核心，存储的核心是IO访问性能。下图是消息的数据流向。
在这里插入图片描述

消息投递与消费流程介绍：

1、生产者如何投递消息：

1.生产者在投递消息到mq服务器端，会将该消息存放在commitlog日志文件中（顺序写）、
2.Mq后台就会开启一个异步的线程将该commitlogoffset实现分配存放到不同队列中。

2、消费者如何消费：
1.消费者消费消息的时候订阅到队列（consumequeue），根据queueoffset 获取到该commitlogoffset
2.在根据commitlogoffset 去commitlog日志文件中查找到该消息主体返回给客户端。

消息的commitlogoffset 如何存放在不同的consumequeue中。 Consumequeue==16 投递消息消息key消息key%16=1 Consumequeue 中 consumeoffset 对应一条消息（没有对应消息主体）—commitlogoffset 消费者消费我们消息

为什么数据存储在commitlog，而consumerQueue只存储commitLogOffs、msgSize、tagsCode？

与kafka的设计不同，根据阿里巴巴消息中间件团队的测试，如果每个topic中的partition 分区存储的消息过多，可能会影响到磁盘io的读写性能，所以采用ConsumeQueue存放少量的数据，消息读取还是通过commitlog文件中查找

在（kafka、rocketmq）中消费成功或者失败都不会立即将该消息删除，日志清理策略删除

五、过期文件删除策略

RocketMQ判断文件是否过期的唯一标准就是非当前写文件的保留时间,并不关心文件当中的消息是否被消费过。
RocketMq的内部有一个定时任务，对文件进行扫描，触发文件删除。在broker.conf指定删除策略

#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#检测物理文件磁盘空间 如果磁盘占用达到阈值也会触发过期文件的删除，官方建议磁盘不要小于4G
diskMaxUsedSpaceRatio=88
1
2
3
4
5
6

六、零拷贝

高效文件写

零拷贝 final FileChannel fromChannel = fromFile.getChannel();
fromChannel.transferTo(position, count, toChannel)
硬盘上文件的位置和应用程序缓冲区(application
buffers)进行映射（建立一种一一对应关系），由于mmap()将文件直接映射到用户空间，所以实际文件读取时根据这个映射关系，
直接将文件从硬盘拷贝到用户空间，只进行了一次数据拷贝，不再有文件内容从硬盘拷贝到内核空间的一个缓冲区

顺序写

在磁盘中提前声明一块连续的空间,每次写数据就避免去寻址,直接在之前的写入地址后面接着写就可以。能达内存速度。

七、保证消息不丢失

1.同步刷盘

把内存数据写入磁盘就叫刷盘。异步刷盘 ASYNC_FLUSH 同步刷盘 SYNC_FLUSH
属性：flushDiskType=ASYNC_FLUSH

2.同步主从复制

同步复制：Master和Slave都写入消息成功后才返回客户端写入成功状态降低吞吐异步复制：
只要写入master就反馈客户端写入成功，然后再异步将消息复制给slave

3.发送端ACK 机制

Producer 发送消息后，Broker 会返回 ACK 确认信号，表示消息已成功发送。如果 Broker 未收到 ACK
确认信号，则会尝试重新发送消息，直到收到确认
SEND_OK：消息发送成功
FLUSH_DISK_TIMEOUT：消息发送成功但是消息刷盘超时。
FLUSH_SLAVE_TIMEOUT：消息发送成功但是消息同步到 slave 节点时超时。
SLAVE_NOT_AVAILABLE：消息发送成功但是 broker 的 slave 节点不可用

4.消费端返回CONSUME_SUCCESS

如果 Consumer 消费成功，返回 CONSUME_SUCCESS，提交 offset 并从 Broker 拉取下一批消息

八、消息的幂等

实现幂等消费原理
在这里插入图片描述

at most once 最多一次
at least once 至少一次
exactly once 刚好一次

官方：msgId来保证幂等（不推荐）我们可以用setKeys方法来设置key来保证幂等（推荐）
MessageId有可能出现冲突（重复）的情况，所以真正安全的幂等处理，不建议以MessageId作为处理依据。而最好的方式是以业务唯一标识作为幂等处理的关键依据（如订单ID）。业务的唯一标识可以在生产者通过消息 Key 设置

生产

Message message = new Message();
message.setKey("ORDERID_001");
SendResult sendResult = producer.send(message);  
1
2
3

消费

consumer.subscribe("ons_test", "*", new MessageListener() {
    public Action consume(Message message, ConsumeContext context) {
        String key = message.getKey()
        // 根据业务唯一标识的 Key 做幂等处理
        //消费者收到消息时可以根据消息的 Key判断是否重复来实现消息幂等。
        //这里我们用到了redis存放消息key值（因为redis读取快），
       // 并且对于key值大存放时长可以设置，超过了时长就会被清除掉
    }
});     
1
2
3
4
5
6
7
8
9

九、消息的顺序性

全局有序

如果要保证消息的全局有序，首先只能由一个生产者往Topic发送消息，并且一个Topic内部只能有一个队列（分区），消费者也必须是单线程消费这个队列。即一个producer -> 一个broker -> 一个consumer，这样的消息就是全局有序的。不过一般情况下我们都不需要全局有序，即使是同步MySQL Binlog也只需要保证单表消息有序即可

//可以通过代码指定创建1个队列即可
producer.setDefaultTopicQueueNums(1);
1
2

部分有序

因此绝大部分的有序需求是部分有序，部分有序我们就可以将Topic内部划分成我们需要的队列数，把消息通过特定的策略发往固定的队列中，然后每个队列对应一个单线程处理的消费者，即n个producer -> n个broker -> n个consumer，这样即完成了部分有序的需求，又可以通过队列数量的并发来提高消息处理效率

producer如何实现把消息发送到指定队列？

Order order = new Order(i,"订单"+i,"创建");

        //添加内容
        byte[] bytes = (JSON.toJSONString(order)).getBytes(CharsetUtil.UTF_8);

        Message message = new Message("topic-order","product-order",bytes);
        message.setKeys("key-"+i);

        //执行发送
        SendResult result = producer.send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Long id = (Long) arg;
                //使用取模算法确定id存放到哪个队列
                int index =(int) (id % mqs.size());
                //index就是要存放的队列的索引
                return mqs.get(index);
            }
//把订单ID作为参数，作为选择器的基础数据
        },order.getId());

        System.out.println(result);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

RocketMQ的顺序消息是通过消息队列来保证的，同一个消息队列中的消息是有序的，但是不同的队列之间是没有顺序保证的
生产者生产消息保证顺序一致生产者生产的消息最终到达 broker 且被处理的顺序一致 broker 保证先后到达消息最终处理的结果顺序一致
消费者可以按照固定顺序去拉取并处理消息。MessageListenerConcurrently是拉取到新消息之后就提交到线程池去消费，而MessageListenerOrderly则是通过加分布式锁和本地锁保证同时只有一条线程去消费一个队列上的数据

如何解决消息顺序一致性的问题？

解决办法：

生产者投递消息根据key 投递到同一个队列中存放
消费者应该订阅到同一个队列实现消费
最终应该使用同一个线程去消费消息（不能够实现多线程消费。）
实际上做业务逻辑开发中，很少有需要保证消息顺序一致性问题。

生产者：

//实际中使用 单ID，产品ID等标识
String uuid = UUID.randomUUID().toString();
SendResult result1 = rocketMQTemplate.syncSendOrderly(RocketMQConfig.TOPIC_SEQUENTIAL, "insert", uuid);
1
2
3

rocketMQTemplate.syncSendOrderly的三个参数

参数一：topic 如果想添加tag,可以使用"topic:tag"的写法
参数二：消息内容
参数三：hashKey 使用此参数选择队列。例如：orderId,productId…

因为broker会管理多个消息队列，这个hashKey参数，主要用来计算选择队列的，一般可以把订单ID，产品ID作为参数值；
发送到一个队列，这样方便搞顺序队列；以及消费端接收的时候，默认是并发多线程去接收消息

消费者：

@Service
@Slf4j
@RocketMQMessageListener(consumerGroup = "mayikt-group20", topic = "topic_seq", consumeMode = ConsumeMode.ORDERLY
)
public class RocketMQConsumer01 implements RocketMQListener<String> {
    @Override
    public void onMessage(String msg) {
        try {
            Random r = new Random(100);
            int i = r.nextInt(500);
            Thread.sleep(i);
        } catch (Exception e) {

        }
        log.info("消费者监听到消息：", msg);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

ConsumeMode.ORDERLY，单线程顺序接收消息

十、消息积压如何解决？

1.临时扩容增加消费者数量
2.排查解决异常问题，消费端是否有bug。
3.如果消费端处理不足，水平扩容提升消费端的并发处理能力。在扩容消费者的同时,必须同步扩容Topic的队列数量，确保消费者的实例数和队列数是相同的，队列是单线程消费

//1.代码设置队列数
producer.setDefaultTopicQueueNums(1);
1
2

通过可视化界面修改Topic中的队列数量，RocketMQ的DashBoard控制台

十一、RocketMQ解决分布式事务问题

原理图：
在这里插入图片描述
RocketMQ实现分布式事务的原理：

1.生产者投递一个半消息给我们RocketMQ服务器端存放，该消息暂时无法被我们消费者消费。
2.RocketMQ将该消息落地存放硬盘中，RocketMQ发送ACK给生产者。
3.生产者收到事件监听之后，开始执行生产者本地事务的操作；
4.如果生产者执行本地的事务操作，如果成功的情况下，则发送一个提交通知给RocketMQ 服务器端，RocketMQ服务器端将该消息，推送给消费者消费。
5.如果生产者执行本地事务操作，如果失败的情况下，则发送一个回滚通知给rocketmq服务器端，rocketmq服务器端在从本地将该消息删除，不会给消费者消费。

核心思想：确保生产者一定将消息投递到mq服务器端，生产者必须先一定执行完成，在执行消费者。

消息的状态

回滚: RocketMQLocalTransactionState.ROLLBACK
提交: RocketMQLocalTransactionState.COMMIT
不是提交也不是回滚: RocketMQLocalTransactionState.UNKNOWN

代码示例：
在这里插入图片描述

生产者：
①入口调用

@Slf4j
@RestController
@RequestMapping("/rocketmq")
public class OrderProducerController {
    @Autowired
    private SyncProducer syncProducer;

    @RequestMapping("/sendMsg")
    public TransactionSendResult sendMsg() {
        String orderId = System.currentTimeMillis() + "";
        // 1.先投递一个半消息到MQ中
        TransactionSendResult msg = syncProducer.sendSyncMessage(orderId, "mayikt-topic2022", "");
        log.info(">>msg:{}<<<", msg);
        return msg;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

②投递消息

@Slf4j
@Component
public class SyncProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    public TransactionSendResult sendSyncMessage(String msg, String topic, String tag) {
        log.info("【发送消息】：{}", msg);
        Message<String> message = MessageBuilder.withPayload(msg).build();
        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction
                ("ESG_WARN_ALARM_GROUP:abc",  message,topic);
        log.info("【发送状态】：{}", result.getLocalTransactionState());
        return result;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

③监听器

@Slf4j
@Component
@RocketMQTransactionListener
public class SyncProducerListener implements RocketMQLocalTransactionListener {
	/**
	* 监听消息发送到MQ服务端成功(当前为UNKNOWN状态)
	**/
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {

        try {
            String orderId = new String((byte[]) message.getPayload());
            if (StringUtils.isEmpty(orderId)) {
                return RocketMQLocalTransactionState.ROLLBACK;
            }
            //orderManage.insertOrder(orderId);
            log.info("【本地业务执行完毕】 msg:{}, Object:{}", message, o);
//            int i = 1 / 0;
            // 本地事务执行成功
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.info(">>e:{}<<", e);
            // 本地事务执行异常
            log.error("【执行本地业务异常】 exception message:{}", e.getMessage());
            return RocketMQLocalTransactionState.ROLLBACK;
        }

    }
	/**
	*每隔60秒来检测一次
	**/
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        log.info("【执行检查任务】");
        String orderId = new String((byte[]) message.getPayload());
       // OrderEntity orderEntity = orderMapper.findOrderId(orderId);
        String orderEntity = null;
        // 如果该不存在对象情况下
        if (orderEntity == null) {
            return RocketMQLocalTransactionState.UNKNOWN;
        }
        // 如果该对象存在的情况下，则提交该事务 返回提交状态给rocketmq
        return RocketMQLocalTransactionState.COMMIT;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

消费者：

@Slf4j
@Service
@RocketMQMessageListener(topic = "ESG_WARN_ALARM_GROUP",consumerGroup = "tx-consumer-group")
public class RocketMQConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        log.info("消费者监听到消息：", message);
    }
}
1
2
3
4
5
6
7
8
9

十二、死信队列

1、什么是死信队列？

当一条消息初次消费失败，消息队列会自动进行消费重试；达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，消息队列不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。这个队列就是死信队列（Dead-Letter Queue，DLQ），而其中的消息则称为死信消息（Dead-Letter Message，DLM）。死信队列是用于处理无法被正常消费的消息的。

2、死信队列的特征

死信队列具有如下特征：
① 死信队列中的消息不会再被消费者正常消费，即DLQ对于消费者是不可见的
② 死信存储有效期与正常消息相同，均为 3 天（commitlog文件的过期时间），3 天后会被自动删除
③ 死信队列就是一个特殊的Topic，名为%DLQ%consumerGroup@consumerGroup ，即每个消费者组都有一个死信队列
④ 如果⼀个消费者组未产生死信消息，则不会为其创建相应的死信队列

3、死信消息的处理

实际上，当⼀条消息进入死信队列，就意味着系统中某些地方出现了问题，从而导致消费者无法正常消费该消息，比如代码中原本就存在Bug。因此，对于死信消息，通常需要开发人员进行特殊处理。最关键的步骤是要排查可疑因素，解决代码中可能存在的Bug，然后再将原来的死信消息再次进行投递消费。

相关阅读:
某大型国有银行 VMware 替换与轻量信创云底座转型实践｜信创专题
 KY90 简单密码
 蛤蟆先生去看心理医生笔记
 《重构代码设计》
RabbitMQ、Kafka对比（超详细），Kafka、RabbitMQ、RocketMQ的区别
 Linux以系统服务的方式启动Kafka(其他服务同理)
R语言使用plot函数可视化数据散点图，使用log函数对X轴数据和Y轴数据进行对数变换后再进行可视化（log(x)，log(y)）
Servlet的生命周期
 matplotlib绘制直方图示例
 从0搭建Vue3组件库(六):前端流程化控制工具gulp的使用
原文地址：https://blog.csdn.net/m0_37635053/article/details/133780160

RocketMQ学习笔记