RabbitMQ:发布确认

发布确认原理

比如说 开启队列持久化，消息持久化之后，生产者给mq一个消息，还没来得及进行持久化，也就是还没有保存到磁盘，mq就挂了，消息丢失！

此时，就需要mq在给生产者回应一下，表示已经持久化完成，也就是发布确认！

描述：

生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的消息都将会被指派一个唯一的 ID(从 1 开始)，一旦消息被投递到所有匹配的队列之后，broker 就会发送一个确认给生产者(包含消息的唯一 ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 broker 也可以设置basic.ack 的multiple 域，表示到这个序列号之前的所有消息都已经得到了处理。

confirm 模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack 消息，生产者应用程序同样可以在回调方法中处理该 nack 消息。

发布确认策略

开启发布确认

默认没有开启，需要在channel调用confirmSelect开启！

单个确认发布

种同步确认发布的方式，发布一个消息之后只有它被确认发布，后续的消息才能继续发布

waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回（布尔），如果在指定时间范围内这个消息没有被确认那么它将抛出异常。

个最大的缺点就是:发布速度特别的慢，

public class ConfirmMessage {
    public static int MESSAGE_COUNT=100;

    public static void main(String[] args) throws Exception {
        ConfirmMessage.publishMessageOne();
    }
    //单个确认
    public static void publishMessageOne() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();
        //批量发消息
        for(int i=0;i<MESSAGE_COUNT;i++){
            String message = i + "";
            channel.basicPublish("",queueName,null,message.getBytes());
            //单个消息就马上进行发布确认
            boolean flag = channel.waitForConfirms();
            if(flag){
                System.out.println("消息发送成功");
            }
       }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"个单独确认消息，耗时"+(end-begin)+"ms");
    }
}

批量确认消息

先发布一批消息然后一起确认可以极大地提高吞吐量

当发生故障导致发布出现问题时，不知道是哪个消息出现问题了

public class ConfirmMessage {
    public static int MESSAGE_COUNT=100;

    public static void main(String[] args) throws Exception {
        ConfirmMessage.publishMessageMore();
    }
    //批量确认
    public static void publishMessageMore() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();
        //批量确认消息大小
        int batchSize = 10 ;
        //批量发消息
        for(int i=0;i<MESSAGE_COUNT;i++){
            String message = i + "";
            channel.basicPublish("",queueName,null,message.getBytes());
            if(i%batchSize == 0){
            //单个消息就马上进行发布确认
             channel.waitForConfirms();
                System.out.println("消息发送成功");
           }
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"个，每10个批量确认一次消息，耗时"+(end-begin)+"ms");
    }
}

异步确认发布

逻辑上更复杂，可靠性和效率更高！

public class ConfirmMessage {
    public static int MESSAGE_COUNT=100;

    public static void main(String[] args) throws Exception {
        ConfirmMessage.publishMessageYibu();
    }

    //异步确认
    public static  void publishMessageYibu() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();

        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.轻松的将序号与消费者进行关联
         * 2.轻松的批量删除条目 只要给到序号
         * 3.支持高并发
         */
        ConcurrentSkipListMap<Long,String> outstandingCinfirms = new ConcurrentSkipListMap<>();

        //消息确认成功 回调函数
        //deliveryTag：消息的标记    multiple：是否批量确认
        ConfirmCallback ackCallback = (deliveryTag,multiple) -> {
            if(multiple){
                // ***2*** 删除已经确认的消息 剩下的就是未确认的消息
                ConcurrentNavigableMap<Long, String> confirmed =  outstandingCinfirms.headMap(deliveryTag);
                confirmed.clear();
            }else {
                outstandingCinfirms.remove(deliveryTag);
            }

            System.out.println("确认消息"+deliveryTag);
        };
        //消息确认失败 回调函数
        /**
         * deliveryTag 消息的标记
         * multiple 是否为批量确认
         */
        ConfirmCallback nackCallback = (deliveryTag,multiple) -> {
            //***3*** 打印未确认的消息
           String message =  outstandingCinfirms.get(deliveryTag);
            System.out.println("未确认的消息："+message);
        };
        //准备监听器，监听哪些消息成功，哪些消息失败
        channel.addConfirmListener(ackCallback,nackCallback);

        //批量发送消息
        for(int i=0;i<MESSAGE_COUNT;i++){
            String message = "m"+i;
            channel.basicPublish("",queueName,null,message.getBytes());
            System.out.println("消息"+message+"发送成功！");
            // ***1*** 此处记录下所有要发送的消息  消息的总和
            outstandingCinfirms.put(channel.getNextPublishSeqNo(),message);
        }
}

如何处理异步未确认消息

最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，比如说用 ConcurrentLinkedQueue 这个队列在 confirm callbacks 与发布线程之间进行消息的传递。

对比

单独发布消息
同步等待确认，简单，但吞吐量非常有限。
批量发布消息
批量同步等待确认，简单，合理的吞吐量，一旦出现问题但很难推断出是那条消息出现了问题。
异步处理：
最佳性能和资源使用，在出现错误的情况下可以很好地控制，但是实现起来稍微难些