• 【RabbitMQ】——发布确认

    <1> 概念

    消息持久化不能完全保证消息不会丢失，再持久化到磁盘 之前可能消息就丢失了。发布确认可以完全保证消息不会丢失。
    生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都将会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号，此外 broker 也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理。
    confirm模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack 消息。
    在这里插入图片描述

    <2> 前提

    队列必须持久
    化消息必须持久化

    <3> 实现

    1. 开启发布确认方法

    默认没有开启需要生产者调用confirmSelect()方法，每当你要想要使用发布确认，都需要再channle上调用该方法。
    在这里插入图片描述

    2. 单个发布确认

    这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布,后续的消息才能继续发布, waitForConfirms这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。

    这种确认方式有一个最大的缺点就是:发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某些应用程序来说这可能已经足够了。

    代码实现如下：

    /**
     * 单个确认模式
     */
    public static void publishMessageIndividually() throws IOException, TimeoutException, InterruptedException {
        //获取信道
        Channel channel = RabbitMqUtils.getChannel();
        /**
         * 开启发布确认
         */
        channel.confirmSelect();
        channel.queueDeclare(QUEUE_NAME_INDIVIDUALLY, true, false, false, null);
        //开始时间
        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = String.valueOf(i);
            channel.basicPublish("", QUEUE_NAME_INDIVIDUALLY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("message send success : " + message);
            }
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布个"+MESSAGE_COUNT+"单独确认消息，耗时"+(end - begin)+"ms");
    }


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27

    运行结果如下：
    在这里插入图片描述

    3. 批量发布确认

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

    代码实现如下：

        /**
     * 批量确认模式
     */
    public static void publishMessageBatch() throws IOException, TimeoutException, InterruptedException {
        //获取信道
        Channel channel = RabbitMqUtils.getChannel();
        /**
         * 开启发布确认
         */
        channel.confirmSelect();
        channel.queueDeclare(QUEUE_NAME_BATCH, true, false, false, null);
        //开始时间
        long begin = System.currentTimeMillis();

        //批量确认的消息数量
        int batchSize = 100;
        for (int i = 0; i < MESSAGE_COUNT; i++) {

            String message = String.valueOf(i);
            channel.basicPublish("", QUEUE_NAME_BATCH, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            if (i % batchSize == 99) {
                boolean flag = channel.waitForConfirms();
                if (flag) {
                    System.out.println("message send success : " + message);
                }
            }

        }
        //结束时间
        long end = System.currentTimeMillis();

        System.out.println("发布个" + MESSAGE_COUNT + "批量确认消息，耗时" + (end - begin) + "ms");
    }


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34

    运行结果如下：
    在这里插入图片描述

    4. 异步发布确认

    异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说，他是利用回调函数来达到消息可靠性传递的,这个中间件也是通过函数回调来保证是否投递成功。
    在这里插入图片描述
    代码实现如下：

    /**
     * 异步批量确认模式
     */
    public static void publishMessageAsync() throws IOException, TimeoutException, InterruptedException {
        //获取信道
        Channel channel = RabbitMqUtils.getChannel();
        /**
         * 开启发布确认
         */
        channel.confirmSelect();
        channel.queueDeclare(QUEUE_NAME_ASYNC, true, false, false, null);

        /**
         * 消息确认成功 回调函数
         */
        ConfirmCallback ackCallback = (deliveryTag, mulitple) -> {
            System.out.println("ackCallback-deliveryTag : "+deliveryTag+" , mulitple : "+mulitple);

        };

        /**
         * 消息失败确认成功 回调函数
         * param1 消息的标记
         * param2 是否为批量标记
         */
        ConfirmCallback nackCallback = (deliveryTag, mulitple) -> {
            System.out.println("nackCallback-deliveryTag : "+deliveryTag+" , mulitple : "+mulitple);
        };

        /**
         * 消息的监听器，监听哪些消息成功了，哪些消息失败了
         * 监听是异步的
         * param1 监听成功消息
         * param2 监听失败成功消息
         */
        channel.addConfirmListener(ackCallback, nackCallback);

        //开始时间
        long begin = System.currentTimeMillis();

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = String.valueOf(i);
            channel.basicPublish("",QUEUE_NAME_ASYNC, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
        }
        //结束时间
        long end = System.currentTimeMillis();

        System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息，耗时" + (end - begin) + "ms");


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49

    运行结果如下：
    在这里插入图片描述

    5. 如何处理异步未确认消息

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

    代码实现如下：

    /**
     * 异步批量确认模式
     */
    public static void publishMessageAsync() throws IOException, TimeoutException, InterruptedException {
        //获取信道
        Channel channel = RabbitMqUtils.getChannel();
        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况下
         * 1. 能够轻松得将序号和消息进行关联
         * 2. 轻松的进行批量根据序号删除消息
         * 3. 支持高并发，支持多线程
         */
        ConcurrentSkipListMap<Long, String> outStandingConfirms = new ConcurrentSkipListMap<>();
        /**
         * 开启发布确认
         */
        channel.confirmSelect();
        channel.queueDeclare(QUEUE_NAME_ASYNC, true, false, false, null);

        /**
         * 消息确认成功 回调函数
         */
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            System.out.println("ackCallback-deliveryTag : " + deliveryTag + " , mulitple : " + multiple);
            // 2. 从哈希表删除掉已经确认的消息，剩下的就是未确认的消息
            if (multiple) {
                /**
                 * headMap : 返回map中键 小于或等于param1的部分
                 * param1
                 * param2 false 小于等于  , true 小于
                 */
                ConcurrentNavigableMap<Long, String> confirmed = outStandingConfirms.headMap(deliveryTag,true);
                System.out.println("ackCallback-confirmed : " + JSONUtil.toJsonStr(confirmed));
                confirmed.clear();
            }

        };

        /**
         * 消息失败确认成功 回调函数
         * param1 消息的标记
         * param2 是否为批量标记
         */
        ConfirmCallback nackCallback = (deliveryTag, mulitple) -> {
            // 3. 剩下的就是失败的消息
            ConcurrentNavigableMap<Long, String> messages = outStandingConfirms.headMap(deliveryTag, true);
            System.out.println("nackCallback-deliveryTag : " + deliveryTag + " , mulitple : " + mulitple + " , messages : " + JSONUtil.toJsonStr(messages));
        };

        /**
         * 消息的监听器，监听哪些消息成功了，哪些消息失败了
         * 监听是异步的
         * param1 监听成功消息
         * param2 监听失败成功消息
         */
        channel.addConfirmListener(ackCallback, nackCallback);

        //开始时间
        long begin = System.currentTimeMillis();

        for (int i = 1; i <= MESSAGE_COUNT; i++) {
            String message = String.valueOf(i);
            //1. 将所有发布的消息记录到哈希表
            outStandingConfirms.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", QUEUE_NAME_ASYNC, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息，耗时" + (end - begin) + "ms");
    }


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    6. 发布确认速度对比

    异步处理最快

  • 相关阅读:
    Debian 11 配置优化指南
    【读书笔记《凤凰架构》- 构架可靠的大型分布式系统.周志明】
    Slf4j + Logback日志框架
    flutter创建不同样式的按钮，背景色，边框，圆角，圆形，大小都可以设置
    web前端期末大作业 html+css家乡旅游主题网页设计 湖北武汉家乡介绍网页设计实例
    Android 实现ProgressBar菊花旋转进度条的 效果（用于耗时loading时显示）
    Redis - 启动
    java计算机毕业设计双峰县在线房屋租售网站源码+系统+数据库+lw文档+mybatis+运行部署
    搜索EE场景排序链路升级
    LeetCode | 一探环形链表的奥秘【快慢双指针妙解BAT等大厂经典算法题】
  • 原文地址：https://blog.csdn.net/qq_42000631/article/details/126306612