RabbitMQ 高級特性-生產端Confirm消息確認機制

生產端Confirm消息確認機制

什麼是消息確認

消息確認，是指生產者投遞消息後，若是Broker收到消息，則會給咱們生產者一個應答。生產者進行接受應答，用來確認消息這條是否正常發送到Broker，這種方式也是消息的可靠性投遞的核心保障！

如何實現Confirm確認消息

1. 在channel上開啓確認模式：channel.confirmSelect();
2. 在channel上添加監聽: addConfirmListener,監聽成功和失敗返回的結果，根據結果對消息重發送或記錄日誌等後續處理。

代碼實現

producer

package com.wyg.rabbitmq.javaclient.confirm;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/** * confirm消息確認機制 * * @author wyg0405@gmail.com * @date 2019-11-22 13:25 * @since JDK1.8 * @version V1.0 */

public class Producer {
    private static final String HOST = "localhost";
    private static final int PORT = 5672;
    private static final String USERNAME = "guest";
    private static final String PASSWORD = "guest";
    private static final String EXCHANGE = "test_confirm_exchange";

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST);
        factory.setVirtualHost("/");
        factory.setPort(PORT);
        factory.setUsername(USERNAME);
        factory.setPassword(PASSWORD);
        Connection connection = factory.newConnection();

        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE, "topic");

        // 開啓confirm確認模式
        channel.confirmSelect();

        // 添加監聽
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("---消息投遞成功，deliveryTag:" + deliveryTag);
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("---消息投遞失敗，deliveryTag:" + deliveryTag);
            }
        });

        String msg1 = "這是一條confirm確認消息save";
        channel.basicPublish(EXCHANGE, "confirm.save", null, msg1.getBytes("UTF-8"));
        String msg2 = "這是一條confirm確認消息abc";
        channel.basicPublish(EXCHANGE, "confirm.abc", null, msg2.getBytes("UTF-8"));
    }

    // 注意，由於要等待broker的confirm消息，暫時不關閉channel和connection
}

複製代碼

運行結果

注意

broker在接受消息成功後，ConfirmListener進入handleAck；失敗進入handleNack，失敗緣由有不少，好比MQ異常，queue容量達到上限，磁盤寫滿了等。

還有極端狀況，ack和nack都沒獲取到，好比網絡閃斷，這時候須要用定時任務去抓取作後續處理。

Consumer

package com.wyg.rabbitmq.javaclient.confirm;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.*;

/** * 消息確認 * * @author wyg0405@gmail.com * @date 2019-11-22 14:07 * @since JDK1.8 * @version V1.0 */

public class Consumer {

    private static final String HOST = "localhost";
    private static final int PORT = 5672;
    private static final String USERNAME = "guest";
    private static final String PASSWORD = "guest";

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST);
        factory.setVirtualHost("/");
        factory.setPort(PORT);
        factory.setUsername(USERNAME);
        factory.setPassword(PASSWORD);
        Connection connection = factory.newConnection();

        Channel channel = connection.createChannel();

        String queueName = "test_confirm_queue";
        String routingKey = "confirm.#";
        String exchangeName = "test_confirm_exchange";
        // 申明隊列
        channel.queueDeclare(queueName, true, false, false, null);
        // 隊列綁定到exchange
        channel.queueBind(queueName, exchangeName, routingKey, null);

        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                try {
                    System.out.println("---消費者：" + new String(message.getBody(), "UTF-8"));
                } finally {
                    // consumer手動 ack 給broker
                    channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
                }
            }
        };
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
                System.out.println("---消費者：cancelCallback");
            }
        };

        // 消費消息
        channel.basicConsume(queueName, false, deliverCallback, cancelCallback);
    }
}

複製代碼

運行結果