RocketMQ重試機(jī)制及消息冪代碼實(shí)例解析
這篇文章主要介紹了RocketMQ重試機(jī)制及消息冪代碼實(shí)例解析,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下
一.重試機(jī)制
1.由于MQ經(jīng)常處于復(fù)雜的分布式系統(tǒng)中，考慮網(wǎng)絡(luò)波動，服務(wù)宕機(jī)，程序異常因素，很有可能出現(xiàn)消息發(fā)送或者消費(fèi)失敗的問題。因此，消息的重試就是所有MQ中間件必須考慮到的一個關(guān)鍵點(diǎn)。如果沒有消息重試，就可能產(chǎn)生消息丟失的問題，可能對系統(tǒng)產(chǎn)生很大的影響。所以，秉承寧可多發(fā)消息，也不可丟失消息的原則，大部分MQ都對消息重試提供了很好的支持。
2.RocketMQ為了使用者封裝了消息重試的處理流程，無需開發(fā)人員手動處理。RocketMQ支持了生產(chǎn)端和消費(fèi)端兩類重試機(jī)制。
模擬異常
Consumer端消息消費(fèi)兩種狀態(tài)：
package com.alibaba.rocketmq.client.consumer.listener;
public enum ConsumeConcurrentlyStatus {
CONSUME_SUCCESS,
RECONSUME_LATER;
private ConsumeConcurrentlyStatus() {
}
}
一個是成功（CONSUME_SUCCESS），一個是失敗&重試（RECONSUME_LATER）;
Consumer為了保證消息消費(fèi)成功，只有使用方明確表示消費(fèi)成功，返回CONSUME_SUCCESS，RocketMQ才會認(rèn)為消息消費(fèi)成功。
如果消息消費(fèi)失敗，只要返回ConsumeConcurrentlyStatus.RECONSUME_LATER,RocketMQ就會認(rèn)為消息消費(fèi)失敗了，需要重新投遞。
1.出現(xiàn)異常
package com.wn.consumer;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.List;
public class MQConsumer {
public static void main(String[] args) throws MQClientException {
//創(chuàng)建消費(fèi)者
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
//設(shè)置NameServer地址
consumer.setNamesrvAddr("192.168.138.187:9876;192.168.138.188:9876");
//設(shè)置消費(fèi)者實(shí)例名稱
consumer.setInstanceName("consumer");
//訂閱topic
consumer.subscribe("wn02","TagA");
//監(jiān)聽消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
//獲取消息
for (MessageExt msg:list){
System.out.println(msg.getMsgId()+"---"+new String(msg.getBody()));
}
try {
int i=1/0;
}catch (Exception e){
e.printStackTrace();
//需要重試
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}//消息成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started...");
}
}
2.網(wǎng)絡(luò)延遲
package com.wn.consumer;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.List;
public class MQConsumer {
public static void main(String[] args) throws MQClientException {
//創(chuàng)建消費(fèi)者
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
//設(shè)置NameServer地址
consumer.setNamesrvAddr("192.168.138.187:9876;192.168.138.188:9876");
//設(shè)置消費(fèi)者實(shí)例名稱
consumer.setInstanceName("consumer");
//訂閱topic
consumer.subscribe("wn03","TagA");
//監(jiān)聽消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
//獲取消息
for (MessageExt msg:list){
System.out.println(msg.getMsgId()+"---"+new String(msg.getBody()));
}
//網(wǎng)絡(luò)延遲
try {
Thread.sleep(600000);
} catch (InterruptedException e) {
e.printStackTrace();
}
//消息成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started...");
}
}
二、消息冪等
1、在什么情況下會發(fā)生RocketMQ的消息重復(fù)消費(fèi)
①、當(dāng)系統(tǒng)的調(diào)用鏈路比較長的時候，比如系統(tǒng)A調(diào)用系統(tǒng)B，系統(tǒng)B再把消息發(fā)送到RocketMQ中，在系統(tǒng)A調(diào)用系統(tǒng)B的時候，如果系統(tǒng)B處理成功，但是遲遲沒有將調(diào)用成功的結(jié)果返回給系統(tǒng)A的時候，系統(tǒng)A就會嘗試重新發(fā)起請求給系統(tǒng)B，造成系統(tǒng)B重復(fù)處理，發(fā)起多條消息給RocketMQ造成重復(fù)消費(fèi)；
②、在系統(tǒng)B發(fā)送給RocketMQ的時候，也有可能會發(fā)生和上面一樣的問題，消息發(fā)送超時，節(jié)骨系統(tǒng)B重試，導(dǎo)致RocketMQ接收到了重讀消息；
?、?、當(dāng)RocketMQ成功接收到消息，并將消息交給消費(fèi)者處理，如果消費(fèi)者消費(fèi)完成后還沒來得及提交offset給RocketMQ，自己宕機(jī)或者重啟了，那么RocketMQ沒有接收到offset，就會認(rèn)為消費(fèi)失敗了，會重發(fā)消息給消費(fèi)者再次消費(fèi)；
2、如何解決消息的重復(fù)消費(fèi)
通過冪等性來保證，只要保證重復(fù)消息不對結(jié)果產(chǎn)生影響，就完美地解決這個問題。
在生產(chǎn)者端保證冪等性，一下兩種方式：
?、?、RocketMQ支持消息查詢的功能，只要去RocketMQ查詢一下是否已經(jīng)發(fā)送過該條消息就可以了，不存在則發(fā)送，存在則不發(fā)送；
②、引入Redis，在發(fā)送消息到RocketMQ成功之后，向Redis中插入一條數(shù)據(jù)，如果發(fā)送重試，則先去Redis查詢一個該條消息是否已經(jīng)發(fā)送過了，存在的話就不重復(fù)發(fā)送消息了；
方法一：RocketMQ消息查詢的性能不是特別好，如果在高并發(fā)的場景下，每條消息在發(fā)送到RocketMQ時都去查詢一下，可能會影響接口的性能；
方法二：在一些極端的場景下，Redis也無法保證消息發(fā)送成功之后，就一定能寫入Redis成功，比如寫入消息成功而Redis此時宕機(jī)，那么再次查詢Redis判斷消息是否已經(jīng)發(fā)送過，是無法得到正確結(jié)果的；
3、生產(chǎn)者
package com.zn.idempotent;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
/**
* 消息冪等生產(chǎn)者
*/
public class IdempotentProvider {
public static void main(String[] args) throws MQClientException, InterruptedException, RemotingException, MQBrokerException {
//創(chuàng)建一個生產(chǎn)者
DefaultMQProducer producer=new DefaultMQProducer("rmq-group");
//設(shè)置NameServer地址
producer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
//設(shè)置生產(chǎn)者實(shí)例名稱
producer.setInstanceName("producer");
//啟動生產(chǎn)者
producer.start();
//發(fā)送消息
for (int i=1;i<=1;i++){
//模擬網(wǎng)絡(luò)延遲，每秒發(fā)送一次MQ
Thread.sleep(1000);
//創(chuàng)建消息，topic主題名稱 tags臨時值代表小分類， body代表消息體
Message message=new Message("itmayiedu-topic03","TagA",("itmayiedu-"+i).getBytes());
//消息的唯一標(biāo)識
message.setKeys("訂單消息："+i);
//發(fā)送消息
SendResult sendResult=producer.send(message);
System.out.println("信息冪等問題來了："+sendResult.toString());
}
producer.shutdown();
}
}
4、消費(fèi)者
package com.zn.idempotent;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.LogManager;
import java.util.logging.Logger;
/**
* 消息冪等消費(fèi)者
*/
public class IdempotentConsumer {
static private Map<String, Object> logMap = new HashMap<>();
public static void main(String[] args) throws MQClientException {
//創(chuàng)建消費(fèi)者
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
//設(shè)置NameServer地址
consumer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
//設(shè)置實(shí)例名稱
consumer.setInstanceName("consumer");
//訂閱topic
consumer.subscribe("itmayiedu-topic03","TagA");
//監(jiān)聽消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
String key=null;
String msgId=null;
for (MessageExt messageExt:list){
key=messageExt.getKeys();
//判讀redis中有沒有當(dāng)前消息key
if (logMap.containsKey(key)) {
// 無需繼續(xù)重試。
System.out.println("key:"+key+",已經(jīng)消費(fèi)，無需重試...");
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
//RocketMQ由于是集群環(huán)境，所以產(chǎn)生的消息ID可能會重復(fù)
msgId = messageExt.getMsgId();
System.out.println("key:" + key + ",msgid:" + msgId + "---" + new String(messageExt.getBody()));
//將當(dāng)前key保存在redis中
logMap.put(messageExt.getKeys(),messageExt);
}
try {
int i=5/0;
}catch (Exception e){
e.printStackTrace();
//人工補(bǔ)償
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//啟動消費(fèi)者
consumer.start();
System.out.println("Consumer Started!");
}
}
以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
SpringBoot整合EasyExcel?3.x的完整示例
EasyExcel 是一個基于 Java 的、快速、簡潔、解決大文件內(nèi)存溢出的 Excel 處理工具，它能讓你在不用考慮性能、內(nèi)存的等因素的情況下，快速完成 Excel 的讀、寫等功能，這篇文章主要介紹了SpringBoot整合EasyExcel3.x的過程,需要的朋友可以參考下2023-07-07
本文主要介紹了Eclipse中自動添加注釋的兩種方法。具有很好的參考價值，下面跟著小編一起來看下吧2017-02-02
JDK1.8源碼下載及idea2021導(dǎo)入jdk1.8源碼的詳細(xì)步驟
這篇文章主要介紹了JDK1.8源碼下載及idea2021導(dǎo)入jdk1.8源碼的詳細(xì)步驟,在文章開頭就給大家分享了JDK1.8源碼下載地址和下載步驟，告訴大家idea2021.1.3導(dǎo)入JDK1.8源碼步驟，需要的朋友可以參考下2022-11-11
SpringBoot程序的打包與運(yùn)行的實(shí)現(xiàn)
本文主要介紹了SpringBoot程序的打包與運(yùn)行的實(shí)現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2022-06-06
關(guān)于pytorch相關(guān)部分矩陣變換函數(shù)的問題分析
這篇文章主要介紹了pytorch相關(guān)部分矩陣變換函數(shù),包括tensor維度順序變換BCHW順序的調(diào)整，矩陣乘法相關(guān)函數(shù),矩陣乘,點(diǎn)乘,求取矩陣對角線元素或非對角線元素的問題，本文給大家介紹的非常詳細(xì)，需要的朋友可以參考下2022-03-03
java數(shù)字和中文算數(shù)驗(yàn)證碼的實(shí)現(xiàn)
這篇文章主要介紹了java數(shù)字和中文算數(shù)驗(yàn)證碼的實(shí)現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-07-07
自定義MyBatis攔截器可以在方法執(zhí)行前后插入自己的邏輯，這非常有利于擴(kuò)展和定制 MyBatis 的功能，本篇文章實(shí)現(xiàn)自定義一個攔截器去改變要插入或者查詢的數(shù)據(jù)源?,需要的朋友可以參考下2023-10-10
這篇文章主要介紹了Java定義泛型接口和類的方法,結(jié)合實(shí)例形式分析了泛型相關(guān)概念、原理及泛型接口與類的定義實(shí)現(xiàn)方法,需要的朋友可以參考下2019-08-08
Java?深入理解創(chuàng)建型設(shè)計(jì)模式之原型模式
原型（Prototype）模式的定義如下：用一個已經(jīng)創(chuàng)建的實(shí)例作為原型，通過復(fù)制該原型對象來創(chuàng)建一個和原型相同或相似的新對象。在這里，原型實(shí)例指定了要創(chuàng)建的對象的種類。用這種方式創(chuàng)建對象非常高效，根本無須知道對象創(chuàng)建的細(xì)節(jié)2022-02-02
java實(shí)現(xiàn)后臺數(shù)據(jù)顯示在前端
這篇文章主要為大家詳細(xì)介紹了java實(shí)現(xiàn)后臺數(shù)據(jù)顯示在前端，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2020-02-02