更多详情内容请访问:RabbitMQ 系列文章导读

1、消息应答

1.1 概念

消费者完成一个任务可能需要一段时间,如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了,会发生这么一种情况:RabbitMQ 一旦向消费者传递了一条消息,便立即将该消息标记为删除。在这种情况下,突然有个消费者挂掉了,我们将丢失正在处理的消息。以及后续发送给该消费这的消息,因为它无法接收到。

为了保证消息在发送过程中不丢失,引入消息应答机制,消息应答就是:消费者在接收到消息并且处理该消息之后,告诉 RabbitMQ 它已经处理了,RabbitMQ 可以把该消息删除了。

1.2 自动应答

消息发送后立即被认为已经传送成功,这种模式需要在高吞吐量和数据传输安全性方面做权衡,因为这种模式如果消息在接收到之前,消费者那边出现连接或者 channel 关闭,那么消息就丢失了,当然另一方面这种模式消费者那边可以传递过载的消息,没有对传递的消息数量进行限制,当然这样有可能使得消费者这边由于接收太多还来不及处理的消息,导致这些消息的积压,最终使得内存耗尽,最终这些消费者线程被操作系统杀死,所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。

1.3 手动应答

手动应答的好处是可以批量应答并且减少网络拥堵

  • void basicAck(long var1, boolean var3):用于肯定确认,RabbitMQ 已知道该消息并且成功的处理消息,可以将其丢弃了
  • void basicNack(long var1, boolean var3, boolean var4):用于否定确认
  • void basicReject(long var1, boolean var3):用于否定确认,与 basicNack() 相比少一个参数,不处理该消息了直接拒绝,可以将其丢弃了

Multiple 的解释

image-20220824131509597

  • true:代表批量应答 channel 上未应答的消息

比如说 channel 上有传送 tag 的消息 5、6、7、8 当前 tag 是 8 ,那么此时 5-8 的这些还未应答的消息都会被确认收到消息应答

  • false:只会应答 tag=8 的消息 5、6、7 这三个消息依然不会被确认收到消息应答

1.4 消息自动重新入列

如果消费者由于某些原因失去连接(其通道已关闭,连接已关闭或 TCP 连接丢失),导致消息未发送 ACK 确认,RabbitMQ 将了解到消息未完全处理,并将对其重新排队。如果此时其他消费者可以处理,它将很快将其重新分发给另一个消费者。这样,即使某个消费者偶尔死亡,也可以确保不会丢失任何消息。

image-20220824132451389

1.5 使用案例

public class Task02 {

    private static final String TASK_QUEUE_NAME = "ack_queue";

    public static void main(String[] args) {
        try {
            Channel channel = RabbitMqUtils.getChannel();
            /**
             * 生成一个队列
             * 1.队列名称
             * 2.队列里面的消息是否持久化(磁盘),默认情况下消息存储在内存中
             * 3.该队列是否只供一个消费者进行消费,是否进行消息共享。true:可以多个消费者; false:只能一个消费者消费
             * 4.是否自动删除,最后一个消费者断开连接以后,该队是否自动删除。true:自动删除;false:不删除
             * 5.其他参数
             */
            channel.queueDeclare(TASK_QUEUE_NAME, false, false, false, null);
            Scanner sc = new Scanner(System.in);
            System.out.println("请输入信息");
            while (sc.hasNext()) {
                String message = sc.nextLine();
                /**
                 * 发送一个消费
                 * 1.发送到哪个交换机
                 * 2.路由的key值是那个,本次是队列的名称
                 * 3.其他参数信息
                 * 4.发送消息的消息体
                 */
                channel.basicPublish("", TASK_QUEUE_NAME, null, message.getBytes("UTF-8"));
                System.out.println("生产者发出消息:" + message);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

public class Work03 {

    private static final String ACK_QUEUE_NAME = "ack_queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("c1 等待接收消息处理时间较短");

        // 消息消费的时候如何处理消息
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String s, Delivery delivery) throws IOException {
                try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
                System.out.println("接收到消息:" + new String(delivery.getBody()));
                /**
                 * 1.消息标记 tag
                 * 2.是否批量应答未应答消息
                 */
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        };

        // 消息消费被取消时执行的操作
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String s) throws IOException {
                System.out.println(s + "消费者取消消费接口回调逻辑");
            }
        };

        // 采用手动应答
        boolean autoAck = false;
        /**
         * 消费者消费消息
         * 1.消费哪个队列
         * 2.消费成功之后是否自动应答。true:代表自动应答;false:手动应答
         * 3.消费者未成功消费的回调
         * 4.消费者取消消费者的回调
         */
        channel.basicConsume(ACK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);
    }
}

public class Work04 {

    private static final String ACK_QUEUE_NAME = "ack_queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("c2 等待接收消息处理时间长");

        // 消息消费的时候如何处理消息
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String s, Delivery delivery) throws IOException {
                try { TimeUnit.SECONDS.sleep(15); } catch (InterruptedException e) { e.printStackTrace(); }
                System.out.println("接收到消息:" + new String(delivery.getBody()));
                /**
                 * 1.消息标记 tag
                 * 2.是否批量应答未应答消息
                 */
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        };

        // 消息消费被取消时执行的操作
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String s) throws IOException {
                System.out.println(s + "消费者取消消费接口回调逻辑");
            }
        };

        // 采用手动应答
        boolean autoAck = false;
        /**
         * 消费者消费消息
         * 1.消费哪个队列
         * 2.消费成功之后是否自动应答。true:代表自动应答;false:手动应答
         * 3.消费者未成功消费的回调
         * 4.消费者取消消费者的回调
         */
        channel.basicConsume(ACK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);
    }
}

image-20220824133511472

当发送者发送消息 DD 到队列,此时是 second 来消费该消息,但是由于它处理时间较长,在还未处理完时间里停止运行,也就是说 second 还没有执行到 ack 代码的时候,second 被停掉了,此时会看到消息被 first 接收到了,说明消息 DD 被重新入队,然后分配给能处理消息的 first 处理了

2、RabbitMQ 持久化

当 RabbitMQ 服务停掉以后,消息生产者发送过来的消息不丢失要如何保障?默认情况下 RabbitMQ 退出或由于某种原因崩溃时,它忽视队列和消息,除非告知它不要这样做。确保消息不会丢失需要做两件事:我们需要将队列和消息都标记为持久化。

2.1 队列持久化

之前我们创建的队列都是非持久化的,RabbitMQ 如果重启的化,该队列就会被删除掉,如果要队列实现持久化需要在声明队列的时候把 durable 参数设置为true,代表开启持久化

image-20220824143640112

注意:如果之前声明的队列不是持久化的,需要把原先队列先删除,或者重新创建一个持久化的队列

2.2 消息持久化

要想让消息持久化需要在消息生产者修改代码,在 basicPublish 方法的第二个参数添加这个属性: MessageProperties.PERSISTENT_TEXT_PLAIN

image-20220824143658638

将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉 RabbitMQ 将消息保存到磁盘,但是这里依然存在当消息刚准备存储在磁盘的时候 但是还没有存储完,消息还在缓存的一个间隔点。此时并没 有真正写入磁盘。持久性保证并不强,但是对于我们的简单任务队列而言,这已经绰绰有余了。

3、消息分发

3.1 轮询消费

轮询消费消息指的是轮流消费消息,即每个工作队列都会获取一个消息进行消费,并且获取的次数按照顺序依次往下轮流。

3.2 不公平分发

如果一个工作队列还没有处理完或者没有应答签收一个消息,则不拒绝 RabbitMQ 分配新的消息到该工作队列。此时 RabbitMQ 会优先分配给其他已经处理完消息或者空闲的工作队列。如果所有的消费者都没有完成手上任务,队列还在不停的添加新任务,队列有可能就会遇到队列被撑满的情况,这个时候就只能添加新的 worker(工作队列)或者改变其他存储任务的策略。

设置不公平分发,在消费者中消费消息之前设置好参数channel.basicQos(1);

image-20220824144618722

成功开启后,在 Web 页面中可以看到如下结果:

image-20220824145407894

3.3 预取值分发

默认消息的发送是异步发送的,所以在任何时候,channel 上不止只有一个消息来自消费者的手动确认,所以本质上是异步的。因此这里就存在一个未确认的消息缓冲区,因此希望开发人员能限制此缓冲区的大小以避免缓冲区里面无限制的未确认消息问题。这个时候就可以通过使用 basic.qos 方法设置「预取计数」值来完成的。

该值定义通道上允许的未确认消息的最大数量。一旦数量达到配置的数量, RabbitMQ 将停止在通道上传递更多消息,除非至少有一个未处理的消息被确认,例如,假设在通道上有未确认的消息 5、6、7,8,并且通道的预取计数设置为 4,此时 RabbitMQ 将不会在该通道上再传递任何消息,除非至少有一个未应答的消息被 ack。比方说 tag=6 这个消息刚刚被确认 ACK,RabbitMQ 将会感知这个情况到并再发送一条消息。消息应答和 QoS 预取值对用户吞吐量有重大影响。

通常,增加预取将提高向消费者传递消息的速度。虽然自动应答传输消息速率是最佳的,但是,在这种情况下已传递但尚未处理的消息的数量也会增加,从而增加了消费者的 RAM 消耗(随机存取存储器)。应该小心使用具有无限预处理的自动确认模式或手动确认模式,消费者消费了大量的消息如果没有确认的话,会导致消费者连接节点的内存消耗变大,所以找到合适的预取值是一个反复试验的过程,不同的负载该值取值也不同 100 到 300 范围内的值通常可提供最佳的吞吐量,并且不会给消费者带来太大的风险。

预取值为 1 是最保守的。当然这将使吞吐量变得很低,特别是消费者连接延迟很严重的情况下,特别是在消费者连接等待时间较长的环境 中。对于大多数应用来说,稍微高一点的值将是最佳的。

不公平分发和预取值分发都用到 basicQos() 方法,如果取值为 1,代表不公平分发,取值不为1,代表预取值分发

4、消息应答与分布的区别

  • 应答功能属于消费者,消费完消息告诉 RabbitMQ 已经消费成功。
  • 发布功能属于生产者,生产消息到 RabbitMQ,RabbitMQ 需要告诉生产者已经收到消息。

END

本文作者:
文章标题:消息应答
本文地址:https://www.pendulumye.com/rabbitmq/509.html
版权说明:若无注明,本文皆PendulumYe原创,转载请保留文章出处。
最后修改:2022 年 08 月 30 日
千山万水总是情,给个一毛行不行💋