RocketMQ学习笔记

港控/mmm° 2024-02-23 08:18 9阅读 0赞

### RocketMQ学习笔记 ###

![在这里插入图片描述][c0ffeb7850874bb7ae04355da69565e2.png]

#### 一、下载安装 ####

##### 1.1 官网下载 #####

`官网地址:`

> http://rocketmq.apache.org

`下载地址：`

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

`控制台下载地址:`

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

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

> 设计架构如下：

<table> 
 <thead> 
  <tr> 
   <th>服务器</th> 
   <th>部署节点</th> 
   <th>部署broker</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>192.168.10.131</td> 
   <td>namesrv、broker</td> 
   <td>broker-a-m、broker-b-s</td> 
  </tr> 
  <tr> 
   <td>192.168.10.132</td> 
   <td>namesrv、broker</td> 
   <td>broker-b-m、broker-a-s</td> 
  </tr> 
  <tr> 
   <td>192.168.10.133</td> 
   <td>namesrv</td> 
   <td>仅部署namesrv</td> 
  </tr> 
 </tbody> 
</table>

###### 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"

`vim runbroker.sh`

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

> 在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

> `修改配置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 配置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

> `修改配置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.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() + " => 成功";
    }

> `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);
            }
        });
    }

> `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.springMVC集成RocketMq #####

> 导入包版本与rocketMQ相匹配

<dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>4.9.0</version>
            </dependency>

###### 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();
                }
            }
        }
    }

> 说明：①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("消费者关闭");
            }
        }
    }

##### 2.springboot集成RocketMq #####

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

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

> 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

###### 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);
                    }
                });
            }
        }
    }

###### 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);
        }
    
    }

#### 四、 存储文件 ####

![在这里插入图片描述][d582b0b040624a7c8c59d485670f0b14.png]

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

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

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

> `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

#### 六、 零拷贝 ####

高效文件写

> 零拷贝 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 拉取下一批消息

#### 八、消息的幂等 ####

**实现幂等消费原理**  
![在这里插入图片描述][a3c8c190f713445d8c9677e317b322b7.png]

> 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);

**消费**

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

#### 九、消息的顺序性 ####

> RocketMQ的顺序消息是通过消息队列来保证的，同一个消息队列中的消息是有序的，但是不同的队列之间是没有顺序保证的  
> 生产者生产消息保证顺序一致 生产者生产的消息最终到达 broker 且被处理的顺序一致 broker 保证先后到达消息最终处理的结果顺序一致  
> 消费者可以按照固定顺序去拉取并处理消息

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

**解决办法：**

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

**生产者：**

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

*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:{}>", msg);
        }
    }

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

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

**原理图：**  
![在这里插入图片描述][894d0c74e39a4ed2a2c235d7de7f4125.png]  
**RocketMQ实现分布式事务的原理：**

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

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

**消息的状态**

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

**代码示例：**  
![在这里插入图片描述][11bc28bf6ed44dfeb7f3f685354947e9.png]

**生产者：**  
`①入口调用`

@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;
        }
    }

`②投递消息`

@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;
        }
    }

`③监听器`

@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;
        }
    }

**消费者：**

@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("消费者监听到消息：<orderId:{}>", message);
        }
    }

#### 十一、死信队列 ####

**1、 什么是死信队列？**

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

**2、 死信队列的特征**

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

**3、 死信消息的处理**

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

[c0ffeb7850874bb7ae04355da69565e2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/70c936dabaf5477db56ee4e60c3d2afd.png
[d582b0b040624a7c8c59d485670f0b14.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/660b541492554855b55dc98cc672c235.png
[f0894a68d36b48e6918c012b8b0e6b1d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/87c3af2b91aa40d980c61727c03f7878.png
[101a840809f6497aacd650f9a79c7d3f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/e0b5f03ee3734d44bd11684fb9d9a216.png
[a3c8c190f713445d8c9677e317b322b7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/1d5b5911d0c84e41bccfb304225baea6.png
[894d0c74e39a4ed2a2c235d7de7f4125.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/8883c15962bc4bd0b47439b5ea20c914.png
[11bc28bf6ed44dfeb7f3f685354947e9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/faa15552ef624a90b3472c7c1cc38b41.png