1. 冪等性

概念

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

消息重復消費

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

解決思路

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

消費端的冪等性保障

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

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

唯一ID+指紋碼機制

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

note Redis 原子性

利用 redis 執(zhí)行 setnx 命令，天然具有冪等性。從而實現(xiàn)不重復消費

2. 優(yōu)先級隊列

使用場景

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

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

如何添加？

控制臺頁面添加

隊列中代碼添加優(yōu)先級

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

消息中代碼添加優(yōu)先級

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

注意事項：

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

實戰(zhàn)

生產者：

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;
/**
 * 生產者 ：發(fā)消息
 */
public class Producer {
    //隊列名稱
    public static final String QUEUE_NAME="hello";
    //發(fā)消息
    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 允許優(yōu)先級范圍為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("消息發(fā)送完畢");
    }
}

消費者：

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

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

兩種模式

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

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

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

內存開銷對比

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

到此這篇關于RabbitMQ冪等性與優(yōu)先級及惰性詳細全面講解的文章就介紹到這了,更多相關RabbitMQ冪等性內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

最新評論