概念--Confirm确认消息

理解Confirm消息确认机制：

消息的确认，是指生产者投递消息后，如果Broket(RabbitMQ消息中心)收到消息，则会给我生产者一个应答。

生产者进行接收应答，用来确定这条消息是否正常的发送到Broket，这种方式也是消息的可靠性投递的核心保障！

image.png

producer 生产者，MQ Broker就是RabbitMQ消息中心

生产者发送一条消息，到 Broker 里面，然后会有一个 Broker confirm 到生产者，生产者会有一个 Config Listener 监听！这个整个过程是异步的，也就是我发送了消息就可以不用管了，只需要监听就好了！

如何实现Confirm 确认消息？

第一步：在 channel 上开启确认模式：channel.confirmSelect();

第二部：在 cahnnel 上添加监听：addConfirmListener ，监听成功和失败的返回结果，根据具体的结果对消息进行重新发送，或记录日志等后续处理！

什么情况下会 ack呢？

比如 磁盘满了，rabbitMQ出现了异常，key的容量达到上限了

如果我 Ack(确认) 和 Nack（没有确认） 都没有接受到消息呢？

需要用到 可靠性投递！

这个情况比如出现了，网络闪断，导致我没有收到，那么就需要用定时任务，去抓取状态，然后进行消息重发！(前提是基于 可靠性投递)

生产者

 package com.example.rabbitmqapi.confirm;

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

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

/**
 * 生产者
 *
 *      是在生产者处理
 *
 *      消息的确认，是指生产者投递消息后，如果Broket(RabbitMQ消息中心)收到消息，
 *      则会给我生产者一个应答。
 *      生产者进行接收应答，用来确定这条消息是否正常的发送到Broket，
 *      这种方式也是消息的可靠性投递的核心保障！
 *
 * @author weiximei on 2019-04-01
 */
public class Producer {

    public static void main(String[] args) throws IOException, TimeoutException {

        // 1.创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // IP地址或者域名
        connectionFactory.setHost("192.168.1.118");
        // 端口号默认是 5672
        connectionFactory.setPort(5672);
        // 指定虚拟主机路径, "/" 是rabbitMQ默认的虚拟主机路径
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("weiximei");
        connectionFactory.setPassword("weiximei");

        // 2.获取Connection
        Connection connection = connectionFactory.newConnection();

        // 3. 创建 channel，也就是连接信道
        Channel channel = connection.createChannel();

        /**
         * 4.指定我们的消息确认模式：消息的确认模式
         */
        channel.confirmSelect();

        // 5. 声明
        // 交换机名称
        String exchangeName = "test_confirm_exchange";
        String routingKey = "confirm_save";

        String msg = "Hello RabbitMQ,Save confirm";

        // 6.发送消息

        /**
         * channel.basicPublish 方法有个其他的重载方法
         * 其中有个参数
         *
         * mandatory的作用：
         *         当mandatory标志位设置为true时，如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，
         *         那么broker会调用basic.return方法将消息返还给生产者;当mandatory设置为false时，
         *         出现上述情况broker会直接将消息丢弃;通俗的讲，mandatory标志告诉broker代理服务器至少将消息route到一个队列中，
         *         否则就将消息return给发送者;
         *
         */

        channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());


        // 7.添加一个确认监听
        //异步监听确认和未确认的消息
        channel.addConfirmListener(new ConfirmListener() {

            /**
             * 异步执行的，消息确认有可能是批量确认的，是否批量确认在于返回的multiple的参数，
             * 此参数为bool值，如果true表示批量执行了deliveryTag这个值以前的所有消息，
             * 如果为false的话表示单条确认。
             */

            /**
             *
             * @param deliveryTag 消息的唯一标签，用来确认消息，标记消息
             * @param multiple 如果true表示批量执行了deliveryTag这个值以前的所有消息，
             *                  如果为false的话表示单条确认。
             * @throws IOException
             */
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println(String.format("已确认消息，标识：%d，多个消息：%b", deliveryTag, multiple));
            }

            /**
             *
             * @param deliveryTag 消息的唯一标签，用来确认消息，标记消息
             * @param multiple 如果true表示批量执行了deliveryTag这个值以前的所有消息，
             *                 如果为false的话表示单条确认。
             * @throws IOException
             */
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println(" ------  no ack! ------  没有被ack确认，标识: " + deliveryTag);
            }
        });





    }





}

package com.example.rabbitmqapi.confirm;

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

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

/**
 * 消费者
 *
 *      confirm 是在生产端
 *
 * @author weiximei on 2019-04-01
 */
public class Consumer {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

        // 1.创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // IP地址或者域名
        connectionFactory.setHost("192.168.1.118");
        // 端口号默认是 5672
        connectionFactory.setPort(5672);
        // 指定虚拟主机路径, "/" 是rabbitMQ默认的虚拟主机路径
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("weiximei");
        connectionFactory.setPassword("weiximei");

        // 2.获取Connection
        Connection connection = connectionFactory.newConnection();

        // 3. 创建 channel，也就是连接信道
        Channel channel = connection.createChannel();

        /**
         * 4.指定我们的消息确认模式：消息的确认模式
         */
        channel.confirmSelect();

        // 5. 声明
        // 交换机名称
        String exchangeName = "test_confirm_exchange";
        String routingKey = "confirm_save";
        String queueName = "test_confirm_queue";

        // 6.声明 exchange
        // 交换机名称，交换机类型，是否持久化
        channel.exchangeDeclare(exchangeName, "topic",true);

        // 7. 声明 队列
        /**
         * declare 表示是否持久化消息类型
         * exclusive 是否为当前连接的专用队列，在连接断开后（close），会自动删除该队列，生产环境中应该很少用到吧。
         *           表示是否独占，true也就是说我这个消息只能是这一个channel进行消费
         *           场景：比如这个10条消息，让一个人消费，要顺序执行，如果多个人消息不能保证消息是顺序执行的
         * autoDelete 表示是否自动删除，true当没有任何消费者使用时（也就是这个队列没有一个消费者后, 也可以说是这个队列的最后一个消费者close后,断开了连接），自动删除该队列
         * arguments 其他的一些参数
         */
        channel.queueDeclare(queueName, true, false, false,null);


        // 8.绑定队列 （也就是把队列绑定到某交换机）
        channel.queueBind(queueName, exchangeName, routingKey);

        // 9.创建消费者
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
        // 创建消费者
        // 消费的队列，是否自动ack，消费者是谁
        channel.basicConsume(queueName, true, queueingConsumer);

        while (true) {
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.out.println("接收到的消息：" + msg);
        }

    }

}