RocketMQ总结（持久化，重发机制，分布式事务，监控机制，顺序消息，重复消费等等）

我就是我 2022-03-07 11:46 351阅读 0赞

### 官网[http://rocketmq.apache.org/docs/code-guidelines/][http_rocketmq.apache.org_docs_code-guidelines] ###

### windows安装RocketMQ ###

在官网下载二进制包

[http://rocketmq.apache.org/release\_notes/release-notes-4.2.0/][http_rocketmq.apache.org_release_notes_release-notes-4.2.0]

![20190316220100874.PNG][]

解压

配置环境变量

![2019031622064674.PNG][]

### 启动RocketMQ ###

在bin目录下，

D:\\RocketMQ\\bin\\mqnamesrv.cmd -n localhost:9876

启动经纪人  
D:\\RocketMQ\\bin\\mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true

不要关闭cmd窗口

### 安装RocketMQ监控平台（console） ###

下载[https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console][https_github.com_apache_rocketmq-externals_tree_master_rocketmq-console]，

配置下项目里面参数

server.port=8081
    //配置监控端口

rocketmq.config.namesrvAddr=127.0.0.1:9876
    //配置NameServer地址

然后D:\\RocketMQ\\rocketmq-externals-master\\rocketmq-console>mvn clean package -Dmaven.test.skip=true

里面会有一些包加载不进去，建议去maven搜索，导入进去

启动java -jar rocketmq-console-ng-1.0.0.jar

访问[http://localhost:8081/][http_localhost_8081]就可以看到界面

方便大家，将jar上传：链接: https://pan.baidu.com/s/1WICnYnnYKY1wVsiUx4i5AA 提取码: irbk 复制这段内容后打开百度网盘手机App，操作更方便哦

### RocketMQ解决顺序问题以及消费重复问题？ ###

[https://dbaplus.cn/news-21-1123-1.html][https_dbaplus.cn_news-21-1123-1.html]

很详细，总结一下：

### 顺序问题 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70][]

清晰把，既然有顺序，必须M1消费完，再消费M2,那么如果保证m1被消费完呢？需要一个响应，ok之后再发送M2消息

将message按照一定规则存储到Queue队列中，这个时候是有顺序的，通过MessageQueueSelector配置，代码：

producer.send(msg, new MessageQueueSelector() {
    
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        System.out.println("arg{}"+arg);
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        //返回选中的队列
                        return mqs.get(index);
                    }
                }, orderId);

配置客户端监听收到的消息

//设置消息监听，这里使用并发的监听 MessageListenerConcurrently，还有一个顺序的监听 MessageListenerOrderly，可以查看源码
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    msgs.forEach(msg->{
                        System.out.println("消费消息："+new String(msg.getBody()));
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

转自[https://blog.csdn.net/convict\_eva/article/details/82627225][https_blog.csdn.net_convict_eva_article_details_82627225]

### 重复消费问题 ###

[https://dbaplus.cn/news-21-1123-1.html][https_dbaplus.cn_news-21-1123-1.html]

**RocketMQ****不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你自己在业务端去重**

### 分布式事务 ###

这个之前也有做过类似的[https://blog.csdn.net/weixin\_38336658/article/details/85783573][https_blog.csdn.net_weixin_38336658_article_details_85783573]

执行完本地事务之后，成功发送消息，使用重发机制（如果消费失败的话），如果超过一定次数，变成死信，这个需要人工手动操作，如果消费成功，执行事务，如果事务执行失败，这个也得人工去调，如果你把mq都做到把之前本地事务也回滚了，代码代价很高的。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70 1][]

调用sendMessageInTransation,发送prepare消息，然后执行本地事务，执行完，发送确认消息

### RocketMQ专业术语 ###

[https://blog.csdn.net/tototuzuoquan/article/details/78325192][https_blog.csdn.net_tototuzuoquan_article_details_78325192]

demo

pom.xml

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

配置

package com.example.demo.conf;
    
    /**
     * mq 配置信息
     */
    public class Configure {
    
        /**
         * name server 地址，多个用分号隔开
         */
        public static String NAMESRV_ADDR = "127.0.0.1:9876";
    
        /**
         * broker 组名
         */
        public static String BROKER_GROUP = "test_broker_group";
    
        /**
         * consumer 组名
         */
        public static String CONSUMER_GROUP = "test_consumer_group";
    
    
        /**
         * 消息主题
         */
        public static String MESSAGE_TOPIC ="test_message_topic";
    
        /**
         * 消息key 前缀，消息key一般是业务的主键
         */
        public static String MESSAGE_KEY_PREFIX = "message_key_prefix:";
    
        /**
         * 消息tag
         */
        public static String MESSAGE_TAG_A ="test_tag_a";
        public static String MESSAGE_TAG_B ="test_tag_b";
    
    
        /**
         * 默认编码格式
         */
        public static String DEFAULT_CHARSET="UTF-8";
    
    }

生产者

package com.example.demo.product;
    
    import com.example.demo.conf.Configure;
    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 org.apache.rocketmq.remoting.common.RemotingHelper;
    
    import java.util.List;
    
    
    public class ProducerClient {
        public static void main(String[] args) throws Exception {
            //Instantiate with a producer group name.
            DefaultMQProducer producer = new DefaultMQProducer(Configure.BROKER_GROUP);
            producer.setNamesrvAddr(Configure.NAMESRV_ADDR);
            //Launch the instance.
            producer.start();
    
            String[] tags = new String[] {Configure.MESSAGE_TAG_A,Configure.MESSAGE_TAG_B};
    
            for (int i = 0; i < 100; i++) {
                int orderId = i % 10;
    
                System.out.println("orderId{}"+orderId);
    
                //创建消息
                Message msg = new Message(Configure.MESSAGE_TOPIC, tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
    
                //发送消息，重写选择MessageQueue 方法，把消息写到对应的ConsumerQueue 中
                // orderId 参数传递到内部方法 select arg 参数
                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        System.out.println("arg{}"+arg);
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        //返回选中的队列
                        return mqs.get(index);
                    }
                }, orderId);
    
                System.out.printf("%s%n", sendResult);
            }
            //server shutdown
            producer.shutdown();
        }
    }

消费者

package com.example.demo.comsumer;
    
    import com.example.demo.conf.Configure;
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
    import org.apache.rocketmq.common.message.MessageExt;
    
    import java.util.List;
    
    public class ConsumerClient {
        public static void main(String[] args) throws Exception{
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(Configure.CONSUMER_GROUP);
            consumer.setNamesrvAddr(Configure.NAMESRV_ADDR);
    
            consumer.subscribe(Configure.MESSAGE_TOPIC, Configure.MESSAGE_TAG_A+" || "+Configure.MESSAGE_TAG_B);
    
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                           ConsumeOrderlyContext context) {
                    //手动确认
                    context.setAutoCommit(false);
                    msgs.forEach(m->{
                        System.out.print("host:"+m.getBornHost()+"--");
                        System.out.print("key:"+m.getKeys()+"--");
                        System.out.print("Topic:"+m.getTopic()+"--");
                        System.out.print("QueueId:"+m.getQueueId()+"--");
                        System.out.print("tags:"+m.getTags()+"--");
                        System.out.print("msg:"+new String(m.getBody()));
                        System.out.println();
                    });
                    return ConsumeOrderlyStatus.SUCCESS;
    
                }
            });
    
            consumer.start();
    
            System.out.println("Consumer Started.%n");
        }
    }

### RocketMQ储存（持久化） ###

使用commitLog实现的，调用CommitLog.putMessage(MessageExtBrokerInner msg)将消息以MapperFile形式写入commitlog,进行刷盘 到磁盘，有异步和同步方式

Broker收到的消息都是存储在CommitLog中，CommitLog会对应一个个独立的数据文件。I/O操作一般都是最为耗时的，RocketMQ为了提升I/O效率，使用了NIO中提供的FileChannel.map方法，将文件以mmap形式映射到内存中。

Broker收到消息之后会将消息写入MappedFile的mappedByteBuffer当中，而后再由刷盘实现将内存中变更的数据刷入磁盘当中。在异步刷盘模式下，因为刷盘操作位于独立线程，因此可以获得很好的性能。

转自[http://soliloquize.org/2018/08/25/RocketMQ-CommitLog%E5%88%B7%E7%9B%98%E6%9C%BA%E5%88%B6/][http_soliloquize.org_2018_08_25_RocketMQ-CommitLog_E5_88_B7_E7_9B_98_E6_9C_BA_E5_88_B6]

我发现它是自动恢复的，真叼。\#- ASYNC\_FLUSH 异步刷盘 \#- SYNC\_FLUSH 同步刷盘 flushDiskType=ASYNC\_FLUSH

上面是MQ自带刷盘技术，那如果它在刷盘之前MQ宕机了，怎么办？

根据发送消息返回值进行判断，如果配置重新消费，客户端要进行去重消费。状态码参考[https://blog.csdn.net/weixin\_34405557/article/details/87482261][https_blog.csdn.net_weixin_34405557_article_details_87482261]

持久化方案：

### Message Persistence ###

消息中间件通常采用的几种持久化方式：

1.  持久化到数据库，例如Mysql。
2.  持久化到KV存储，例如levelDB、伯克利DB等KV存储系统。
3.  文件记录形式持久化，例如Kafka，RocketMQ
4.  对内存数据做一个持久化镜像，例如beanstalkd，VisiNotify
5.  (1)、(2)、(3)三种持久化方式都具有将内存队列Buffer进行扩展的能力，(4)只是一个内存的镜像，作用是当Broker挂掉重启后仍然能将之前内存的数据恢复出来。

JMS与CORBA Notification规范没有明确说明如何持久化，但是持久化部分的性能直接决定了整个消息中间件的性能。

RocketMQ充分利用Linux文件系统内存cache来提高性能。

转自[http://jm.taobao.org/2017/01/12/rocketmq-quick-start-in-10-minutes/][http_jm.taobao.org_2017_01_12_rocketmq-quick-start-in-10-minutes]

### RocketMQ重发机制 ###

[https://blog.csdn.net/weixin\_34452850/article/details/82746852][https_blog.csdn.net_weixin_34452850_article_details_82746852]

分为生产者重试，消费者重试

生产者重试：

重试5次

DefaultMQProducer producer = new DefaultMQProducer(Configure.BROKER_GROUP);
            producer.setNamesrvAddr(Configure.NAMESRV_ADDR);
            //重试5次
            producer.setRetryTimesWhenSendFailed(5);

5秒

SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        System.out.println("arg{}"+arg);
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        //返回选中的队列
                        return mqs.get(index);
                    }
                }, orderId,5000L);

客户端重试：

修改重试时间

messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

在配置文件中，Broker.conf配置上面这个参数

另一种是在客户端配置监听器，如果出现超时，业务异常进行返回特定的标识，需要稍后重试

如下：

//逐条消费
                    Message message = msgs.get(0);
                    try {
                        int i = 1 / 0;
                        //模拟业务异常
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } catch (Exception e) {
                        System.out.println("尝试次数:" + ((MessageExt) message).getReconsumeTimes());
                        if (((MessageExt) message).getReconsumeTimes() >= Configure.RetryTime) {
                            //列为死信,记录消息内容，然后人工处理
                            System.out.println("出现业务异常的消息："+new String(message.getBody()) + " " + ((MessageExt) message).getMsgId() + " " + message.getTopic() + " " + message.getTags());
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }

模拟业务异常，会标记成RECONSUME\_LATER,RocketMQ会自己重发，如果重发次数超过一定阈值，进行返回成功，并储存起消息，人工进行处理。

超时重试：

客户端设置超时时间，以分钟为单位

public class ConsumerClient {
        public static void main(String[] args) throws Exception {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(Configure.CONSUMER_GROUP);
            consumer.setNamesrvAddr(Configure.NAMESRV_ADDR);
            // 批量消费,每次拉取10条
            consumer.setConsumeMessageBatchMaxSize(10);
            // 如果非第一次启动，那么按照上次消费的位置继续消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //设置消费超时时间
            consumer.setConsumeTimeout(Configure.TIMEOUT_SECONDS);
            consumer.subscribe(Configure.MESSAGE_TOPIC, Configure.MESSAGE_TAG_A + " || " + Configure.MESSAGE_TAG_B);
    
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
                      
                    //逐条消费
                    Message message = msgs.get(0);
                    
                    try {
                        //int i = 1 / 0;
                        //模拟业务异常
                        Thread.sleep(1000L*60*2);
                        System.out.println("消息内容："+new String(message.getBody()) + " " + ((MessageExt) message).getMsgId() + " " + message.getTopic() + " " + message.getTags());
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } catch (Exception e) {
                        System.out.println("尝试次数:" + ((MessageExt) message).getReconsumeTimes());
                        if (((MessageExt) message).getReconsumeTimes() >= Configure.RetryTime) {
                            //列为死信,记录消息内容，然后人工处理
                            System.out.println("出现异常的消息："+new String(message.getBody()) + " " + ((MessageExt) message).getMsgId() + " " + message.getTopic() + " " + message.getTags());
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
    
    
                    //RECONSUME_LATER 消费失败，需要稍后重新消费
                    //return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            });
    
            consumer.start();
    
            System.out.println("Consumer Started.%n");
        }
    }

/**
     * mq 配置信息
     */
    public class Configure {
    
        /**
         * name server 地址，多个用分号隔开
         */
        public static String NAMESRV_ADDR = "127.0.0.1:9876";
    
        /**
         * broker 组名
         */
        public static String BROKER_GROUP = "test_broker_group";
    
        /**
         * consumer 组名
         */
        public static String CONSUMER_GROUP = "test_consumer_group";
    
    
        /**
         * 消息主题
         */
        public static String MESSAGE_TOPIC = "test_message_topic";
    
        /**
         * 消息key 前缀，消息key一般是业务的主键
         */
        public static String MESSAGE_KEY_PREFIX = "message_key_prefix:";
    
        /**
         * 消息tag
         */
        public static String MESSAGE_TAG_A = "test_tag_a";
        public static String MESSAGE_TAG_B = "test_tag_b";
    
    
        /**
         * 默认编码格式
         */
        public static String DEFAULT_CHARSET = "UTF-8";
    
        /**
         * 重发次数
         */
        public static int RetryTime = 2;
    
        /**
         * 超时时间
         */
        public static long TIMEOUT_SECONDS = 1L;
    
    }

Thread.sleep模拟超时，有两个客户端，经过一分钟之后，我们把这个超时的关掉，发现第二个客户端还会接受到信息，说明brocker对超时信息进行重发，但是包含一些重复的message\_id,所以需要去重

### RocketMQ事务 ###

它保证本地事务的执行完成，先发送Prepare给mq，然后执行本地事务，执行完再发送确认消息给mq

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70 2][]

调用sendMessageInTransation

### RocketMQ消费 ACK返回消费情况 ###

consumer.registerMessageListener(new MessageListenerConcurrently() {  
      
                 @Override  
                 public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,  
                                                                 ConsumeConcurrentlyContext context) {  
                     for (MessageExt messageExt : msgs) {    
                        String messageBody = new String(messageExt.getBody());    
                        System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(
                        		 new Date())+"消费响应：msgId : " + messageExt.getMsgId() + ",  msgBody : " + messageBody);//输出消息内容    
                     }    
                       
                     //返回消费状态  
                     //CONSUME_SUCCESS 消费成功  
                     //RECONSUME_LATER 消费失败，需要稍后重新消费  
                     return ConsumeConcurrentlyStatus.RECONSUME_LATER;  
                 }  
             });

完整代码：[https://github.com/dajitui/RocketMQDemo][https_github.com_dajitui_RocketMQDemo]

[http_rocketmq.apache.org_docs_code-guidelines]: http://rocketmq.apache.org/docs/code-guidelines/
[http_rocketmq.apache.org_release_notes_release-notes-4.2.0]: http://rocketmq.apache.org/release_notes/release-notes-4.2.0/
[20190316220100874.PNG]: /images/20220307/fa25d870ff554229a43891bfce5b8fa7.png
[2019031622064674.PNG]: /images/20220307/a7a138050df5499999f8791fed87cc21.png
[https_github.com_apache_rocketmq-externals_tree_master_rocketmq-console]: https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console
[http_localhost_8081]: http://localhost:8081/#/producer
[https_dbaplus.cn_news-21-1123-1.html]: https://dbaplus.cn/news-21-1123-1.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70]: /images/20220307/55f8068752d74494b92699c2e8a69557.png
[https_blog.csdn.net_convict_eva_article_details_82627225]: https://blog.csdn.net/convict_eva/article/details/82627225
[https_blog.csdn.net_weixin_38336658_article_details_85783573]: https://blog.csdn.net/weixin_38336658/article/details/85783573
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70 1]: /images/20220307/1d00dad493934b0a9083557ffac395f7.png
[https_blog.csdn.net_tototuzuoquan_article_details_78325192]: https://blog.csdn.net/tototuzuoquan/article/details/78325192
[http_soliloquize.org_2018_08_25_RocketMQ-CommitLog_E5_88_B7_E7_9B_98_E6_9C_BA_E5_88_B6]: http://soliloquize.org/2018/08/25/RocketMQ-CommitLog%E5%88%B7%E7%9B%98%E6%9C%BA%E5%88%B6/
[https_blog.csdn.net_weixin_34405557_article_details_87482261]: https://blog.csdn.net/weixin_34405557/article/details/87482261
[http_jm.taobao.org_2017_01_12_rocketmq-quick-start-in-10-minutes]: http://jm.taobao.org/2017/01/12/rocketmq-quick-start-in-10-minutes/
[https_blog.csdn.net_weixin_34452850_article_details_82746852]: https://blog.csdn.net/weixin_34452850/article/details/82746852
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70 2]: /images/20220307/40af0f6492de4c0baead682896812b95.png
[https_github.com_dajitui_RocketMQDemo]: https://github.com/dajitui/RocketMQDemo