RabbitMQ 高级特性-生产端Confirm消息确认机制

2,143 阅读2分钟

生产端Confirm消息确认机制

什么是消息确认

消息确认,是指生产者投递消息后,如果Broker收到消息,则会给我们生产者一个应答。生产者进行接受应答,用来确认消息这条是否正常发送到Broker,这种方式也是消息的可靠性投递的核心保障!

Confirm消息确认

如何实现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
}

运行结果

producer运行结果

注意

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);
    }
}

运行结果

consumer运行结果