RocketMQ重試機制和消息冪等

時間  2020-02-21
標籤 rocketmq 重試 機制 消息 欄目 應用數學 简体版
原文   原文鏈接

1、重試機制

　　因爲MQ常常處於複雜的分佈式系統中，考慮網絡波動，服務宕機，程序異常因素，頗有可能出現消息發送或者消費失敗的問題。所以，消息的重試就是全部MQ中間件必須考慮到的一個關鍵點。若是沒有消息重試，就可能產生消息丟失的問題，可能對系統產生很大的影響。因此，秉承寧肯多發消息，也不可丟失消息的原則，大部分MQ都對消息重試提供了很好的支持。java

　　MQ 消費者的消費邏輯失敗時，能夠經過設置返回狀態達到消息重試的結果。redis

　　MQ 消息重試只針對集羣消費方式生效；廣播方式不提供失敗重試特性，即消費失敗後，失敗消息再也不重試，繼續消費新的消息。網絡

一、模擬異常

消費者：併發

package com.zn.retry;

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;

/**
 * RocketMQ重試機制消費者
 */
public class RetryConsumer {
    public static void main(String[] args) throws MQClientException {
        //建立消費者
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
        //設置NameServer地址
        consumer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
        //設置實例名稱
        consumer.setInstanceName("consumer");
        //訂閱topic
        consumer.subscribe("itmayiedu-topic","TagA");

        //監聽消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //獲取消息
                for (MessageExt messageExt:list){
                    System.out.println(messageExt.getMsgId()+"---"+new String(messageExt.getBody()));
                }
                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!");
    }
}

控制檯效果：分佈式

　　

二、模擬網絡延遲

package com.zn.retry;

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;

/**
 * RocketMQ重試機制消費者
 */
public class RetryConsumer {
    public static void main(String[] args) throws MQClientException {
        //建立消費者
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
        //設置NameServer地址
        consumer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
        //設置實例名稱
        consumer.setInstanceName("consumer");
        //訂閱topic
        consumer.subscribe("itmayiedu-topic","TagA");

        //監聽消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //獲取消息
                for (MessageExt messageExt:list){
                    System.out.println(messageExt.getMsgId()+"---"+new String(messageExt.getBody()));
                }
                try {
                    //網絡延遲
                    Thread.sleep(600000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                //消費成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //啓動消費者
        consumer.start();
        System.out.println("Consumer Started!");
    }
}

2、消息冪等

一、在什麼狀況下會發生RocketMQ的消息重複消費

　　 ①、當系統的調用鏈路比較長的時候，好比系統A調用系統B，系統B再把消息發送到RocketMQ中，在系統A調用系統B的時候，若是系統B處理成功，可是遲遲沒有將調用成功的結果返回給系統A的時候，系統A就會嘗試從新發起請求給系統B，形成系統B重複處理，發起多條消息給RocketMQ形成重複消費；ide

　　②、在系統B發送給RocketMQ的時候，也有可能會發生和上面同樣的問題，消息發送超時，節骨系統B重試，致使RocketMQ接收到了重讀消息；高併發

　　③、當RocketMQ成功接收到消息，並將消息交給消費者處理，若是消費者消費完成後還沒來得及提交offset給RocketMQ，本身宕機或者重啓了，那麼RocketMQ沒有接收到offset，就會認爲消費失敗了，會重發消息給消費者再次消費；性能

二、如何解決消息的重複消費

　　經過冪等性來保證，只要保證重複消息不對結果產生影響，就完美地解決這個問題。spa

在生產者端保證冪等性，一下兩種方式：code

　　①、RocketMQ支持消息查詢的功能，只要去RocketMQ查詢一下是否已經發送過該條消息就能夠了，不存在則發送，存在則不發送；

　　②、引入Redis，在發送消息到RocketMQ成功以後，向Redis中插入一條數據，若是發送重試，則先去Redis查詢一個該條消息是否已經發送過了，存在的話就不重複發送消息了；

　　方法一：RocketMQ消息查詢的性能不是特別好，若是在高併發的場景下，每條消息在發送到RocketMQ時都去查詢一下，可能會影響接口的性能；

　　方法二：在一些極端的場景下，Redis也沒法保證消息發送成功以後，就必定能寫入Redis成功，好比寫入消息成功而Redis此時宕機，那麼再次查詢Redis判斷消息是否已經發送過，是沒法獲得正確結果的；

三、生產者

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;

/**
 * 消息冪等生產者
 */
public class IdempotentProvider {
    public static void main(String[] args) throws MQClientException, InterruptedException, RemotingException, MQBrokerException {
        //建立一個生產者
        DefaultMQProducer producer=new DefaultMQProducer("rmq-group");
        //設置NameServer地址
        producer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
        //設置生產者實例名稱
        producer.setInstanceName("producer");
        //啓動生產者
        producer.start();

            //發送消息
            for (int i=1;i<=1;i++){
                //模擬網絡延遲，每秒發送一次MQ
                Thread.sleep(1000);
                //建立消息，topic主題名稱  tags臨時值表明小分類， body表明消息體
                Message message=new Message("itmayiedu-topic03","TagA",("itmayiedu-"+i).getBytes());
                //消息的惟一標識
                message.setKeys("訂單消息："+i);
                //發送消息
                SendResult sendResult=producer.send(message);
                System.out.println("信息冪等問題來了："+sendResult.toString());
            }
        producer.shutdown();
    }
}

四、消費者

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 {
        //建立消費者
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
        //設置NameServer地址
        consumer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
        //設置實例名稱
        consumer.setInstanceName("consumer");
        //訂閱topic
        consumer.subscribe("itmayiedu-topic03","TagA");

        //監聽消息
        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)) {
                            // 無需繼續重試。
                            System.out.println("key:"+key+",已經消費，無需重試...");
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        //RocketMQ因爲是集羣環境，因此產生的消息ID可能會重複
                        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!");
    }
}

五、控制檯效果

　　

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程