RocketMQ基础

曾经终败给现在 2022-10-30 13:29 89阅读 0赞

**目录**

1.MQ优缺点

2.各种MQ产品的比较

3.RocketMQ集群介绍

3.1各角色介绍

3.2集群特点

3.3集群模式

3.4集群工作流程

4.双主双从同步配置说明

4.1broker1主配置文 件（broker-a.properties)

4.2broker1从配置文件（broker-a-s.properties）

4.3broker2主配置文件（broker-b.properties）

4.4broker2从配置文件（b roker-b-s.properties）

5.生产者发送消息步骤

6.消费者获取消息步骤

7.生产者发送消息方式

7.1同步消息

7.2异步消息

7.3单向消息

7.4顺序消息（在某一类消息有先后顺序时，讲有顺序的消息下发到同一个broker的同一个队列中，消费者必须使用负载均衡模式接收）

7.5延时消息

7.6批量消息（不支持顺序发送）

7.7过滤消息（在消费时根据消费者的过滤条件进行消息筛选，tag过滤很简单不赘述，只介绍sql过滤）

7.8事务消息（消息在下发到broker之后需要等待生产者的后续逻辑操作完成之后，才能进行下发或者消息删除）

8.消费者监听消息方式

8.1监听消息模式（messageModel：广播/负载均衡）

8.2处理消息模式（consumeMode：有序/有序）

8.3过滤消息模式（selectorType：标签/Sql）

8.4样例

9.备注

--------------------

# 1.MQ优缺点 #

优点：

*  解耦
    
    系统的耦合性越高，容错性就越低。以电商应用为例，用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障或者因为升级等原因暂时不可用，都会造成下单操作异常，影响用户使用体验。
 *  流量削峰
    
    应用系统如果遇到系统请求流量的瞬间猛增，有可能会将系统压垮。有了消息队列可以将大量请求缓存起来，分散到很长一段时间处理，这样可以大大提到系统的稳定性和用户体验。
 *  数据分发
    
    通过消息队列可以让数据在多个系统更加之间进行流通。数据的产生方不需要关心谁来使用数据，只需要将数据发送到消息队列，数据使用方直接在消息队列中直接获取数据即可

缺点：

*  系统可用性降低
    
    系统引入的外部依赖越多，系统稳定性越差。一旦MQ宕机，就会对业务造成影响。
    
    如何保证MQ的高可用？
 *  系统复杂度提高
    
    MQ的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过MQ进行异步调用。
    
    如何保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？
 *  一致性问题
    
    A系统处理完业务，通过MQ给B、C、D三个系统发消息数据，如果B系统、C系统处理成功，D系统处理失败。
    
    如何保证消息数据处理的一致性？

# 2.各种MQ产品的比较 #

常见的MQ产品包括Kafka、ActiveMQ、RabbitMQ、RocketMQ

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70][]

# 3.RocketMQ集群介绍 #

## 3.1各角色介绍 ##

*  Producer：消息的发送者；举例：发信者
 *  Consumer：消息接收者；举例：收信者
 *  Broker：暂存和传输消息；举例：邮局
 *  NameServer：管理Broker；举例：各个邮局的管理机构
 *  Topic：区分消息的种类；一个发送者可以发送消息给一个或者多个Topic；一个消息的接收者可以订阅一个或者多个Topic消息
 *  Message Queue：相当于是Topic的分区；用于并行发送和接收消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 1][]

## 3.2集群特点 ##

*  NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。
 *  Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。
 *  Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。
 *  Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。

## 3.3集群模式 ##

1）单Master模式

这种方式风险较大，一旦Broker重启或者宕机时，会导致整个服务不可用。不建议线上环境使用,可以用于本地测试。

2）多Master模式

一个集群无Slave，全是Master，例如2个Master或者3个Master，这种模式的优缺点如下：

*  优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢），性能最高；
 *  缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅，消息实时性会受到影响。

3）多Master多Slave模式（异步）

每个Master配置一个Slave，有多对Master-Slave，HA采用异步复制方式，主备有短暂消息延迟（毫秒级），这种模式的优缺点如下：

*  优点：即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时Master宕机后，消费者仍然可以从Slave消费，而且此过程对应用透明，不需要人工干预，性能同多Master模式几乎一样；
 *  缺点：Master宕机，磁盘损坏情况下会丢失少量消息。

4）多Master多Slave模式（同步）

每个Master配置一个Slave，有多对Master-Slave，HA采用同步双写方式，即只有主备都写成功，才向应用返回成功，这种模式的优缺点如下：

*  优点：数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高；
 *  缺点：性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。

## 3.4集群工作流程 ##

1.  启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
2.  Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
3.  收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。
4.  Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。
5.  Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。

# 4.双主双从同步配置说明 #

配置文件：rocket安装路径\\\\conf\\2m-2s-sync\\

## 4.1broker1主配置文 件（broker-a.properties) ##

#所属集群名字
    brokerClusterName=rocketmq-cluster
    #broker名字，注意此处不同的配置文件填写的不一样
    brokerName=broker-a
    #0 表示 Master，>0 表示 Slave
    brokerId=0
    #nameServer地址，分号分割
    namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2: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/store
    #commitLog 存储路径
    storePathCommitLog=/usr/local/rocketmq/store/commitlog
    #消费队列存储路径存储路径
    storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
    #消息索引存储路径
    storePathIndex=/usr/local/rocketmq/store/index
    #checkpoint 文件存储路径
    storeCheckpoint=/usr/local/rocketmq/store/checkpoint
    #abort 文件存储路径
    abortFile=/usr/local/rocketmq/store/abort
    #限制的消息大小
    maxMessageSize=65536
    #flushCommitLogLeastPages=4
    #flushConsumeQueueLeastPages=2
    #flushCommitLogThoroughInterval=10000
    #flushConsumeQueueThoroughInterval=60000
    #Broker 的角色
    #- ASYNC_MASTER 异步复制Master
    #- SYNC_MASTER 同步双写Master
    #- SLAVE
    brokerRole=SYNC_MASTER
    #刷盘方式
    #- ASYNC_FLUSH 异步刷盘
    #- SYNC_FLUSH 同步刷盘
    flushDiskType=SYNC_FLUSH
    #checkTransactionMessageEnable=false
    #发消息线程池数量
    #sendMessageThreadPoolNums=128
    #拉消息线程池数量
    #pullMessageThreadPoolNums=128

## 4.2broker1从配置文件（broker-a-s.properties） ##

#所属集群名字
    brokerClusterName=rocketmq-cluster
    #broker名字，注意此处不同的配置文件填写的不一样
    brokerName=broker-a
    #0 表示 Master，>0 表示 Slave
    brokerId=1
    #nameServer地址，分号分割
    namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
    #在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
    defaultTopicQueueNums=4
    #是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
    autoCreateTopicEnable=true
    #是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
    autoCreateSubscriptionGroup=true
    #Broker 对外服务的监听端口
    listenPort=11011
    #删除文件时间点，默认凌晨 4点
    deleteWhen=04
    #文件保留时间，默认 48 小时
    fileReservedTime=120
    #commitLog每个文件的大小默认1G
    mapedFileSizeCommitLog=1073741824
    #ConsumeQueue每个文件默认存30W条，根据业务情况调整
    mapedFileSizeConsumeQueue=300000
    #destroyMapedFileIntervalForcibly=120000
    #redeleteHangedFileInterval=120000
    #检测物理文件磁盘空间
    diskMaxUsedSpaceRatio=88
    #存储路径
    storePathRootDir=/usr/local/rocketmq/store
    #commitLog 存储路径
    storePathCommitLog=/usr/local/rocketmq/store/commitlog
    #消费队列存储路径存储路径
    storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
    #消息索引存储路径
    storePathIndex=/usr/local/rocketmq/store/index
    #checkpoint 文件存储路径
    storeCheckpoint=/usr/local/rocketmq/store/checkpoint
    #abort 文件存储路径
    abortFile=/usr/local/rocketmq/store/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

## 4.3broker2主配置文件（broker-b.properties） ##

#所属集群名字
    brokerClusterName=rocketmq-cluster
    #broker名字，注意此处不同的配置文件填写的不一样
    brokerName=broker-b
    #0 表示 Master，>0 表示 Slave
    brokerId=0
    #nameServer地址，分号分割
    namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2: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/store
    #commitLog 存储路径
    storePathCommitLog=/usr/local/rocketmq/store/commitlog
    #消费队列存储路径存储路径
    storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
    #消息索引存储路径
    storePathIndex=/usr/local/rocketmq/store/index
    #checkpoint 文件存储路径
    storeCheckpoint=/usr/local/rocketmq/store/checkpoint
    #abort 文件存储路径
    abortFile=/usr/local/rocketmq/store/abort
    #限制的消息大小
    maxMessageSize=65536
    #flushCommitLogLeastPages=4
    #flushConsumeQueueLeastPages=2
    #flushCommitLogThoroughInterval=10000
    #flushConsumeQueueThoroughInterval=60000
    #Broker 的角色
    #- ASYNC_MASTER 异步复制Master
    #- SYNC_MASTER 同步双写Master
    #- SLAVE
    brokerRole=SYNC_MASTER
    #刷盘方式
    #- ASYNC_FLUSH 异步刷盘
    #- SYNC_FLUSH 同步刷盘
    flushDiskType=SYNC_FLUSH
    #checkTransactionMessageEnable=false
    #发消息线程池数量
    #sendMessageThreadPoolNums=128
    #拉消息线程池数量
    #pullMessageThreadPoolNums=128

## 4.4broker2从配置文件（b roker-b-s.properties） ##

#所属集群名字
    brokerClusterName=rocketmq-cluster
    #broker名字，注意此处不同的配置文件填写的不一样
    brokerName=broker-b
    #0 表示 Master，>0 表示 Slave
    brokerId=1
    #nameServer地址，分号分割
    namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
    #在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
    defaultTopicQueueNums=4
    #是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
    autoCreateTopicEnable=true
    #是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
    autoCreateSubscriptionGroup=true
    #Broker 对外服务的监听端口
    listenPort=11011
    #删除文件时间点，默认凌晨 4点
    deleteWhen=04
    #文件保留时间，默认 48 小时
    fileReservedTime=120
    #commitLog每个文件的大小默认1G
    mapedFileSizeCommitLog=1073741824
    #ConsumeQueue每个文件默认存30W条，根据业务情况调整
    mapedFileSizeConsumeQueue=300000
    #destroyMapedFileIntervalForcibly=120000
    #redeleteHangedFileInterval=120000
    #检测物理文件磁盘空间
    diskMaxUsedSpaceRatio=88
    #存储路径
    storePathRootDir=/usr/local/rocketmq/store
    #commitLog 存储路径
    storePathCommitLog=/usr/local/rocketmq/store/commitlog
    #消费队列存储路径存储路径
    storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
    #消息索引存储路径
    storePathIndex=/usr/local/rocketmq/store/index
    #checkpoint 文件存储路径
    storeCheckpoint=/usr/local/rocketmq/store/checkpoint
    #abort 文件存储路径
    abortFile=/usr/local/rocketmq/store/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

# 5.生产者发送消息步骤 #

1.   创建消息生产者producer，并制定生产者组名
2.  指定Nameserver地址
3.  启动producer
4.  创建消息对象，指定主题Topic、Tag和消息体
5.  发送消息
6.  关闭生产者producer

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.client.producer.SendStatus;
    import org.apache.rocketmq.common.message.Message;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * 发送同步消息
     */
    public class SyncProducer {
    
        public static void main(String[] args) throws Exception {
            //1.创建消息生产者producer，并制定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("group1");
            //2.指定Nameserver地址
            producer.setNamesrvAddr("127.0.0.1:9876;127.0.0.1:9877");
            //3.启动producer
            producer.start();
    
            for (int i = 0; i < 10; i++) {
                //4.创建消息对象，指定主题Topic、Tag和消息体
                /**
                 * 参数一：消息主题Topic
                 * 参数二：消息Tag
                 * 参数三：消息内容
                 */
                Message msg = new Message("springboot-mq", "Tag1", ("Hello World" + i).getBytes());
                //5.发送消息
                SendResult result = producer.send(msg);
                //发送状态
                SendStatus status = result.getSendStatus();
    
                System.out.println("发送结果:" + result);
    
                //线程睡1秒
                TimeUnit.SECONDS.sleep(1);
            }
    
            //6.关闭生产者producer
            producer.shutdown();
        }
    }

# 6.消费者获取消息步骤 #

1.  创建消费者Consumer，制定消费者组名
2.  指定Nameserver地址
3.  订阅主题Topic和Tag
4.  设置回调函数，处理消息
5.  启动消费者consumer

# 7.生产者发送消息方式 #

## 7.1同步消息 ##

第一种实现方式

第二种实现方式![gif.latex][]

## 7.2异步消息 ##

第一种实现方式

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 java.util.concurrent.TimeUnit;
    
    /**
     * 发送异步消息
     */
    public class AsyncProducer {
    
        public static void main(String[] args) throws Exception {
            //1.创建消息生产者producer，并制定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("group1");
            //2.指定Nameserver地址
            producer.setNamesrvAddr("127.0.0.1:9876;127.0.0.1:9877");
            //3.启动producer
            producer.start();
    
            for (int i = 0; i < 10; i++) {
                //4.创建消息对象，指定主题Topic、Tag和消息体
                /**
                 * 参数一：消息主题Topic
                 * 参数二：消息Tag
                 * 参数三：消息内容
                 */
                Message msg = new Message("base", "Tag2", ("Hello World" + i).getBytes());
                //5.发送异步消息
                producer.send(msg, new SendCallback() {
                    /**
                     * 发送成功回调函数
                     * @param sendResult
                     */
                    public void onSuccess(SendResult sendResult) {
                        System.out.println("发送结果：" + sendResult);
                    }
    
                    /**
                     * 发送失败回调函数
                     * @param e
                     */
                    public void onException(Throwable e) {
                        System.out.println("发送异常：" + e);
                    }
                });
    
                //线程睡1秒
                TimeUnit.SECONDS.sleep(1);
            }
    
            //6.关闭生产者producer
            producer.shutdown();
        }
    }

第二种实现方式

package com.anran.mq.mqproductservice.listener;
    
    import com.alibaba.fastjson.JSONObject;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
    
    /**
     * rocketmq异步回调监听
     */
    @Slf4j
    public class SendCallbackListener implements SendCallback {
    
        private int id;
    
        public SendCallbackListener(int id) {
            this.id = id;
        }
    
        @Override
        public void onSuccess(SendResult sendResult) {
            log.info("CallBackListener on success : " + JSONObject.toJSONString(sendResult));
        }
    
        @Override
        public void onException(Throwable throwable) {
            log.error("CallBackListener on exception : ", throwable);
        }
    }

## 7.3单向消息 ##

第一种实现方式

import org.apache.rocketmq.client.exception.MQBrokerException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.common.message.Message;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * 发送单向消息
     */
    public class OneWayProducer {
    
        public static void main(String[] args) throws Exception, MQBrokerException {
            //1.创建消息生产者producer，并制定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("group1");
            //2.指定Nameserver地址
            producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
            //3.启动producer
            producer.start();
    
            for (int i = 0; i < 3; i++) {
                //4.创建消息对象，指定主题Topic、Tag和消息体
                /**
                 * 参数一：消息主题Topic
                 * 参数二：消息Tag
                 * 参数三：消息内容
                 */
                Message msg = new Message("base", "Tag3", ("Hello World，单向消息" + i).getBytes());
                //5.发送单向消息
                producer.sendOneway(msg);
    
                //线程睡1秒
                TimeUnit.SECONDS.sleep(5);
            }
    
            //6.关闭生产者producer
            producer.shutdown();
        }
    }

第二种实现方式

## 7.4顺序消息（在某一类消息有先后顺序时，讲有顺序的消息下发到同一个broker的同一个队列中，消费者必须使用负载均衡模式接收） ##

第一种实现方式

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.MessageQueueSelector;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.common.message.MessageQueue;
    
    import java.util.List;
    
    public class Producer {
    
        public static void main(String[] args) throws Exception {
            //1.创建消息生产者producer，并制定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("group1");
            //2.指定Nameserver地址
            producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
            //3.启动producer
            producer.start();
            //构建消息集合
            List<OrderStep> orderSteps = OrderStep.buildOrders();
            //发送消息
            for (int i = 0; i < orderSteps.size(); i++) {
                String body = orderSteps.get(i) + "";
                Message message = new Message("OrderTopic", "Order", "i" + i, body.getBytes());
                /**
                 * 参数一：消息对象
                 * 参数二：消息队列的选择器
                 * 参数三：选择队列的业务标识（订单ID）
                 */
                SendResult sendResult = producer.send(message, new MessageQueueSelector() {
                    /**
                     *
                     * @param mqs：队列集合
                     * @param msg：消息对象
                     * @param arg：业务标识的参数
                     * @return
                     */
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        long orderId = (long) arg;
                        long index = orderId % mqs.size();
                        return mqs.get((int) index);
                    }
                }, orderSteps.get(i).getOrderId());
    
                System.out.println("发送结果：" + sendResult);
            }
            producer.shutdown();
        }
    
    }

import java.util.ArrayList;
    import java.util.List;
    
    /**
     * 订单构建者
     */
    public class OrderStep {
        private long orderId;
        private String desc;
    
        public long getOrderId() {
            return orderId;
        }
    
        public void setOrderId(long orderId) {
            this.orderId = orderId;
        }
    
        public String getDesc() {
            return desc;
        }
    
        public void setDesc(String desc) {
            this.desc = desc;
        }
    
        @Override
        public String toString() {
            return "OrderStep{" +
                    "orderId=" + orderId +
                    ", desc='" + desc + '\'' +
                    '}';
        }
    
        public static List<OrderStep> buildOrders() {
            //  1039L   : 创建    付款 推送 完成
            //  1065L   ： 创建   付款
            //  7235L   ：创建    付款
            List<OrderStep> orderList = new ArrayList<OrderStep>();
    
            OrderStep orderDemo = new OrderStep();
            orderDemo.setOrderId(1039L);
            orderDemo.setDesc("创建");
            orderList.add(orderDemo);
    
            orderDemo = new OrderStep();
            orderDemo.setOrderId(1065L);
            orderDemo.setDesc("创建");
            orderList.add(orderDemo);
    
            orderDemo = new OrderStep();
            orderDemo.setOrderId(1039L);
            orderDemo.setDesc("付款");
            orderList.add(orderDemo);
    
            orderDemo = new OrderStep();
            orderDemo.setOrderId(7235L);
            orderDemo.setDesc("创建");
            orderList.add(orderDemo);
    
            orderDemo = new OrderStep();
            orderDemo.setOrderId(1065L);
            orderDemo.setDesc("付款");
            orderList.add(orderDemo);
    
            orderDemo = new OrderStep();
            orderDemo.setOrderId(7235L);
            orderDemo.setDesc("付款");
            orderList.add(orderDemo);
    
            orderDemo = new OrderStep();
            orderDemo.setOrderId(1065L);
            orderDemo.setDesc("完成");
            orderList.add(orderDemo);
    
            orderDemo = new OrderStep();
            orderDemo.setOrderId(1039L);
            orderDemo.setDesc("推送");
            orderList.add(orderDemo);
    
            orderDemo = new OrderStep();
            orderDemo.setOrderId(7235L);
            orderDemo.setDesc("完成");
            orderList.add(orderDemo);
    
            orderDemo = new OrderStep();
            orderDemo.setOrderId(1039L);
            orderDemo.setDesc("完成");
            orderList.add(orderDemo);
    
            return orderList;
        }
    }

第二种实现方式

package com.anran.mq.mqproductservice.controller;
    
    import com.alibaba.fastjson.JSONObject;
    import com.anran.mq.mqproductservice.listener.SendCallbackListener;
    import com.anran.mq.mqproductservice.model.OrderStep;
    import com.anran.mq.mqproductservice.model.ResponseMsg;
    import com.anran.mq.mqproductservice.util.ListSplitter;
    import org.apache.rocketmq.client.producer.MessageQueueSelector;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.client.producer.SendStatus;
    import org.apache.rocketmq.client.producer.TransactionSendResult;
    import org.apache.rocketmq.common.message.MessageQueue;
    import org.apache.rocketmq.spring.core.RocketMQTemplate;
    import org.apache.rocketmq.spring.support.RocketMQHeaders;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.ArrayList;
    import java.util.List;
    
    @RestController
    @RequestMapping("/mqMessageController")
    public class MqMessageController {
        private static final Logger log = LoggerFactory.getLogger(MqMessageController.class);
    
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
        @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.sync-tag}")
        private String syncTag;
    
        /**
         * 发送包含顺序的单向消息
         *
         * @param id 消息
         * @return 结果
         */
        @RequestMapping("/pushSequeueMessage.action")
        public ResponseMsg pushSequeueMessage(@RequestParam("id") int id) {
            log.info("pushSequeueMessage start : " + id);
            // 创建三个不同订单的不同步骤
            for (int i = 0; i < 3; i++) {
                // 处理当前订单唯一标识
                String myId = id + "" + i;
                // 获取当前订单的操作步骤列表
                List<OrderStep> myOrderSteps = OrderStep.buildOrderSteps(myId);
                // 依次操作步骤下发消息队列
                for (OrderStep orderStep : myOrderSteps) {
                    // 构建消息
                    String messageStr = String.format("order id : %s, desc : %s", orderStep.getId(), orderStep.getDesc());
                    Message<String> message = MessageBuilder.withPayload(messageStr)
                            .setHeader(RocketMQHeaders.KEYS, orderStep.getId())
                            .build();
                    // 设置顺序下发
                    rocketMQTemplate.setMessageQueueSelector(new MessageQueueSelector() {
                        /**
                         * 设置放入同一个队列的规则
                         * @param list 消息列表
                         * @param message 当前消息
                         * @param o 比较的关键信息
                         * @return 消息队列
                         */
                        @Override
                        public MessageQueue select(List<MessageQueue> list, org.apache.rocketmq.common.message.Message message, Object o) {
                            // 根据当前消息的id，使用固定算法获取需要下发的队列
                            // （使用当前id和消息队列个数进行取模获取需要下发的队列，id和队列数量一样时，选择的队列坑肯定一样）
                            int queueNum = Integer.valueOf(String.valueOf(o)) % list.size();
                            log.info(String.format("queueNum : %s, message : %s", queueNum, new String(message.getBody())));
                            return list.get(queueNum);
                        }
                    });
                    // 设置发送地和消息信息并发送消息（Orderly）
                    rocketMQTemplate.syncSendOrderly(syncTag, message, orderStep.getId());
                }
            }
            log.info("pushSequeueMessage finish : " + id);
            ResponseMsg msg = new ResponseMsg();
            msg.setSuccessData(null);
            return msg;
        }
    }

package com.anran.mq.mqproductservice.model;
    
    import lombok.Data;
    
    import java.util.ArrayList;
    import java.util.List;
    
    /**
     * 订单消息模型
     */
    @Data
    public class OrderStep {
        /**
         * 订单id
         */
        private String id;
    
        /**
         * 操作步骤
         */
        private String desc;
    
        /**
         * 步骤构建方法
         * @param id 订单唯一标识
         * @param desc 步骤描述
         */
        public OrderStep(String id, String desc) {
            this.id = id;
            this.desc = desc;
        }
    
        /**
         * 构建订单的所有操作步骤：创建、支付、回调
         * @param id 订单唯一标识
         * @return 订单操作步骤列表
         */
        public static List<OrderStep> buildOrderSteps(String id) {
            List<OrderStep> orderSteps = new ArrayList<>();
            OrderStep createStep = new OrderStep(id, "create order");
            orderSteps.add(createStep);
            OrderStep payStep = new OrderStep(id, "pay order");
            orderSteps.add(payStep);
            OrderStep callbackStep = new OrderStep(id, "call back order");
            orderSteps.add(callbackStep);
            return orderSteps;
        }
    }

## 7.5延时消息 ##

// 现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18  
  // private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

第一种实现方式

import org.apache.rocketmq.client.exception.MQBrokerException;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.client.producer.SendStatus;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.exception.RemotingException;
    
    import java.util.concurrent.TimeUnit;
    
    public class Producer {
    
        public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
            //1.创建消息生产者producer，并制定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("group1");
            //2.指定Nameserver地址
            producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
            //3.启动producer
            producer.start();
    
            for (int i = 0; i < 10; i++) {
                //4.创建消息对象，指定主题Topic、Tag和消息体
                /**
                 * 参数一：消息主题Topic
                 * 参数二：消息Tag
                 * 参数三：消息内容
                 */
                Message msg = new Message("DelayTopic", "Tag1", ("Hello World" + i).getBytes());
                //设定延迟时间
                msg.setDelayTimeLevel(2);
                //5.发送消息
                SendResult result = producer.send(msg);
                //发送状态
                SendStatus status = result.getSendStatus();
    
                System.out.println("发送结果:" + result);
    
                //线程睡1秒
                TimeUnit.SECONDS.sleep(1);
            }
    
            //6.关闭生产者producer
            producer.shutdown();
        }
    
    }

第二种实现方式

package com.anran.mq.mqproductservice.controller;
    
    import com.alibaba.fastjson.JSONObject;
    import com.anran.mq.mqproductservice.listener.SendCallbackListener;
    import com.anran.mq.mqproductservice.model.OrderStep;
    import com.anran.mq.mqproductservice.model.ResponseMsg;
    import com.anran.mq.mqproductservice.util.ListSplitter;
    import org.apache.rocketmq.client.producer.MessageQueueSelector;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.client.producer.SendStatus;
    import org.apache.rocketmq.client.producer.TransactionSendResult;
    import org.apache.rocketmq.common.message.MessageQueue;
    import org.apache.rocketmq.spring.core.RocketMQTemplate;
    import org.apache.rocketmq.spring.support.RocketMQHeaders;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.ArrayList;
    import java.util.List;
    
    @RestController
    @RequestMapping("/mqMessageController")
    public class MqMessageController {
        private static final Logger log = LoggerFactory.getLogger(MqMessageController.class);
    
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
        @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.sync-tag}")
        private String syncTag;
    
        /**
         * rocketmq 延迟消息
         *
         * @param id 消息
         * @return 结果
         */
        @RequestMapping("/pushDelayMessage.action")
        public ResponseMsg pushDelayMessage(@RequestParam("id") int id) {
            log.info("pushDelayMessage start : " + id);
            // 构建消息
            String messageStr = "order id : " + id;
            Message<String> message = MessageBuilder.withPayload(messageStr)
                    .setHeader(RocketMQHeaders.KEYS, id)
                    .build();
            // 设置超时和延时推送
            // 超时时针对请求broker然后结果返回给product的耗时
            // 现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18
            // private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
            SendResult sendResult = rocketMQTemplate.syncSend(syncTag, message, 1 * 1000l, 4);
            log.info("pushDelayMessage finish : " + id + ", sendResult : " + JSONObject.toJSONString(sendResult));
            ResponseMsg msg = new ResponseMsg();
            // 解析发送结果
            if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
                msg.setSuccessData(sendResult.getMsgId() + " : " + sendResult.getSendStatus());
            }
            return msg;
        }
    }

## 7.6批量消息（不支持顺序发送，最大支持4M的消息大小，因此需要进行消息切割） ##

消息切割对象见：[https://blog.csdn.net/qq\_26154077/article/details/111013842][https_blog.csdn.net_qq_26154077_article_details_111013842] 中ListSplitter类

第一种实现方式

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.client.producer.SendStatus;
    import org.apache.rocketmq.common.message.Message;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.TimeUnit;
    
    public class Producer {
    
        public static void main(String[] args) throws Exception {
            //1.创建消息生产者producer，并制定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("group1");
            //2.指定Nameserver地址
            producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
            //3.启动producer
            producer.start();
    
    
            List<Message> msgs = new ArrayList<Message>();
    
    
            //4.创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg1 = new Message("BatchTopic", "Tag1", ("Hello World" + 1).getBytes());
            Message msg2 = new Message("BatchTopic", "Tag1", ("Hello World" + 2).getBytes());
            Message msg3 = new Message("BatchTopic", "Tag1", ("Hello World" + 3).getBytes());
    
            msgs.add(msg1);
            msgs.add(msg2);
            msgs.add(msg3);
    
            //5.发送消息
            SendResult result = producer.send(msgs);
            //发送状态
            SendStatus status = result.getSendStatus();
    
            System.out.println("发送结果:" + result);
    
            //线程睡1秒
            TimeUnit.SECONDS.sleep(1);
    
    
            //6.关闭生产者producer
            producer.shutdown();
        }
    
    }

第二种实现方式

package com.anran.mq.mqproductservice.controller;
    
    import com.alibaba.fastjson.JSONObject;
    import com.anran.mq.mqproductservice.listener.SendCallbackListener;
    import com.anran.mq.mqproductservice.model.OrderStep;
    import com.anran.mq.mqproductservice.model.ResponseMsg;
    import com.anran.mq.mqproductservice.util.ListSplitter;
    import org.apache.rocketmq.client.producer.MessageQueueSelector;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.client.producer.SendStatus;
    import org.apache.rocketmq.client.producer.TransactionSendResult;
    import org.apache.rocketmq.common.message.MessageQueue;
    import org.apache.rocketmq.spring.core.RocketMQTemplate;
    import org.apache.rocketmq.spring.support.RocketMQHeaders;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.ArrayList;
    import java.util.List;
    
    @RestController
    @RequestMapping("/mqMessageController")
    public class MqMessageController {
        private static final Logger log = LoggerFactory.getLogger(MqMessageController.class);
    
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
        @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.sync-tag}")
        private String syncTag;
    
        /**
         * 同时发送10个单向消息（真正的批量）
         *
         * @param id 消息
         * @return 结果
         */
        @RequestMapping("/pushBatchMessage.action")
        public ResponseMsg pushBatchMessage(@RequestParam("id") int id) {
            log.info("pushBatchMessage start : " + id);
            // 创建消息集合
            List<Message> messages = new ArrayList<>();
            for (int i = 0; i < 3; i++) {
                String myId = id + "" + i;
                // 处理当前订单唯一标识
                String messageStr = "order id : " + myId;
                Message<String> message = MessageBuilder.withPayload(messageStr)
                        .setHeader(RocketMQHeaders.KEYS, myId)
                        .build();
                messages.add(message);
            }
            // 批量下发消息到broker,不支持消息顺序操作，并且对消息体有大小限制（不超过4M）
            ListSplitter splitter = new ListSplitter(messages, 1024 * 1024 * 4);
            while (splitter.hasNext()) {
                List<Message> listItem = splitter.next();
                rocketMQTemplate.syncSend(syncTag, listItem);
            }
            log.info("pushBatchMessage finish : " + id);
            ResponseMsg msg = new ResponseMsg();
            msg.setSuccessData(null);
            return msg;
        }
    }

## 7.7过滤消息（在消费时根据消费者的过滤条件进行消息筛选，tag过滤很简单不赘述，只介绍sql过滤） ##

第一种实现方式

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.client.producer.SendStatus;
    import org.apache.rocketmq.common.message.Message;
    
    import java.util.concurrent.TimeUnit;
    
    public class Producer {
    
        public static void main(String[] args) throws Exception {
            //1.创建消息生产者producer，并制定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("group1");
            //2.指定Nameserver地址
            producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
            //3.启动producer
            producer.start();
    
            for (int i = 0; i < 10; i++) {
                //4.创建消息对象，指定主题Topic、Tag和消息体
                /**
                 * 参数一：消息主题Topic
                 * 参数二：消息Tag
                 * 参数三：消息内容
                 */
                Message msg = new Message("FilterSQLTopic", "Tag1", ("Hello World" + i).getBytes());
    
                msg.putUserProperty("i", String.valueOf(i));
    
                //5.发送消息
                SendResult result = producer.send(msg);
                //发送状态
                SendStatus status = result.getSendStatus();
    
                System.out.println("发送结果:" + result);
    
                //线程睡1秒
                TimeUnit.SECONDS.sleep(2);
            }
    
            //6.关闭生产者producer
            producer.shutdown();
        }
    
    }

第二种实现方式

***在使用顺序消息时broker配置文件必须设置：enablePropertyFilter=true，否则会在设置SQL过滤时出现启动失败，提示：当前broker不支持***

## 7.8事务消息（消息在下发到broker之后需要等待生产者的后续逻辑操作完成之后，才能进行下发或者消息删除） ##

第一种实现方式

import org.apache.commons.lang3.StringUtils;
    import org.apache.rocketmq.client.producer.*;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.common.message.MessageExt;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * 发送同步消息
     */
    public class Producer {
    
        public static void main(String[] args) throws Exception {
            //1.创建消息生产者producer，并制定生产者组名
            TransactionMQProducer producer = new TransactionMQProducer("group5");
            //2.指定Nameserver地址
            producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
    
            //添加事务监听器
            producer.setTransactionListener(new TransactionListener() {
                /**
                 * 在该方法中执行本地事务
                 * @param msg
                 * @param arg
                 * @return
                 */
                @Override
                public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                    if (StringUtils.equals("TAGA", msg.getTags())) {
                        return LocalTransactionState.COMMIT_MESSAGE;
                    } else if (StringUtils.equals("TAGB", msg.getTags())) {
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    } else if (StringUtils.equals("TAGC", msg.getTags())) {
                        return LocalTransactionState.UNKNOW;
                    }
                    return LocalTransactionState.UNKNOW;
                }
    
                /**
                 * 该方法时MQ进行消息事务状态回查
                 * @param msg
                 * @return
                 */
                @Override
                public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                    System.out.println("消息的Tag:" + msg.getTags());
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
            });
    
            //3.启动producer
            producer.start();
    
            String[] tags = {"TAGA", "TAGB", "TAGC"};
    
            for (int i = 0; i < 3; i++) {
                //4.创建消息对象，指定主题Topic、Tag和消息体
                /**
                 * 参数一：消息主题Topic
                 * 参数二：消息Tag
                 * 参数三：消息内容
                 */
                Message msg = new Message("TransactionTopic", tags[i], ("Hello World" + i).getBytes());
                //5.发送消息
                SendResult result = producer.sendMessageInTransaction(msg, null);
                //发送状态
                SendStatus status = result.getSendStatus();
    
                System.out.println("发送结果:" + result);
    
                //线程睡1秒
                TimeUnit.SECONDS.sleep(2);
            }
    
            //6.关闭生产者producer
            //producer.shutdown();
        }
    }

第二种实现方式

package com.anran.mq.mqproductservice.listener;
    
    import com.alibaba.fastjson.JSONObject;
    import lombok.RequiredArgsConstructor;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
    import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
    import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
    import org.apache.rocketmq.spring.support.RocketMQHeaders;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.MessageHeaders;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * rocketmq异步回调监听
     */
    @Slf4j
    @RocketMQTransactionListener
    @RequiredArgsConstructor(onConstructor = @__(@Autowired))
    public class TransactionListener implements RocketMQLocalTransactionListener {
        private static final Map<String, RocketMQLocalTransactionState> TRANSACTION_STATE_MAP = new HashMap<>();
    
        /**
         * 处理本地事务
         * @param message
         * @param o
         * @return
         */
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
            log.info("执行本地事务");
            MessageHeaders headers = message.getHeaders();
            //获取事务ID
            String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
            TRANSACTION_STATE_MAP.put(transactionId, RocketMQLocalTransactionState.UNKNOWN);
            log.info("transactionId is {}",transactionId);
            try {
                Thread.sleep(10 * 1000l);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            RocketMQLocalTransactionState state = RocketMQLocalTransactionState.ROLLBACK;
            if (Integer.parseInt(transactionId) % 2 == 0) {
                //执行成功，可以提交事务
                state = RocketMQLocalTransactionState.COMMIT;
            }
            log.info("transactionId is {}, state {}",transactionId, state.toString());
            TRANSACTION_STATE_MAP.remove(transactionId);
            return state;
        }
    
        /**
         * 校验事务状态
         * @param message
         * @return
         */
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
            MessageHeaders headers = message.getHeaders();
            //获取事务ID
            String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
            log.info("检查本地事务,事务ID:{}",transactionId);
            RocketMQLocalTransactionState state = TRANSACTION_STATE_MAP.get(transactionId);
            if(null != state){
                return state;
            }
            return RocketMQLocalTransactionState.ROLLBACK;
    
        }
    }

# 8.消费者监听消息方式 #

必须设置消费者组信息和指定自己监听topic信息

## 8.1监听消息模式（messageModel：广播/负载均衡） ##

可选类型

package org.apache.rocketmq.spring.annotation;
    
    public enum MessageModel {
        // 广播模式
        BROADCASTING("BROADCASTING"),
        // 负载均衡
        CLUSTERING("CLUSTERING");
    
        private final String modeCN;
    
        private MessageModel(String modeCN) {
            this.modeCN = modeCN;
        }
    
        public String getModeCN() {
            return this.modeCN;
        }
    }

默认类型：负载均衡（MessageModel.CLUSTERING）

## 8.2处理消息模式（consumeMode：有序/有序） ##

可选类型

package org.apache.rocketmq.spring.annotation;
    
    public enum ConsumeMode {
        // 同时接收传递的任意消息
        CONCURRENTLY,
        // 有序接收传递的消息。 只允许一个线程操作一个broker中的一个队列
        ORDERLY;
    
        private ConsumeMode() {
        }
    }

默认类型：同时接收传递的任意消息（ConsumeMode.CONCURRENTLY）

## 8.3过滤消息模式（selectorType：标签/Sql） ##

可选类型

package org.apache.rocketmq.spring.annotation;
    
    public enum SelectorType {
        // 标签过滤
        TAG,
        // sql语法过滤
        SQL92;
    
        private SelectorType() {
        }
    }

默认类型：标签过滤（SelectorType.TAG）

SQL语法：

RocketMQ只定义了一些基本语法来支持这个特性。你也可以很容易地扩展它。

*  数值比较，比如：**>，>=，<，<=，BETWEEN，=；**
 *  字符比较，比如：**=，<>，IN；**
 *  **IS NULL** 或者 **IS NOT NULL；**
 *  逻辑符号 **AND，OR，NOT；**

常量支持类型为：

*  数值，比如：**123，3.1415；**
 *  字符，比如：**'abc'，必须用单引号包裹起来；**
 *  **NULL**，特殊的常量
 *  布尔值，**TRUE** 或 **FALSE**

## 8.4样例 ##

第一种使用方式

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.common.message.MessageExt;
    import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
    
    import java.util.List;
    
    /**
     * 消息的接受者
     */
    public class Consumer {
    
        public static void main(String[] args) throws Exception {
            //1.创建消费者Consumer，制定消费者组名
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
            //2.指定Nameserver地址
            consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
    
            //设定消费模式：负载均衡|广播模式
            consumer.setMessageModel(MessageModel.BROADCASTING);
            //sql过滤
            consumer.subscribe("FilterSQLTopic", MessageSelector.bySql("i>5"));
    
            //3.设置回调函数，处理消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {
    
                //接受消息内容
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
    //        //顺序消息监听
    //        consumer.registerMessageListener(new MessageListenerOrderly() {
    //
    //            @Override
    //            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
    //                for (MessageExt msg : msgs) {
    //                    System.out.println("线程名称：【" + Thread.currentThread().getName() + "】:" + new String(msg.getBody()));
    //                }
    //                return ConsumeOrderlyStatus.SUCCESS;
    //            }
    //        });
            //4.启动消费者consumer
            consumer.start();
        }
    }

第二种使用方式

package com.anran.mq.mqconsumer1service.consumer;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.spring.annotation.ConsumeMode;
    import org.apache.rocketmq.spring.annotation.MessageModel;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.annotation.SelectorType;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.springframework.stereotype.Component;
    
    /**
     * 设置消息监听
     * 1.监听组(consumerGroup)：监听topic(topic)：监听tag(selectorExpression)(默认监听topic下所有)
     * 2.监听消费模式(messageModel):默认负载均衡：CLUSTERING（每一个消息只发给一个消费者）、广播模式：BROADCASTING（发送给所有消费者）
     * 3.设置顺序消息处理模式(consumeMode)（默认是所有线程可以处理同一个消息队列（ConsumeMode.CONCURRENTLY）,当前消息没有线程在执行时其他线程才能够执行（ConsumeMode.ORDERLY）。
     *   ps:一个线程顺序执行一个队列表时消息监听必须使用负载均衡messageModel = MessageModel.BROADCASTING）
     * 4.设置过滤tags类型：默认时tag（SelectorType.TAG）,可以修改为sql语法(SelectorType.SQL92)：
     */
    @Component
    @RocketMQMessageListener(consumerGroup = "${rocketmq.consumer.group}", topic = "${rocketmq.consumer.topic}",
            selectorExpression = "${rocketmq.consumer.tags}", messageModel = MessageModel.CLUSTERING,
            consumeMode = ConsumeMode.ORDERLY, selectorType = SelectorType.SQL92)
    @Slf4j
    public class Consumer implements RocketMQListener<String> {
        @Override
        public void onMessage(String message) {
            log.info(message);
        }
    }

# 9.备注 #

第一种是参考的别人的代码，本人使用的是第二种方式实现，

具体实现可参考链接：

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70]: /images/20221024/acecc719e48f422bb585dd506e17df7d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 1]: /images/20221024/282d1f73bc83470580e899d3e1940cb5.png
[gif.latex]: https://private.codecogs.com/gif.latex?
[https_blog.csdn.net_qq_26154077_article_details_111013842]: https://blog.csdn.net/qq_26154077/article/details/111013842