RabbitMQ冪等性與優先級及惰性詳細全面講解

Posted on by

1. 冪等性

概念

用戶對於同一操作發起的一次請求或者多次請求的結果是一致的，不會因為多次點擊而產生瞭副作用。 舉個最簡單的例子，那就是支付，用戶購買商品後支付，支付扣款成功，但是返回結果的時候網絡異常， 此時錢已經扣瞭，用戶再次點擊按鈕，此時會進行第二次扣款，返回結果成功，用戶查詢餘額發現多扣錢 瞭，流水記錄也變成瞭兩條。在以前的單應用系統中，我們隻需要把數據操作放入事務中即可，發生錯誤立即回滾，但是再響應客戶端的時候也有可能出現網絡中斷或者異常等等

消息重復消費

消費者在消費 MQ 中的消息時，MQ 已把消息發送給消費者，消費者在給 MQ 返回 ack 時網絡中斷， 故 MQ 未收到確認信息，該條消息會重新發給其他的消費者，或者在網絡重連後再次發送給該消費者，但實際上該消費者已成功消費瞭該條消息，造成消費者消費瞭重復的消息。

解決思路

MQ 消費者的冪等性的解決一般使用全局 ID 或者寫個唯一標識比如時間戳 或者 UUID 或者訂單消費者消費 MQ 中的消息也可利用 MQ 的該 id 來判斷，或者可按自己的規則生成一個全局唯一 id，每次消費消息時用該 id 先判斷該消息是否已消費過。

消費端的冪等性保障

在海量訂單生成的業務高峰期，生產端有可能就會重復發生瞭消息，這時候消費端就要實現冪等性， 這就意味著我們的消息永遠不會被消費多次，即使我們收到瞭一樣的消息。

業界主流的冪等性有兩種操作:a. 唯一 ID+指紋碼機制,利用數據庫主鍵去重, b.利用 redis 的原子性去實現

唯一ID+指紋碼機制

指紋碼：我們的一些規則或者時間戳加別的服務給到的唯一信息碼,它並不一定是我們系統生成的，基本都是由我們的業務規則拼接而來，但是一定要保證唯一性，然後就利用查詢語句進行判斷這個 id 是否存在數據庫中，優勢就是實現簡單就一個拼接，然後查詢判斷是否重復；劣勢就是在高並發時，如果是單個數據庫就會有寫入性能瓶頸當然也可以采用分庫分表提升性能，但也不是我們最推薦的方式。

note Redis 原子性

利用 redis 執行 setnx 命令，天然具有冪等性。從而實現不重復消費

2. 優先級隊列

使用場景

在我們系統中有一個訂單催付的場景，我們的客戶在天貓下的訂單，淘寶會及時將訂單推送給我們，如果在用戶設定的時間內未付款那麼就會給用戶推送一條短信提醒，很簡單的一個功能對吧。

但是，tmall 商傢對我們來說，肯定是要分大客戶和小客戶的對吧，比如像蘋果，小米這樣大商傢一年起碼能給我們創造很大的利潤，所以理應當然，他們的訂單必須得到優先處理，而曾經我們的後端系統是使用 redis 來存放的定時輪詢，大傢都知道 redis 隻能用 List 做一個簡簡單單的消息隊列，並不能實現一個優先級的場景，所以訂單量大瞭後采用 RabbitMQ 進行改造和優化，如果發現是大客戶的訂單給一個相對比較高的優先級， 否則就是默認優先級。

如何添加？

控制臺頁面添加

隊列中代碼添加優先級

Map<String, Object> params = new HashMap();
params.put("x-max-priority", 10);
channel.queueDeclare("hello", true, false, false, params);

消息中代碼添加優先級

AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(10).build();

註意事項：

要讓隊列實現優先級需要做的事情有如下事情：隊列需要設置為優先級隊列，消息需要設置消息的優先級，消費者需要等待消息已經發送到隊列中才去消費因為，這樣才有機會對消息進行排序

實戰

生產者：

package com.jm.rabbitmq.one;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;
/**
 * 生產者 ：發消息
 */
public class Producer {
    //隊列名稱
    public static final String QUEUE_NAME="hello";
    //發消息
    public static void main(String[] args) throws Exception {
		Channel channel = RabbitMqUtils.getChannel();
        Map<String, Object> arguments=new HashMap<>();
        arguments.put("x-max-priority",10);//官方允許0-255 此處設置10 允許優先級范圍為0-10 不要設置過大，浪費cpu和內存
        channel.queueDeclare(QUEUE_NAME,true,false,false,arguments);
        for (int i = 1; i < 11; i++) {
            String message="info"+i;
            if(i==5){
                AMQP.BasicProperties properties=
                        new AMQP.BasicProperties().builder().priority(5).build();
                channel.basicPublish("",QUEUE_NAME,properties,message.getBytes());
            }else{
                channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            }
        }
        System.out.println("消息發送完畢");
    }
}

消費者：

package com.jm.rabbitmq.one;
import com.rabbitmq.client.*;
/**
 * 消費者 ：接收消息的
 */
public class Consumer {
    //隊列的名稱
    public static final String QUEUE_NAME="hello";
    //接收消息
    public static void main(String[] args) throws Exception{
		Channel channel = RabbitMqUtils.getChannel();
        //聲明 接收消息
        DeliverCallback deliverCallback= (consumerTag,message) ->{
            System.out.println(new String(message.getBody()));
        };
        //取消消息時的回調
        CancelCallback cancelCallback= consumerTag ->{
            System.out.println("消息消費被中斷");
        };
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

3. 惰性隊列

使用場景

RabbitMQ 從 3.6.0 版本開始引入瞭惰性隊列的概念。惰性隊列會盡可能的將消息存入磁盤中，而在消費者消費到相應的消息時才會被加載到內存中，它的一個重要的設計目標是能夠支持更長的隊列，即支持更多的消息存儲。當消費者由於各種各樣的原因(比如消費者下線、宕機亦或者是由於維護而關閉等)而致使長時間內不能消費消息造成堆積時，惰性隊列就很有必要瞭。

默認情況下，當生產者將消息發送到 RabbitMQ 的時候，隊列中的消息會盡可能的存儲在內存之中， 這樣可以更加快速的將消息發送給消費者。即使是持久化的消息，在被寫入磁盤的同時也會在內存中駐留一份備份。當RabbitMQ 需要釋放內存的時候，會將內存中的消息換頁至磁盤中，這個操作會耗費較長的時間，也會阻塞隊列的操作，進而無法接收新的消息。雖然 RabbitMQ 的開發者們一直在升級相關的算法， 但是效果始終不太理想，尤其是在消息量特別大的時候。

兩種模式

隊列具備兩種模式：default 和 lazy。默認的為default 模式，在3.6.0 之前的版本無需做任何變更。lazy 模式即為惰性隊列的模式，可以通過調用 channel.queueDeclare 方法的時候在參數中設置，也可以通過 Policy 的方式設置，如果一個隊列同時使用這兩種方式設置的話，那麼 Policy 的方式具備更高的優先級。 如果要通過聲明的方式改變已有隊列的模式的話，那麼隻能先刪除隊列，然後再重新聲明一個新的。

在隊列聲明的時候可以通過“x-queue-mode”參數來設置隊列的模式，取值為“default”和“lazy”。下面示例中演示瞭一個惰性隊列的聲明細節：

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-queue-mode", "lazy");
channel.queueDeclare("myqueue", false, false, false, args);

內存開銷對比

在發送 1 百萬條消息，每條消息大概占 1KB 的情況下，普通隊列占用內存是 1.2GB，而惰性隊列僅僅 占用 1.5MB

到此這篇關於RabbitMQ冪等性與優先級及惰性詳細全面講解的文章就介紹到這瞭,更多相關RabbitMQ冪等性內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：