RabbitMQ发布确认COOY全球资源网-软件资源-干货分享-知识求知COOY全球资源网

消息持久化的两个前提：
1、设置要求队列必须在持久化，保证RabbitMQ宕机后，信道不会消失
2、设置要求队列中的消息必须持久话，保证RabbitMQ宕机后，消息不会消失

设置两个值后，并不能完全保证消息不丢失，因为将消息保存到磁盘的过程中，服务器可能宕机，无法将消息完整地保留到磁盘上。所以，就需要发布确认。

发布确认：发布者将消息发送给消息队列，消息队列将消息保存到磁盘上，然后告诉给发布者，确认已经保存到磁盘中。这个过程就叫做发布确认。

发布确认的策略

开启发布确认的方法

// 发布确认 channel.confirmSelect();

单个确认发布

这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它
被确认发布，后续的消息才能继续发布,waitForConfirmsOrDie(long)这个方法只有在消息被确认
的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。
这种确认方式有一个最大的缺点就是:发布速度特别的慢，因为如果没有确认发布的消息就会
阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某
些应用程序来说这可能已经足够了。

ConfirmMessage.class

// 批量发消息的个数
    public static final int MESSAGE_COUNT=1000;

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
       // 单个确认
        publishMessageIndividually();
    }

    public static void publishMessageIndividually() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMqUitls.rabbitMqConnection();
        // 队列声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false, false, null);
        // 开启发布确认
        channel.confirmSelect();
        // 开始时间
        long startTime = System.currentTimeMillis();
        for (int i = 0; i

耗时：

单个发布确认.png

RabbitMQ控制台：

单个发布确认RabbitMQ控制台.png

批量确认发布

与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是:当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布。

在代码中，批量发布相对于单个发布，发布确认channel.waitForConfirms()的位置有所不同，需要在生产者发送批量确认数量时，进行发布确认。当然，这是最简单的示例。代码如下：

public static void publishMessageBatch() throws Exception {
        Channel channel = RabbitMqUitls.rabbitMqConnection();
        // 队列声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false, false, null);
        // 开启发布确认
        channel.confirmSelect();
        // 开始时间
        long startTime = System.currentTimeMillis();

        // 定义批量确认数量100, 发送100条确认一次
        int batchCount = 100;
        for (int i = 0; i

批量确认发布所耗时间：

批量发布确认.png

异步确认发布

是性价比最高，无论是可靠性还是效率都没得说，他是利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否投递成功，

 public static void publishMessageAsync() throws Exception {
        Channel channel = RabbitMqUitls.rabbitMqConnection();
        // 队列声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false, false, null);
        // 开启发布确认
        channel.confirmSelect();
        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目 只要给到序列号
         * 3.支持并发访问
         */
        ConcurrentSkipListMap concurrentSkipListMap = new  ConcurrentSkipListMap();

        /**
         * 确认收到消息的一个回调
         * 1.消息序列号
         * 2.true 可以确认小于等于当前序列号的消息
         * false 确认当前序列号消息
         */
        // 消息成功
        ConfirmCallback ackCallBack = (deliveryTag, multiple) -> {
            System.out.println("确认的消息内容"+concurrentSkipListMap.get(deliveryTag) + "========确认的消息序号："+ deliveryTag);
            if (multiple) {
                ConcurrentNavigableMap concurrentNavigableMap =
                        concurrentSkipListMap.headMap(deliveryTag);
                concurrentNavigableMap.clear();
            } else {
                concurrentSkipListMap.clear();
            }
        };
        // 消息失败
        ConfirmCallback nackCallBack = (deliveryTag, multiple) -> {
            String s = concurrentSkipListMap.get(deliveryTag);
            System.out.println("未确认的消息内容"+s + "========未确认的消息序号："+ deliveryTag);
        };

        channel.addConfirmListener(ackCallBack, nackCallBack);
        // 开始时间
        long startTime = System.currentTimeMillis();

        // 发送消息
        for (int i = 0; i

异步处理耗时：