Java RabbitMQ中的持久化與發布確認實作方法

1. 持久化

當RabbitMQ服務停掉以後訊息生產者發送過的訊息不遺失。預設情況下RabbitMQ退出或崩潰時，會忽略掉佇列和訊息。為了保證訊息不遺失需要將佇列和訊息都標記為持久化。

1.1 實作持久化

1.佇列持久化：在建立佇列時將channel.queueDeclare();第二個參數改為true。

2.訊息持久化：使用頻道發送訊息時channel.basicPublish();將第三個參數改為：MessageProperties.PERSISTENT_TEXT_PLAIN表示持久化消息。

/**
 * @Description 持久化MQ
 * @date 2022/3/7 9:14
 */
public class Producer3 {
    private static final String LONG_QUEUE = "long_queue";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        // 持久化队列
        channel.queueDeclare(LONG_QUEUE,true,false,false,null);
        Scanner scanner = new Scanner(System.in);
        int i = 0;
        while (scanner.hasNext()){
            i++;
            String msg = scanner.next() + i;
            // 持久化消息
            channel.basicPublish("",LONG_QUEUE, MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes(StandardCharsets.UTF_8));
            System.out.println("发送消息：'" + msg + "'成功");
        }
    }
}

但是儲存訊息還有存在一個快取的間隔點，沒有真正的寫入磁碟，持久性保證不夠強，但是對於簡單佇列而言也綽綽有餘。

1.2 不公平分發

輪詢分發的方式在消費者處理效率不同的情況下並不適用。所以真正的公平應該是遵循能者多勞的前提。

在消費者處修改channel.basicQos(1);表示開啟不公平分發

/**
 * @Description 不公平分发消费者
 * @date 2022/3/7 9:27
 */
public class Consumer2 {
    private static final String LONG_QUEUE = "long_queue";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            // 模拟并发沉睡三十秒
            try {
                Thread.sleep(30000);
                System.out.println("线程B接收消息："+ new String(message.getBody(), StandardCharsets.UTF_8));
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };
        // 设置不公平分发
        channel.basicQos(1);
        channel.basicConsume(LONG_QUEUE,false,deliverCallback,
                consumerTag -> {
                    System.out.println(consumerTag + "消费者取消消费");
                });
    }
}

1.3 測試不公平分發

#測試目的：是否能實現能者多勞。

測試方法：兩個消費者睡眠不同的事件來模擬處理事件不同，如果處理時間(睡眠時間)短的能夠處理多個訊息就代表目的達成。

先啟動生產者建立佇列，再分別啟動兩個消費者。

生產者依照順序發送四個訊息：

#睡眠時間短的執行緒A接收到了三個訊息

而睡眠時間長的執行緒B只接收到的第二個訊息：

因為執行緒B在處理訊息時消耗的時間較長，所以就將其他訊息分配給了線程A。

實驗成功！

1.4 預取值

訊息的傳送和手動確認都是非同步完成的，因此就存在一個未確認訊息的緩衝區，開發人員希望能夠限制緩衝區的大小，用來避免緩衝區裡面無限制的未確認訊息問題。

這裡的預期值就值得是上述方法channel.basicQos();裡面的參數，如果在當前頻道上存在等於參數的訊息就不會在安排當前頻道進行消費訊息。

1.4.1 程式碼測試

測試方法：

#1.新建兩個不同的消費者分別給定預期值5個2。

2.給予睡眠時間長的指定為5，時間短的指定為2。

3.假如依照指定的預期值取得訊息則表示測試成功，但並不是代表一定會依照5和2分配，這個類似權重的判別。

程式碼根據上述程式碼修改預期值即可。

2. 發布確認

發布確認就是生產者發布訊息到佇列之後，佇列確認進行持久化完畢再通知給生產者的過程。這樣才能保證訊息不會遺失。

要注意的是需要開啟佇列持久化才能使用確認發布。
開啟方法：channel.confirmSelect();

2.1 單一確認發布

是一種同步發布的方式，即發送完一個訊息之後只有確認它確認發布後，後續的消息才會繼續發布，在指定的時間內沒有確認就會拋出異常。缺點就是特別慢。

/**
 * @Description 确认发布——单个确认
 * @date 2022/3/7 14:49
 */
public class SoloProducer {
    private static final int MESSAGE_COUNT = 100;
    private static final String QUEUE_NAME = "confirm_solo";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        // 产生队列
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        // 开启确认发布
        channel.confirmSelect();
        // 记录开始时间
        long beginTime = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String msg = ""+i;
            channel.basicPublish("",QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes(StandardCharsets.UTF_8));
            // 单个发布确认
            boolean flag = channel.waitForConfirms();
            if (flag){
                System.out.println("发送消息：" + i);
            }
        }
        // 记录结束时间
        long endTime = System.currentTimeMillis();
        System.out.println("发送" + MESSAGE_COUNT + "条消息消耗："+(endTime - beginTime) + "毫秒");   }
}

2.2 批次確認發布

一批一批的確認發布可以提高系統的吞吐量。但缺點是故障導致發佈出現問題時，需要將整個批次保存在記憶體中，後面再重新發布。

/**
 * @Description 确认发布——批量确认
 * @date 2022/3/7 14:49
 */
public class BatchProducer {
    private static final int MESSAGE_COUNT = 100;
    private static final String QUEUE_NAME = "confirm_batch";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        // 产生队列
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        // 开启确认发布
        channel.confirmSelect();
        // 设置一个多少一批确认一次。
        int batchSize = MESSAGE_COUNT / 10;
        // 记录开始时间
        long beginTime = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String msg = ""+i;
            channel.basicPublish("",QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes(StandardCharsets.UTF_8));
            // 批量发布确认
            if (i % batchSize == 0){
                if (channel.waitForConfirms()){
                    System.out.println("发送消息：" + i);
                }
            }
        }
        // 记录结束时间
        long endTime = System.currentTimeMillis();
        System.out.println("发送" + MESSAGE_COUNT + "条消息消耗："+(endTime - beginTime) + "毫秒");
    }
}

顯然效率比單一確認發布的高很多。

2.3 非同步確認發布

在程式設計上比上述兩個複雜，但是性價比很高，無論是可靠性還行效率的都好很多，利用回呼函數來達到訊息可靠性傳遞的。

/**
 * @Description 确认发布——异步确认
 * @date 2022/3/7 14:49
 */
public class AsyncProducer {
    private static final int MESSAGE_COUNT = 100;
    private static final String QUEUE_NAME = "confirm_async";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        // 产生队列
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        // 开启确认发布
        channel.confirmSelect();
        // 记录开始时间
        long beginTime = System.currentTimeMillis();
        // 确认成功回调
        ConfirmCallback ackCallback = (deliveryTab,multiple) ->{
            System.out.println("确认成功消息：" + deliveryTab);
        };
        // 确认失败回调
        ConfirmCallback nackCallback = (deliveryTab,multiple) ->{
            System.out.println("未确认的消息：" + deliveryTab);
        };
        // 消息监听器
        /**
         * addConfirmListener:
         *                  1. 确认成功的消息；
         *                  2. 确认失败的消息。
         */
        channel.addConfirmListener(ackCallback,nackCallback);
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String msg = "" + i;
            channel.basicPublish("",QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes(StandardCharsets.UTF_8));
        }

        // 记录结束时间
        long endTime = System.currentTimeMillis();
        System.out.println("发送" + MESSAGE_COUNT + "条消息消耗："+(endTime - beginTime) + "毫秒");
    }
}

2.4 處理未確認的訊息

最好的處理方式把未確認的訊息放到一個基於記憶體的能被發佈線程存取的佇列。

例如：ConcurrentLinkedQueue可以在確認佇列confirm callbacks與發佈執行緒之間進行訊息的傳遞。

處理方式：

1.記錄要傳送的全部訊息；

2.在發布成功確認處刪除；

#3.列印未確認的訊息。

使用一個雜湊表儲存訊息，它的優點：

可以将需要和消息进行关联；轻松批量删除条目；支持高并发。

ConcurrentSkipListMap<Long,String > map = new ConcurrentSkipListMap<>();

/**
 * @Description 异步发布确认，处理未发布成功的消息
 * @date 2022/3/7 18:09
 */
public class AsyncProducerRemember {
    private static final int MESSAGE_COUNT = 100;
    private static final String QUEUE_NAME = "confirm_async_remember";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        // 产生队列
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        // 开启确认发布
        channel.confirmSelect();
        // 线程安全有序的一个hash表，适用与高并发
        ConcurrentSkipListMap< Long, String > map = new ConcurrentSkipListMap<>();
        // 记录开始时间
        long beginTime = System.currentTimeMillis();
        // 确认成功回调
        ConfirmCallback ackCallback = (deliveryTab, multiple) ->{
            //2. 在发布成功确认处删除；
            // 批量删除
            if (multiple){
                ConcurrentNavigableMap<Long, String> confirmMap = map.headMap(deliveryTab);
                confirmMap.clear();
            }else {
                // 单独删除
                map.remove(deliveryTab);
            }
            System.out.println("确认成功消息：" + deliveryTab);
        };
        // 确认失败回调
        ConfirmCallback nackCallback = (deliveryTab,multiple) ->{
            // 3. 打印未确认的消息。
            System.out.println("未确认的消息：" + map.get(deliveryTab) + ",标记：" + deliveryTab);
        };
        // 消息监听器
        /**
         * addConfirmListener:
         *                  1. 确认成功的消息；
         *                  2. 确认失败的消息。
         */
        channel.addConfirmListener(ackCallback,nackCallback);
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String msg = "" + i;
            channel.basicPublish("",QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes(StandardCharsets.UTF_8));
            // 1. 记录要发送的全部消息；
            map.put(channel.getNextPublishSeqNo(),msg);
        }

        // 记录结束时间
        long endTime = System.currentTimeMillis();
        System.out.println("发送" + MESSAGE_COUNT + "条消息消耗："+(endTime - beginTime) + "毫秒");
    }
}

以上是Java RabbitMQ中的持久化與發布確認實作方法的詳細內容。更多資訊請關注PHP中文網其他相關文章！