0
点赞
收藏
分享

微信扫一扫

消息中间件 RabbitMQ 之 发布确认

文风起武 2022-04-18 阅读 78
java中间件

4. 发布确认

4.1 发布确认原理

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-f4jLASOM-1650282876336)(C:\Users\Administrator\AppData\Roaming\Typora\typora-user-images\image-20220418145008481.png)]

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

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ddYkK6Yh-1650282876338)(C:\Users\Administrator\AppData\Roaming\Typora\typora-user-images\image-20220418105530054.png)]

4.2 发布确认的策略

4.2.1 开启发布确认的方法

发布确认默认是没有开启的，如果要开启需要调用方法 confirmSelect( )。每当你想要使用发布确认，都需要在生产者的 channel 上调用该方法

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ogcmZL9K-1650282876339)(C:\Users\Administrator\AppData\Roaming\Typora\typora-user-images\image-20220418101841925.png)]

package com.example.three;

import com.example.utils.RabbitUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;

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

/**
 * @author 且听风吟
 * @version 1.0
 * @description: 生产者
 *   消息在手动应答时是不丢失的，被丢失时消息会被放回队列中，重新消费
 * @date 2022/4/16 0016 15:15
 */
public class Task2 {

    /**
     * 队列名称
     */
    public static final String TASK_NAME = "ack_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        //通过工具类获取信道
        Channel channel = RabbitUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();

        /**
         * 生成一个队列，参数的含义如下：
         *      1.队列名称
         *      2.队列里面的消息是否持久化（存储在磁盘上），默认情况false（存储在内存中）
         *      3.该队列是否进行消息共享，默认false
         *      4.是否自动删除，最后一个消费者断开连接后，该队列是否自动删除
         *      5.其他参数
         */
        //使消息队列持久化
        boolean durable = true;
        channel.queueDeclare(TASK_NAME,durable,false,false,null);

        //从控制台中输入信息
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()){
            String message = scanner.next();
            /**
             * 发送一个消息，参数含义如下：
             *    1.发送到哪个交换机，null表示使用默认交换机
             *    2.路由的key值是哪个 本次是队列的名称
             *    3.其他参数信息
             *    4.发送消息的消息体（发送消息的二进制码）
             */
            channel.basicPublish("",TASK_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));
            System.out.println("生产者发出消息："+message);
        }
    }

}

4.2.2 单个确认发布

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

/**
     * 单个确认
     */
    public static void publishMessageOne() throws IOException, TimeoutException, InterruptedException {
        //通过工具类获取信道
        Channel channel = RabbitUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        //生成队列
        channel.queueDeclare(queueName,true,false,false,null);

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

        //批量发消息
        for (int i=0;i<MESSAGE_COUNT;i++){
            //要发布的消息体
            String message = i+" ";
            //发布消息
            channel.basicPublish("",queueName,null,message.getBytes());
            //单个消息就马上发布消息确认
            boolean confirms = channel.waitForConfirms();
            if (confirms){
                System.out.println("消息发送成功");
            }
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"个单独确认消息，耗时"+(end-begin)+"ms");
    }

消耗时间：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-CNyQGvTq-1650282876340)(C:\Users\Administrator\AppData\Roaming\Typora\typora-user-images\image-20220418153338586.png)]

4.2.3 批量确认发布

  • 上面哪种方式非常慢，与单个等待确认消息相比，先发布消息然后再批量确认，这样可以提高吞吐量。

  • 这种方式也存在缺点：当发生故障导致发布出现问题的时候，不知道具体是哪个消息出现了问题。必须将整个批处理保存在内存中，以记录重要信息而后重新发布消息。

  • 这种方案依然是同步的，也一样会阻塞消息的发布。

/**
     * 批量发布确认
     */
    public static void publishMessageBatch() throws IOException, TimeoutException, InterruptedException {
        //通过工具类获取信道
        Channel channel = RabbitUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        //生成队列
        channel.queueDeclare(queueName,true,false,false,null);

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

        //批量确认消息大小
        int batchSize = 100;

        //批量发布消息，批量发布确认
        for (int i=0;i<MESSAGE_COUNT;i++){
            //要发布的消息体
            String message = i+" ";
            //发布消息
            channel.basicPublish("",queueName,null,message.getBytes());

            //批量消息发布消息确认
            if (i % batchSize == 0){
                //发布确认
                boolean confirms = channel.waitForConfirms();
                System.out.println("消息发送成功");
            }
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"个批量确认消息，耗时"+(end-begin)+"ms");
    }

消耗时间：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-blZIZWD1-1650282876341)(C:\Users\Administrator\AppData\Roaming\Typora\typora-user-images\image-20220418153505679.png)]

4.2.4 异步确认发布

下面来详细讲解异步请求确认是如何实现的

/**
     * 异步发布确认
     */
    public static void publishMessageAsync() throws IOException, InterruptedException, TimeoutException {
        //通过工具类获取信道
        Channel channel = RabbitUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        //生成队列
        channel.queueDeclare(queueName,true,false,false,null);

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

        //消息确认成功 回调函数
        ConfirmCallback ackCallback = (var1,var3)->{
            System.out.println("确认的消息："+var1);
        };
        //消息确认失败 回调函数
        /**
         * 参数含义：
         * 1.消息的标记
         * 2.是否批量确认
         */
        ConfirmCallback nackCallBack = (var1,var3)->{
            System.out.println("未确认的消息："+var1);
        };
        //准备消息的监听器 监听哪些消息成功了，哪些消息失败了
        channel.addConfirmListener(ackCallback,nackCallBack);

        //批量发送消息，批量发布确认
        for (int i=0;i<MESSAGE_COUNT;i++){
            //要发布的消息体
            String message = i+" ";
            //发布消息
            channel.basicPublish("",queueName,null,message.getBytes());
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"个异步确认消息，耗时"+(end-begin)+"ms");
    }

消耗时间：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-iTdxdQ7E-1650282876342)(C:\Users\Administrator\AppData\Roaming\Typora\typora-user-images\image-20220418153640508.png)]

4.2.5 全部程序展示

package com.example.four;

import com.example.utils.RabbitUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

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

/**
 * @author 且听风吟
 * @version 1.0
 * @description:
 *       发布确认模式：
 *       比较使用的时间  找到哪种确认方式是最好的
 *       1.单个确认
 *       2.批量确认
 *       3.异步批量确认
 *
 * @date 2022/4/18 0018 10:38
 */
public class ConfirmMessage {

    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        //1.单个确认
        //publishMessageOne();  //发布1000个单独确认消息，耗时21384ms
        //2.批量确认
        //publishMessageBatch();  //发布1000个批量确认消息，耗时271ms
        //3.异步批量确认
        publishMessageAsync();  //发布1000个异步确认消息，耗时45ms
    }

    /**
     * 单个确认
     */
    public static void publishMessageOne() throws IOException, TimeoutException, InterruptedException {
        //通过工具类获取信道
        Channel channel = RabbitUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        //生成队列
        channel.queueDeclare(queueName,true,false,false,null);

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

        //批量发消息
        for (int i=0;i<MESSAGE_COUNT;i++){
            //要发布的消息体
            String message = i+" ";
            //发布消息
            channel.basicPublish("",queueName,null,message.getBytes());
            //单个消息就马上发布消息确认
            boolean confirms = channel.waitForConfirms();
            if (confirms){
                System.out.println("消息发送成功");
            }
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"个单独确认消息，耗时"+(end-begin)+"ms");
    }

    /**
     * 批量发布确认
     */
    public static void publishMessageBatch() throws IOException, TimeoutException, InterruptedException {
        //通过工具类获取信道
        Channel channel = RabbitUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        //生成队列
        channel.queueDeclare(queueName,true,false,false,null);

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

        //批量确认消息大小
        int batchSize = 100;

        //批量发布消息，批量发布确认
        for (int i=0;i<MESSAGE_COUNT;i++){
            //要发布的消息体
            String message = i+" ";
            //发布消息
            channel.basicPublish("",queueName,null,message.getBytes());

            //批量消息发布消息确认
            if (i % batchSize == 0){
                //发布确认
                boolean confirms = channel.waitForConfirms();
                System.out.println("消息发送成功");
            }
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"个批量确认消息，耗时"+(end-begin)+"ms");
    }

    /**
     * 异步发布确认
     */
    public static void publishMessageAsync() throws IOException, InterruptedException, TimeoutException {
        //通过工具类获取信道
        Channel channel = RabbitUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        //生成队列
        channel.queueDeclare(queueName,true,false,false,null);

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

        //消息确认成功 回调函数
        ConfirmCallback ackCallback = (var1,var3)->{
            System.out.println("确认的消息："+var1);
        };
        //消息确认失败 回调函数
        /**
         * 参数含义：
         * 1.消息的标记
         * 2.是否批量确认
         */
        ConfirmCallback nackCallBack = (var1,var3)->{
            System.out.println("未确认的消息："+var1);
        };
        //准备消息的监听器 监听哪些消息成功了，哪些消息失败了
        channel.addConfirmListener(ackCallback,nackCallBack);

        //批量发送消息，批量发布确认
        for (int i=0;i<MESSAGE_COUNT;i++){
            //要发布的消息体
            String message = i+" ";
            //发布消息
            channel.basicPublish("",queueName,null,message.getBytes());
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"个异步确认消息，耗时"+(end-begin)+"ms");
    }

}

4.2.6 如何处理异步未确认消息

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

步骤：

  1. 在发送消息的时候记录下所有要发送的消息 即消息的总和
  2. 删除掉已经确认的消息，剩下的就是未确认的消息
  3. 打印出未确认的消息
/**
     * 异步发布确认
     */
    public static void publishMessageAsync() throws IOException, InterruptedException, TimeoutException {
        //通过工具类获取信道
        Channel channel = RabbitUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        //生成队列
        channel.queueDeclare(queueName,true,false,false,null);

        /**
         * 线程安全有序的一个哈希表 适用于高并发的情况下
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目
         * 3.支持高并发（多线程）
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms =
                new ConcurrentSkipListMap<>();

        //消息确认成功 回调函数
        ConfirmCallback ackCallback = (var1,var3)->{
            //2.删除掉已经确认的消息，剩下的就是未确认的消息
            if (var3){
                //var3代表消息是否批量确认
                ConcurrentNavigableMap<Long, String> confirmed =
                        outstandingConfirms.headMap(var1);
                confirmed.clear();
            }else {
                outstandingConfirms.remove(var1);
            }
            System.out.println("确认的消息："+var1);
        };
        //消息确认失败 回调函数
        /**
         * 参数含义：
         * 1.消息的标记
         * 2.是否批量确认
         */
        ConfirmCallback nackCallBack = (var1,var3)->{
            //3.打印一下未确认的消息
            String message = outstandingConfirms.get(var1);

            System.out.println("未确认的消息是："+message+"++++++++未确认的消息de标记tag："+var1);
        };
        //准备消息的监听器 监听哪些消息成功了，哪些消息失败了
        channel.addConfirmListener(ackCallback,nackCallBack);

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

        //批量发送消息，批量发布确认
        for (int i=0;i<MESSAGE_COUNT;i++){
            //要发布的消息体
            String message = i+" ";
            //发布消息
            channel.basicPublish("",queueName,null,message.getBytes());

            //1.此处记录下所有要发送的消息  消息的总和
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"个异步确认消息，耗时"+(end-begin)+"ms");
    }

4.2.7 以上三种发布确认速度对比

  • 单独发布消息：

    同步等待确认，简单，但吞吐量非常有限

  • 批量发布消息：

    批量同步等待确认，简单，合理的吞吐量，一旦出现问题很难判断是哪条消息出现了问题

  • 异步处理：

    最佳性能和资源利用，在出现错误的情况下很容易控制，但是实现起来稍微难些

举报

相关推荐

消息中间件 RabbitMQ 之 发布高级确认 详解

java中间件

【RabbitMQ消息中间件】6.消息的确认模式

RabbitMQnextDeliverybasicAckbasicConsumeJavaScript前端开发

消息中间件-RabbitMQ

rabbitmqjava分布式

RabbitMQ消息中间件

前端javascript开发语言

消息中间件：RabbitMQ

分类数据挖掘人工智能

RabbitMQ 消息中间件

RabbitMQJava系统设计Java web

消息中间件RabbitMQ

java后端

消息中间件RabbitMQ（三）——常见消息中间件比较

rabbitmq消息队列

RabbitMQ 消息中间件 消息队列

java-rabbitmqmongodbredismysqlvue.jselasticsearchpostman

SSM框架之RabbitMQ——消息中间件

rabbitmq分布式SpringSpringBoot工作模式
0 条评论