RabbitMq——发布确认模式

87 阅读 0 评论 58 点赞

我是靠谱客的博主无限百合，最近开发中收集的这篇文章主要介绍RabbitMq——发布确认模式，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

rabbitmq有三种发布确认模式，分别为：

1.单个确认模式：每发送一条消息，确认一次，发布同数量消息，其耗时最长；

2.批量确认模式：每发送一部分消息，批量同步确认一次，若有消息无法发出，该模式无法确认是哪个消息无法发送；

3.异步批量确认模式：批量异步确认，该模式性能最好，在有错误情况下很好处理。

确认三种模式速度案例：

通过模拟发布1000条消息，通过其确认总时间确认速度

1.创建获取信道工具类，创建连接工程，设置ip地址，端口号，用户名，密码，连接获取信道

//连接工厂，创建信道工具类
public class RabbitUtils {
    // 得到一个连接的 channel
    public static Channel getChannel() throws Exception {
        // 创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.23.129");
        factory.setUsername("user");
        factory.setPassword("123");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        return channel;
    }
}

2.测试单个确认模式：

（1）获取连接，得到信道；

（2）使用uuid随机产生队列名称，并调用queueDeclare（队列名称，是否持久化，是否供一个消费者使用，是否自动删除，其他参数）方法声明信道；

（3）调用confirmSelect()方法开启消息发布确认模式；

（4）记录发消息前时间段

（5）开始发布1000条消息，并发一次，调用waitForConfirms（）进行确认应答

（6）记录发消息结束后时间段

（7）计算发消息前后时间差，得出运行时间

public static void publishSingle() throws Exception {
        //获得连接
        Channel channel = RabbitUtils.getChannel();
        //随机产生信道名称
        String queueName= UUID.randomUUID().toString();
        //声明信道
        channel.queueDeclare(queueName,false,false,false,null);
        //开启消息确认发布应答模式
        channel.confirmSelect();
        //记录开始时间
        long begin=System.currentTimeMillis();
        //发送1000条消息
        for (int i = 0; i < 1000; i++) {
            //模拟消息
            String message=i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //单个确认
            boolean b = channel.waitForConfirms();
            //b为true表示发送成功
            if (b){
                System.out.println("消息发布成功");
            }
        }
        //记录结束时间
        long end=System.currentTimeMillis();

        System.out.println("此模式为单个发布确认模式，一共耗时为:"+(end-begin)+"ms");

    }

3.测试同步批量确认模式：

（1）获取连接，得到信道；

（2）使用uuid随机产生队列名称，并调用queueDeclare（队列名称，是否持久化，是否供一个消费者使用，是否自动删除，其他参数）方法声明信道；

（3）调用confirmSelect()方法开启消息发布确认模式；

（4）记录发消息前时间段

（5）开始发布1000条消息，每发送100条消息，调用waitForConfirms（）进行确认应答

（6）记录发消息结束后时间段

（7）计算发消息前后时间差，得出运行时间

  //同步批量确认模式
    public static  void publishMore() throws Exception {
        Channel channel = RabbitUtils.getChannel();
        //随机生成队列名称
        String queueName=UUID.randomUUID().toString();
        //声明队列
        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认模式
        channel.confirmSelect();
        //记录发消息前时间
        long begin=System.currentTimeMillis();
        //开启发送消息
        for (int i = 0; i < 1000; i++) {
            String message=i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //每100条消息确认一次
            if(i%100==0){
                channel.waitForConfirms();
            }
        }
        //记录发消息后时间
        long end=System.currentTimeMillis();
        System.out.println("此模式为同步批量确认模式，耗时为："+(end-begin)+"ms");
    }

4.测试异步批量确认模式：

（1）获取连接，得到信道；

（2）使用uuid随机产生队列名称，并调用queueDeclare（队列名称，是否持久化，是否供一个消费者使用，是否自动删除，其他参数）方法声明信道；

（3）调用confirmSelect()方法开启消息发布确认模式；

（4）调用addConfirmListener（消息发送成功回调函数，消息发送失败回调函数）注册监听器监听消息，此监听器为异步通知

（4）记录发消息前时间段

（5）开始发布1000条消息

（6）记录发消息结束后时间段

（7）计算发消息前后时间差，得出运行时间

//异步批量确认消息
    public static void publishAsny() throws Exception {
        Channel channel = RabbitUtils.getChannel();
        String queueName=UUID.randomUUID().toString();
        //声明队列
        channel.queueDeclare(queueName,false,false,false,null);
        channel.confirmSelect();

        //消息发送成功回调函数
        ConfirmCallback ackCallback=(deliveryTag,multiple)->{
            System.out.println("消息法送成功");
        };

        //消息法送失败回调函数
        ConfirmCallback nackCallback=(deliveryTag,multiple)->{
            System.out.println("消息发送失败");
        };

        //注册监听器监听,异步通知
        channel.addConfirmListener(ackCallback,nackCallback);

        //记录发消息前时间
        long begin=System.currentTimeMillis();
        //发送消息
        for (int i = 0; i < 1000; i++) {
            String message=i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
        }
        //记录发消息后时间 
        long end=System.currentTimeMillis();
        System.out.println("该模式为异步批量确认模式，耗时："+(end-begin)+"ms");;

    }

5、测试结果如下：

可以发现异步确认处理最快，且能监听出发送成功的消息以及发送失败的消息，以便处理发送失败消息，防止消息丢失。