RocketMQ批量消费、消息重试、消费模式、刷盘方式

港控/mmm° 2022-05-26 07:06 1227阅读 0赞

转载:  https://blog.csdn.net/u010634288/article/details/56049305

一、Consumer 批量消费

可以通过

\[java\]  [view plain][]   [copy][view plain]

1.  consumer.setConsumeMessageBatchMaxSize(10);//每次拉取10条

这里需要分为2种情况1、Consumer端先启动  2、Consumer端后启动.   正常情况下：应该是Consumer需要先启动

1、Consumer端先启动

Consumer代码如下

\[java\]  [view plain][]   [copy][view plain]

1.  **package** quickstart;  
2.    
3.  **import** java.util.List;  
4.    
5.  **import** com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;  
6.  **import** com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;  
7.  **import** com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;  
8.  **import** com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;  
9.  **import** com.alibaba.rocketmq.client.exception.MQClientException;  
10. **import** com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;  
11. **import** com.alibaba.rocketmq.common.message.MessageExt;  
12.   
13. /\*\* 
14.  \* Consumer，订阅消息 
15.  \*/  
16. **public** **class** Consumer \{  
17.   
18.     **public** **static** **void** main(String\[\] args) **throws** InterruptedException, MQClientException \{  
19.         DefaultMQPushConsumer consumer = **new** DefaultMQPushConsumer("please\_rename\_unique\_group\_name\_4");  
20.         consumer.setNamesrvAddr("192.168.100.145:9876;192.168.100.146:9876");  
21.         consumer.setConsumeMessageBatchMaxSize(10);  
22.         /\*\* 
23.          \* 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br> 
24.          \* 如果非第一次启动，那么按照上次消费的位置继续消费 
25.          \*/  
26.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME\_FROM\_FIRST\_OFFSET);  
27.   
28.         consumer.subscribe("TopicTest", "\*");  
29.   
30.         consumer.registerMessageListener(**new** MessageListenerConcurrently() \{  
31.   
32.             **public** ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) \{  
33.                   
34.                 **try** \{  
35.                     System.out.println("msgs的长度" + msgs.size());  
36.                     System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);  
37.                 \} **catch** (Exception e) \{  
38.                     e.printStackTrace();  
39.                     **return** ConsumeConcurrentlyStatus.RECONSUME\_LATER;  
40.                 \}  
41.                   
42.                   
43.                   
44.                 **return** ConsumeConcurrentlyStatus.CONSUME\_SUCCESS;  
45.             \}  
46.         \});  
47.   
48.         consumer.start();  
49.   
50.         System.out.println("Consumer Started.");  
51.     \}  
52. \}

由于这里是Consumer先启动，所以他回去轮询MQ上是否有订阅队列的消息，由于每次producer插入一条，Consumer就拿一条所以测试结果如下（每次size都是1）：

![Center][]

2、Consumer端后启动，也就是Producer先启动

由于这里是Consumer后启动，所以MQ上也就堆积了一堆数据，Consumer的

consumer.setConsumeMessageBatchMaxSize(10);//每次拉取10条

所以这段代码就生效了测试结果如下（每次size最多是10）：

![Center 1][]

二、消息重试机制：消息重试分为2种1、Producer端重试 2、Consumer端重试

1、Producer端重试

也就是Producer往MQ上发消息没有发送成功，我们可以设置发送失败重试的次数

\[java\]  [view plain][]   [copy][view plain]

1.  **package** quickstart;  
2.    
3.  **import** com.alibaba.rocketmq.client.exception.MQClientException;  
4.  **import** com.alibaba.rocketmq.client.producer.DefaultMQProducer;  
5.  **import** com.alibaba.rocketmq.client.producer.SendResult;  
6.  **import** com.alibaba.rocketmq.common.message.Message;  
7.    
8.  /\*\* 
9.   \* Producer，发送消息 
10.  \*  
11.  \*/  
12. **public** **class** Producer \{  
13.     **public** **static** **void** main(String\[\] args) **throws** MQClientException, InterruptedException \{  
14.         DefaultMQProducer producer = **new** DefaultMQProducer("please\_rename\_unique\_group\_name");  
15.         producer.setNamesrvAddr("192.168.100.145:9876;192.168.100.146:9876");  
16.         producer.setRetryTimesWhenSendFailed(10);//失败的 情况发送10次  
17.         producer.start();  
18.   
19.         **for** (**int** i = 0; i < 1000; i++) \{  
20.             **try** \{  
21.                 Message msg = **new** Message("TopicTest",// topic  
22.                         "TagA",// tag  
23.                         ("Hello RocketMQ " + i).getBytes()// body  
24.                 );  
25.                 SendResult sendResult = producer.send(msg);  
26.                 System.out.println(sendResult);  
27.             \} **catch** (Exception e) \{  
28.                 e.printStackTrace();  
29.                 Thread.sleep(1000);  
30.             \}  
31.         \}  
32.   
33.         producer.shutdown();  
34.     \}  
35. \}

![Center 2][]

2、Consumer端重试

2.1、exception的情况，一般重复16次 10s、30s、1分钟、2分钟、3分钟等等

上面的代码中消费异常的情况返回

return ConsumeConcurrentlyStatus.RECONSUME\_LATER;//重试

正常则返回：

return ConsumeConcurrentlyStatus.CONSUME\_SUCCESS;//成功

\[java\]  [view plain][]   [copy][view plain]

1.  **package** quickstart;  
2.    
3.    
4.  **import** java.util.List;  
5.    
6.  **import** com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;  
7.  **import** com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;  
8.  **import** com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;  
9.  **import** com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;  
10. **import** com.alibaba.rocketmq.client.exception.MQClientException;  
11. **import** com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;  
12. **import** com.alibaba.rocketmq.common.message.MessageExt;  
13.   
14. /\*\* 
15.  \* Consumer，订阅消息 
16.  \*/  
17. **public** **class** Consumer \{  
18.   
19.     **public** **static** **void** main(String\[\] args) **throws** InterruptedException, MQClientException \{  
20.         DefaultMQPushConsumer consumer = **new** DefaultMQPushConsumer("please\_rename\_unique\_group\_name\_4");  
21.         consumer.setNamesrvAddr("192.168.100.145:9876;192.168.100.146:9876");  
22.         consumer.setConsumeMessageBatchMaxSize(10);  
23.         /\*\* 
24.          \* 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br> 
25.          \* 如果非第一次启动，那么按照上次消费的位置继续消费 
26.          \*/  
27.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME\_FROM\_FIRST\_OFFSET);  
28.   
29.         consumer.subscribe("TopicTest", "\*");  
30.   
31.         consumer.registerMessageListener(**new** MessageListenerConcurrently() \{  
32.   
33.             **public** ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) \{  
34.   
35.                 **try** \{  
36.                     // System.out.println("msgs的长度" + msgs.size());  
37.                     System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);  
38.                     **for** (MessageExt msg : msgs) \{  
39.                         String msgbody = **new** String(msg.getBody(), "utf-8");  
40.                         **if** (msgbody.equals("Hello RocketMQ 4")) \{  
41.                             System.out.println("======错误=======");  
42.                             **int** a = 1 / 0;  
43.                         \}  
44.                     \}  
45.   
46.                 \} **catch** (Exception e) \{  
47.                     e.printStackTrace();  
48.                     **if**(msgs.get(0).getReconsumeTimes()==3)\{  
49.                         //记录日志  
50.                           
51.                         **return** ConsumeConcurrentlyStatus.CONSUME\_SUCCESS;// 成功  
52.                     \}**else**\{  
53.                           
54.                     **return** ConsumeConcurrentlyStatus.RECONSUME\_LATER;// 重试  
55.                     \}  
56.                 \}  
57.   
58.                 **return** ConsumeConcurrentlyStatus.CONSUME\_SUCCESS;// 成功  
59.             \}  
60.         \});  
61.   
62.         consumer.start();  
63.   
64.         System.out.println("Consumer Started.");  
65.     \}  
66. \}

打印结果：

![Center 3][]

![Center 4][]

假如超过了多少次之后我们可以让他不再重试记录 日志。

if(msgs.get(0).getReconsumeTimes()==3)\{  
//记录日志   
return ConsumeConcurrentlyStatus.CONSUME\_SUCCESS;// 成功  
\}

2.2超时的情况，这种情况MQ会无限制的发送给消费端。

就是由于网络的情况，MQ发送数据之后，Consumer端并没有收到导致超时。也就是消费端没有给我返回return ConsumeConcurrentlyStatus.CONSUME\_SUCCESS;这样的就认为没有到达Consumer端。

这里模拟Producer只发送一条数据。consumer端暂停1分钟并且不发送接收状态给MQ

\[java\]  [view plain][]   [copy][view plain]

1.  **package** model;  
2.    
3.  **import** java.util.List;  
4.    
5.  **import** com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;  
6.  **import** com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;  
7.  **import** com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;  
8.  **import** com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;  
9.  **import** com.alibaba.rocketmq.client.exception.MQClientException;  
10. **import** com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;  
11. **import** com.alibaba.rocketmq.common.message.MessageExt;  
12.   
13. /\*\* 
14.  \* Consumer，订阅消息 
15.  \*/  
16. **public** **class** Consumer \{  
17.   
18.     **public** **static** **void** main(String\[\] args) **throws** InterruptedException, MQClientException \{  
19.         DefaultMQPushConsumer consumer = **new** DefaultMQPushConsumer("message\_consumer");  
20.         consumer.setNamesrvAddr("192.168.100.145:9876;192.168.100.146:9876");  
21.         consumer.setConsumeMessageBatchMaxSize(10);  
22.         /\*\* 
23.          \* 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br> 
24.          \* 如果非第一次启动，那么按照上次消费的位置继续消费 
25.          \*/  
26.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME\_FROM\_FIRST\_OFFSET);  
27.   
28.         consumer.subscribe("TopicTest", "\*");  
29.   
30.         consumer.registerMessageListener(**new** MessageListenerConcurrently() \{  
31.   
32.             **public** ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) \{  
33.   
34.                 **try** \{  
35.   
36.                     // 表示业务处理时间  
37.                     System.out.println("=========开始暂停===============");  
38.                     Thread.sleep(60000);  
39.   
40.                     **for** (MessageExt msg : msgs) \{  
41.                         System.out.println(" Receive New Messages: " + msg);  
42.                     \}  
43.   
44.                 \} **catch** (Exception e) \{  
45.                     e.printStackTrace();  
46.                     **return** ConsumeConcurrentlyStatus.RECONSUME\_LATER;// 重试  
47.                 \}  
48.   
49.                 **return** ConsumeConcurrentlyStatus.CONSUME\_SUCCESS;// 成功  
50.             \}  
51.         \});  
52.   
53.         consumer.start();  
54.   
55.         System.out.println("Consumer Started.");  
56.     \}  
57. \}

![Center 5][]

![Center 6][]

三、消费模式

广播消费：rocketMQ默认是集群消费,我们可以通过在Consumer来支持广播消费

consumer.setMessageModel(MessageModel.BROADCASTING);// 广播消费

\[java\]  [view plain][]   [copy][view plain]

1.  **package** model;  
2.    
3.  **import** java.util.List;  
4.    
5.  **import** com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;  
6.  **import** com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;  
7.  **import** com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;  
8.  **import** com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;  
9.  **import** com.alibaba.rocketmq.client.exception.MQClientException;  
10. **import** com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;  
11. **import** com.alibaba.rocketmq.common.message.MessageExt;  
12. **import** com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;  
13.   
14. /\*\* 
15.  \* Consumer，订阅消息 
16.  \*/  
17. **public** **class** Consumer2 \{  
18.   
19.     **public** **static** **void** main(String\[\] args) **throws** InterruptedException, MQClientException \{  
20.         DefaultMQPushConsumer consumer = **new** DefaultMQPushConsumer("message\_consumer");  
21.         consumer.setNamesrvAddr("192.168.100.145:9876;192.168.100.146:9876");  
22.         consumer.setConsumeMessageBatchMaxSize(10);  
23.         consumer.setMessageModel(MessageModel.BROADCASTING);// 广播消费  
24.       
25.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME\_FROM\_FIRST\_OFFSET);  
26.   
27.         consumer.subscribe("TopicTest", "\*");  
28.   
29.         consumer.registerMessageListener(**new** MessageListenerConcurrently() \{  
30.   
31.             **public** ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) \{  
32.   
33.                 **try** \{  
34.   
35.                     **for** (MessageExt msg : msgs) \{  
36.                         System.out.println(" Receive New Messages: " + msg);  
37.                     \}  
38.   
39.                 \} **catch** (Exception e) \{  
40.                     e.printStackTrace();  
41.                     **return** ConsumeConcurrentlyStatus.RECONSUME\_LATER;// 重试  
42.                 \}  
43.   
44.                 **return** ConsumeConcurrentlyStatus.CONSUME\_SUCCESS;// 成功  
45.             \}  
46.         \});  
47.   
48.         consumer.start();  
49.   
50.         System.out.println("Consumer Started.");  
51.     \}  
52. \}

如果我们有2台节点，Producerw往MQ上写入20条数据 其中Consumer1中拉取了12条 。Consumer2中拉取了8 条，这种情况下，加入Consumer1宕机，那么我们消费数据的时候，只能消费到Consumer2中的8条，Consumer1中的12条已经持久化到中。需要Consumer1回复之后这12条数据才能继续被消费。其实这种先启动producer往MQ上写数据，然后再启动Consumer的情况本来就是违规操作，正确的情况应该是先启动Consumer后再启动producer。

![Center 7][]

异步复制和同步双写主要是主和从的关系。消息需要实时消费的，就需要采用主从模式部署

异步复制:比如这里有一主一从，我们发送一条消息到主节点之后，这样消息就算从producer端发送成功了，然后通过异步复制的方法将数据复制到从节点

同步双写:比如这里有一主一从，我们发送一条消息到主节点之后，这样消息就并不算从producer端发送成功了，需要通过同步双写的方法将数据同步到从节点后， 才算数据发送成功。

四、刷盘方式

同步刷盘：在消息到达MQ后，RocketMQ需要将数据持久化，同步刷盘是指数据到达内存之后，必须刷到commitlog日志之后才算成功，然后返回producer数据已经发送成功。

![Center 8][]

异步刷盘：，同步刷盘是指数据到达内存之后,返回producer说数据已经发送成功。，然后再写入commitlog日志。

![Center 9][]

commitlog：

commitlog就是来存储所有的元信息，包含消息体，类似于Mysql、Oracle的redolog,所以主要有CommitLog在，Consume Queue即使数据丢失，仍然可以恢复出来。

consumequeue：记录数据的位置,以便Consume快速通过consumequeue找到commitlog中的数据

[view plain]: https://blog.csdn.net/u010634288/article/details/56049305#
[Center]: /images/20220526/3fcc56f0f6a342df9ff7cd317c31b45a.png
[Center 1]: /images/20220526/6524e9f566714b809fc9a092b86a4e28.png
[Center 2]: /images/20220526/dde37897e86e4f3a8bc3a1326a1527d2.png
[Center 3]: /images/20220526/598d55e37e0945a99aeba97498b4b3b1.png
[Center 4]: /images/20220526/b4615079cd814fd8b1b5c2db9bec1107.png
[Center 5]: /images/20220526/8354ae3c0adf4edc9be2984cc036ce93.png
[Center 6]: /images/20220526/2592128adf0049c4ad909a4eb59b6c78.png
[Center 7]: /images/20220526/87dad9eb1e014bcba0a568cf34aa5d35.png
[Center 8]: /images/20220526/d488c83b5ae84f479c8a82c8f0ebf95b.png
[Center 9]: /images/20220526/81c674f066a2447095bd2137a8db65fd.png