這篇文章主要介紹RocketMQ重試機制的示例分析,文中介紹的非常詳細,具有一定的參考價值,感興趣的小伙伴們一定要看完!
公司主營業(yè)務(wù):網(wǎng)站設(shè)計制作、做網(wǎng)站、移動網(wǎng)站開發(fā)等業(yè)務(wù)。幫助企業(yè)客戶真正實現(xiàn)互聯(lián)網(wǎng)宣傳,提高企業(yè)的競爭能力。成都創(chuàng)新互聯(lián)公司是一支青春激揚、勤奮敬業(yè)、活力青春激揚、勤奮敬業(yè)、活力澎湃、和諧高效的團隊。公司秉承以“開放、自由、嚴謹、自律”為核心的企業(yè)文化,感謝他們對我們的高要求,感謝他們從不同領(lǐng)域給我們帶來的挑戰(zhàn),讓我們激情的團隊有機會用頭腦與智慧不斷的給客戶帶來驚喜。成都創(chuàng)新互聯(lián)公司推出馬關(guān)免費做網(wǎng)站回饋大家。
一.重試機制
1.由于MQ經(jīng)常處于復(fù)雜的分布式系統(tǒng)中,考慮網(wǎng)絡(luò)波動,服務(wù)宕機,程序異常因素,很有可能出現(xiàn)消息發(fā)送或者消費失敗的問題。因此,消息的重試就是所有MQ中間件必須考慮到的一個關(guān)鍵點。如果沒有消息重試,就可能產(chǎn)生消息丟失的問題,可能對系統(tǒng)產(chǎn)生很大的影響。所以,秉承寧可多發(fā)消息,也不可丟失消息的原則,大部分MQ都對消息重試提供了很好的支持。
2.RocketMQ為了使用者封裝了消息重試的處理流程,無需開發(fā)人員手動處理。RocketMQ支持了生產(chǎn)端和消費端兩類重試機制。
模擬異常
Consumer端消息消費兩種狀態(tài):
package com.alibaba.rocketmq.client.consumer.listener; public enum ConsumeConcurrentlyStatus { CONSUME_SUCCESS, RECONSUME_LATER; private ConsumeConcurrentlyStatus() { } }
一個是成功(CONSUME_SUCCESS),一個是失敗&重試(RECONSUME_LATER);
Consumer為了保證消息消費成功,只有使用方明確表示消費成功,返回CONSUME_SUCCESS,RocketMQ才會認為消息消費成功。
如果消息消費失敗,只要返回ConsumeConcurrentlyStatus.RECONSUME_LATER,RocketMQ就會認為消息消費失敗了,需要重新投遞。
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)建消費者 DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group"); //設(shè)置NameServer地址 consumer.setNamesrvAddr("192.168.138.187:9876;192.168.138.188:9876"); //設(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)建消費者 DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group"); //設(shè)置NameServer地址 consumer.setNamesrvAddr("192.168.138.187:9876;192.168.138.188:9876"); //設(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ù)消費
①、當系統(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ù)消費;
②、在系統(tǒng)B發(fā)送給RocketMQ的時候,也有可能會發(fā)生和上面一樣的問題,消息發(fā)送超時,節(jié)骨系統(tǒng)B重試,導(dǎo)致RocketMQ接收到了重讀消息;
③、當RocketMQ成功接收到消息,并將消息交給消費者處理,如果消費者消費完成后還沒來得及提交offset給RocketMQ,自己宕機或者重啟了,那么RocketMQ沒有接收到offset,就會認為消費失敗了,會重發(fā)消息給消費者再次消費;
2、如何解決消息的重復(fù)消費
通過冪等性來保證,只要保證重復(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此時宕機,那么再次查詢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)者實例名稱 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()); //消息的唯一標識 message.setKeys("訂單消息:"+i); //發(fā)送消息 SendResult sendResult=producer.send(message); System.out.println("信息冪等問題來了:"+sendResult.toString()); } producer.shutdown(); } }
4、消費者
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; /** * 消息冪等消費者 */ public class IdempotentConsumer { static private Map<String, Object> logMap = new HashMap<>(); public static void main(String[] args) throws MQClientException { //創(chuàng)建消費者 DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group"); //設(shè)置NameServer地址 consumer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876"); //設(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中有沒有當前消息key if (logMap.containsKey(key)) { // 無需繼續(xù)重試。 System.out.println("key:"+key+",已經(jīng)消費,無需重試..."); 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())); //將當前key保存在redis中 logMap.put(messageExt.getKeys(),messageExt); } try { int i=5/0; }catch (Exception e){ e.printStackTrace(); //人工補償 return ConsumeConcurrentlyStatus.RECONSUME_LATER; } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); //啟動消費者 consumer.start(); System.out.println("Consumer Started!"); } }
以上是“RocketMQ重試機制的示例分析”這篇文章的所有內(nèi)容,感謝各位的閱讀!希望分享的內(nèi)容對大家有幫助,更多相關(guān)知識,歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道!
當前名稱:RocketMQ重試機制的示例分析
分享URL:http://bm7419.com/article4/pcieie.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供商城網(wǎng)站、移動網(wǎng)站建設(shè)、網(wǎng)站制作、企業(yè)網(wǎng)站制作、品牌網(wǎng)站建設(shè)、企業(yè)建站
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時需注明來源: 創(chuàng)新互聯(lián)