RabbitMQ消息应答与发布

消息应答

RabbitMQ一旦向消费者发送了一个消息,便立即将该消息,标记为删除.

消费者完成一个任务可能需要一段时间,如果其中一个消费者处理一个很长的任务并仅仅执行了一半就突然挂掉了,在这种情况下,我们将丢失正在处理的消息,后续给消费者发送的消息也就无法接收到了.

为了确保消息不丢失,我们引入了消息应答机制.

消息应答就是:消费者在接收到生产者的消息并且处理该消息之后,告诉RabbitMQ已经处理完成了,rabbitMQ可以进行删除了.

一个生产者,两个消费者.

/**
消费者1
*/
public class Consumer1 {
 public static final String QueueName = "TaskQueueName";
 public static void main(String[] args) throws IOException, TimeoutException {
     ConnectionFactory factory = new ConnectionFactory();
     factory.setHost("127.0.0.1");
     factory.setUsername("DGZ");
     factory.setPassword("Dgz@#151");
     Connection connection = factory.newConnection();
     Channel channel = connection.createChannel();
     //声明接收消息
     System.out.println("消费者1等待接收消息时间较短");
     DeliverCallback deliverCallback = (consumerTag,message) -> {
         Util.sleep(1);  //等待1s
         System.out.println("正在处理该消息" + new String(message.getBody()));

            /**
             * 手动应答
             * 1.消息的标记Tag
             * 2.是否批量应答 false表示不批量应答信道中的消息
             */
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };
        //取消消息时的回调
        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消息消费被中断");
        };
        /**
         * 消费者消费消息
         * 1.消费哪个队列
         * 2.消费成功之后是否要自动应答true：代表自动应答      false:代表手动应答
         * 3.消费者未成功消费的回调
         * 4.消费者取消消费的回调
         */
        boolean autoAck = false;
        channel.basicConsume(QueueName,autoAck,deliverCallback,cancelCallback);
    }
}

/**
消费者2
*/
public class Consumer2 {
 public static final String QueueName = "TaskQueueName";
 public static void main(String[] args) throws IOException, TimeoutException {
     ConnectionFactory factory = new ConnectionFactory();
     factory.setHost("127.0.0.1");
     factory.setUsername("DGZ");
     factory.setPassword("Dgz@#151");
     Connection connection = factory.newConnection();
     Channel channel = connection.createChannel();
     System.out.println("消费者2等待接收消息时间较长");
     //声明接收消息
     DeliverCallback deliverCallback = (consumerTag,message) -> {
         Util.sleep(10); //等待10s
         System.out.println("正在处理该消息" + new String(message.getBody()));

            /**
             * 手动应答
             * 1.消息的标记Tag
             * 2.是否批量应答 false表示不批量应答信道中的消息
             */
            //当该消息还没有被处理的时候,如果此时这个应用挂掉,
            // 由于这个手动应答的机制,就不会删除该消息,而是将给消息交给其他应用去处理
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };
        //取消消息时的回调
        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消息消费被中断");
        };
        /**
         * 消费者消费消息
         * 1.消费哪个队列
         * 2.消费成功之后是否要自动应答true：代表自动应答     false:代表手动应答
         * 3.消费者未成功消费的回调
         * 4.消费者取消消费的回调
         */
        boolean autoAck = false;
        channel.basicConsume(QueueName,autoAck,deliverCallback,cancelCallback);
    }
}

消费者1 1s处理一个消息,消费者2 10s处理一个消息

/**
生产者
*/
public class Producer {
 //队列名称
 public static final String QueueName = "TaskQueueName";
 public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
     ConnectionFactory factory = new ConnectionFactory();   //创建连接工厂
     factory.setHost("127.0.0.1");  //主机
     factory.setUsername("DGZ"); //用户名
     factory.setPassword("Dgz@#151"); // 密码
     Connection connection = factory.newConnection();  //通过连接工厂创建一个连接
     Channel channel = connection.createChannel();  //获取信道
     /**
      * 生产一个对列
      * 1.对列名称
      * 2.对列里面的消息是否持久化，默认情况下，消息存储在内存中
      * 3.该队列是否只供一个消费者进行消费，是否进行消息共享，true可以多个消费者消费 false：只能一个消费者消费
      * 4.是否自动删除，最后一个消费者端开链接以后，该队列是否自动删除，true表示自动删除
      * 5.其他参数
      */
     channel.queueDeclare(QueueName,false,false,false,null);
        Scanner scanner = new Scanner(System.in);
        System.out.println("请输入消息:");
        while(scanner.hasNext()) {
            String message = scanner.next();
            channel.basicPublish("",QueueName,null,message.getBytes());
            System.out.println("生产者发出消息: " + message);
        }
        /**
         * 发送一个消息
         * 1.发送到哪个交换机
         * 2.路由的key值是哪个本次是队列的名称
         * 3.其他参数信息
         * 4.发送消息的消息体
         */
        System.out.println("消息发送完毕");
    }
}

自动应答和手动应答:

自动应答就是MQ只要把消息发出去,不会管消息是否收到,都会立刻把这个消息进行删除.

手动应答就是MQ把消息发送出去之后,消费者在接收到生产者的消息并且处理该消息之后,告诉RabbitMQ已经处理完成了,rabbitMQ可以进行删除了.

RabbitMQ持久化

当RabbitMQ服务突然挂掉之后,消息生产者发送过来的消息如何保证不丢失.

默认情况下当RabbitMQ服务挂掉之后,它会忽略队列和消息,这时刚刚发送的消息和队列都会丢失.

确保消息不丢失我们需要干两件事,就是将队列和消息标记为持久化.

队列持久化

之前创建的队列都是非持久化的,如果RabbitMQ重启的话,队列就是丢失,如果需要实现持久化队列,那么就需要在声明队列的时候把durable参数设置为true.表示代表开启持久化.

public class Producer {
    public static final String QueueName = "CJ_QUEUE";
    public static void main(String[] args) throws IOException, TimeoutException {
        /**
         * RabbitMQ工具类,用来创建连接等信息
         */
        Channel channel = RabbitMQUtil.RabbitMQ_getChannel();
        boolean durable = true;
        //第二个参数为队列持久化的参数,设置为true,表示队列开启持久化,false表示不开启持久化
        channel.queueDeclare(QueueName,durable,false,false,null);
    }
}

此时web管理端就会看见:

表示队列持久化

消息持久化

消息持久化是指消息生产者发布消息的时候,开启消息持久化,

public class Producer {
    public static final String QueueName = "CJ_QUEUE";

    public static void main(String[] args) throws IOException, TimeoutException {
        /**
         * RabbitMQ工具类,用来创建连接等信息
         */
        Channel channel = RabbitMQUtil.RabbitMQ_getChannel();
        boolean durable = true;
        //第二个参数为队列持久化的参数,设置为true,表示队列开启持久化,false表示不开启持久化
        channel.queueDeclare(QueueName,durable,false,false,null);
        Scanner scanner = new Scanner(System.in);
        System.out.println("请输入要发送的消息");
        while (scanner.hasNext()) {
            String message = scanner.next();
            /**
             * 将发布消息的basicPublish方法的第三个参数设置为MessageProperties.PERSISTENT_TEXT_PLAIN
             * 表示开启消息持久化
             */
            channel.basicPublish
                    ("",QueueName, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));
            System.out.println("生产者发出消息" + message);
        }
    }
}

不公平分发

在最开始的时候RabbitMQ采用的是轮询分发,但是在某种场景下这种策略并不是很好,比如说两个消费者在处理消息,此时消费者1处理消息比较慢,而消费者2处理消息比较快,如果这个时候还是采用轮询分发的方式,那么处理慢的消费者就会一直在处理消息,而处理快的消费者就会有很大时间处于空闲状态.

为了解决这个问题,引入了不公平分发

不公平分发: 如果一个工作队列(消费者)还没有处理完一个消息或者没有应答签收一个消息,则RabbitMQ不会分配新的消息给该队列.

如果所有的消费者都没有完成手上的消息,生产者还在不停地生产消息,队列还在不停地添加新任务,这是不会给消费者分发消息,就有可能导致队列被撑爆.

这时就只能添加新的工作队列或者改变存储策略了.

设置不公平分发:

public class Consumer1 {
    public static final String QueueName = "CJ_QUEUE";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtil.RabbitMQ_getChannel();
        System.out.println("消费消息时间较长-----------");
        DeliverCallback deliverCallback = ((consumerTag,message)->{
            RabbitMQUtil.sleep(10);
            System.out.println("消费消息:" + new String(message.getBody()));
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        });
        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消息消费被中断");
        };
        //设置不公平分发
        int prefetchCount = 1;
        channel.basicQos(prefetchCount);
        //手动应答
        channel.basicConsume(QueueName,false,deliverCallback,cancelCallback);
    }
}

public class Consumer2 {
    public static final String QueueName = "CJ_QUEUE";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtil.RabbitMQ_getChannel();
        System.out.println("消费消息时间较短-----------");
        DeliverCallback deliverCallback = ((consumerTag,message)->{
            RabbitMQUtil.sleep(1);
            System.out.println("消费消息:" + new String(message.getBody()));
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        });
        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消息消费被中断");
        };
        //设置不公平分发
        int prefetchCount = 1;
        channel.basicQos(prefetchCount);

        //采取手动应答
        channel.basicConsume(QueueName,false,deliverCallback,cancelCallback);
    }
}

预期值分发

带权的消息分发

默认的消息发送是异步的,所以在任何时候,channel中不止一个来自消费者收到确认的消息,因此这里就存在一个未确认的消息缓冲区.

因此我们希望限制这里的缓冲区的大小,避免缓冲区中无休止的未确认消息.

这时我们就可以通过basicqos()方法来设置(预取计数来完成).

basicqos方法里面设置通道上允许的未确认消息的最大数量,一旦数据达到配置的数量,RabbitMQ将停止在通道上传递更多的消息.除非有未确认的消息被确认.

例如:假设此时通道上未确认的消息有 4,6,7,9,5,10,并且通道上设置预取计数值为6,这时RabbitMQ将不会在该通道上传递消息.除非这里的消息被确认一个,RabbitMQ将感知到这一变化,并且在发送一条信息.

消息应答和Qos预取值对用户的吞吐量有着重大影响.

不公平分发和预取值分发都用到 basic.qos 方法，如果取值为 1，代表不公平分发，取值不为1，代表预取值分发

public class Consumer2 {
    public static final String QueueName = "CJ_QUEUE";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtil.RabbitMQ_getChannel();
        System.out.println("消费消息时间较短-----------");
        DeliverCallback deliverCallback = ((consumerTag,message)->{
            RabbitMQUtil.sleep(1);
            System.out.println("消费消息:" + new String(message.getBody()));
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        });
        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消息消费被中断");
        };
        //设置不公平分发
        /*int prefetchCount = 1;
        channel.basicQos(prefetchCount);*/

        //设置预期值分发 值为4
        int prefetchCount = 4;
        channel.basicQos(prefetchCount);

        //采取手动应答
        channel.basicConsume(QueueName,false,deliverCallback,cancelCallback);
    }
}

发布确认

生产者发送消息到RabbitMQ后,需要RabbitMQ返回一个ack,表示RabbitMQ已经收到生产者发送的消息.这样生产者就知道自己发送的信息成功了.

发布确认逻辑

如果消息和队列是可持久化的,那么确认消息会在消息写入磁盘之后发出.

broker 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 broker 也可以设置 basic.ack 的 multiple 域，表示到这个序列号之前的所有消息都已经得到了处理。

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

发布确认默认是没有开启的,如果要开启需要调用方法confirmSelect(),

channel.confirmSelect();

单个发布确认

这是一种简单的确认发布,他是一种同步确认发布的方式,也就是发布一个消息之后,只有它被确认,后续的消息才能被发布出去.

waitforconfirms()这个方法只有在消息被确认的时候才返回.

如果在指定的时间范围内没有返回,则抛出异常.

这种确认的方式就是发布速度特别慢,因为如果没有确认发布的消息,那么其他消息就只能阻塞等待.

这种方式最多每秒不超过百条的发送量.


/**
 * Created with IntelliJ IDEA.
 *
 * @Author: DongGuoZhen
 * @Date: 2024/01/05/10:28
 * @Description: 单个确认发布
 */
//确认发布指的是成功发送到了队列，并不是消费者消费了消息。
public class Producer1 {
    public static final int MESSAGE_MAX = 1000;
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        pushMessage();
    }

    private static void pushMessage() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMQUtil.RabbitMQ_getChannel();
        String QueueName = UUID.randomUUID().toString();
        channel.queueDeclare(QueueName,false,true,false,null);
        //开启发布确认
        channel.confirmSelect();

        //开始时间
        long start = System.currentTimeMillis();
        //依次发送1000个消息
        for (int i = 0; i < 1000; i++) {
            String message = i+"";
            channel.basicPublish("",QueueName,null,message.getBytes());
            //发送单个消息立马进行发布确认
            boolean flag = channel.waitForConfirms();
            if(flag) {  //如果成功 true
                System.out.println("消息: "+ message + " 成功发送到队列:" + QueueName);
            }
        }
        long end = System.currentTimeMillis();
        System.out.println("发布: " + MESSAGE_MAX + "条消息共耗时:"+ (end-start)+ "ms");
    }
}

批量发布确认

单个确认发布的速度非常慢,与单个确认等待相比,如果发送一批然后在一起确认,这样就大大提高了消息的发送速度和吞吐量.

当然这种方式的缺点就是,当有一个消息出现问题时,我们无法知道是那个消息没有发布出去.

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: DongGuoZhen
 * @Date: 2024/01/05/10:28
 * @Description: 批量确认发布
 */
//确认发布指的是成功发送到了队列，并不是消费者消费了消息。
public class Producer2 {
    public static final int MESSAGE_MAX = 1000;
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        pushMessage();
    }

    private static void pushMessage() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMQUtil.RabbitMQ_getChannel();
        String QueueName = UUID.randomUUID().toString();
        channel.queueDeclare(QueueName,false,true,false,null);
        //开启发布确认
        channel.confirmSelect();
        int bachSize = 100;  //批量发布确认的条数
        //开始时间
        long start = System.currentTimeMillis();
        //依次发送1000个消息
        for (int i = 0; i < 1000; i++) {
            String message = i+"";
            channel.basicPublish("",QueueName,null,message.getBytes());
            if((i+1) % bachSize == 0) {  //当发送的消息达到100条,进行确认发布一次
                channel.waitForConfirms();   //发布确认
                System.out.println("第"+ i+"条消息被确认");
            }
        }
        long end = System.currentTimeMillis();
        System.out.println("发布: " + MESSAGE_MAX + "条消息共耗时:"+ (end-start)+ "ms");
    }
}

异步发布确认

异步确认虽然编程上逻辑比较复杂,但是性价比最高,可靠性和效率都很好,利用了回调函数来达到消息的可靠性传递.


/**
 * Created with IntelliJ IDEA.
 *
 * @Author: DongGuoZhen
 * @Date: 2024/01/05/11:02
 * @Description: 异步确认发布
 */
public class Producer3 {
    public static final int MESSAGE_MAX = 1000;

    public static void main(String[] args) throws IOException, TimeoutException {
        pushMessage();
    }

    private static void pushMessage() throws IOException, TimeoutException {
        Channel channel = RabbitMQUtil.RabbitMQ_getChannel();
        String QueueName = UUID.randomUUID().toString();
        channel.queueDeclare(QueueName,false,true,false,null);
        channel.confirmSelect();
        long start = System.currentTimeMillis();
        /**
         * deliveryTag 消息的标记
         * multiple 是否为批量确认
         */
        ConfirmCallback ackCallback = (deliveryTag,multiple) ->{
            System.out.println("确认的消息:" + deliveryTag);
        };
        ConfirmCallback nackCallback = (deliveryTag,multiple) ->{
            System.out.println("未确认的消息:" + deliveryTag);
        };
        //消息监听器  监听那些消息成功了,那些消息失败了
        channel.addConfirmListener(ackCallback,nackCallback);
        //批量发送消息
        for (int i = 0; i < 1000; i++) {
            String message = i+"消息";
            channel.basicPublish("", QueueName,null,message.getBytes());
        }
        long end = System.currentTimeMillis();
        System.out.println("异步确认发布: " + MESSAGE_MAX + "条消息共耗时:"+ (end-start)+ "ms");
    }
}