RabbitMQ 基础教程
首页 死磕 Java 优质系列 Java 技术总览 Java 面试宝典 关于我

RabbitMQ 基础教程-消息应答

转
 2024-01-14  阅读(8)
©版权
原文作者：Java充电社 原文地址： http://www.itsoku.com/course/22/384

对应视频：https://www.bilibili.com/video/BV1cb4y1o7zz?p=22

6.1、概念

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

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

6.2、自动应答

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

6.3、消息应答的方法

  • Channel.basicAck(用于肯定确认)

    RabbitMQ 已知道该消息并且成功的处理消息，可以将其丢弃了

  • Channel.basicNack(用于否定确认)

  • Channel.basicReject(用于否定确认)

    与 Channel.basicNack 相比少一个参数，不处理该消息了直接拒绝，可以将其丢弃了

6.4、Multiple 的解释

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

202401142236174791.png

multiple 的 true 和 false 代表不同意思

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

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

  • false 同上面相比

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

202401142236181642.png

6.5、消息自动重新入队

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

202401142236188203.png

6.6、消息手动应答代码

默认消息采用的是自动应答，所以我们要想实现消息消费过程中不丢失，需要把自动应答改为手动应答，消费者在上面代码的基础上增加下面画红色部分代码。

202401142236194254.png

消息生产者

    public class Task02 {
        private static final String TASK_QUEUE_NAME = "ack_queue";
        public static void main(String[] argv) throws Exception {
            try (Channel channel = RabbitMqUtils.getChannel()) {
                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();
                    channel.basicPublish("", TASK_QUEUE_NAME, null, message.getBytes("UTF-8"));
                    System.out.println("生产者发出消息" + message);
                }
            }
        }
    }

消费者 01

    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=(consumerTag,delivery)->{
                String message= new String(delivery.getBody());
                SleepUtils.sleep(1);
                System.out.println("接收到消息:"+message);
                /**
                * 1.消息标记 tag
                * 2.是否批量应答未应答消息
                */
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
            };
            //采用手动应答
            boolean autoAck=false;
            channel.basicConsume(ACK_QUEUE_NAME,autoAck,deliverCallback,(consumerTag)->{
                System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
            });
        }
    }

消费者 02

    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=(consumerTag,delivery)->{
                String message= new String(delivery.getBody());
                SleepUtils.sleep(30);
                System.out.println("接收到消息:"+message);
                /**
                * 1.消息标记 tag
                * 2.是否批量应答未应答消息
                */
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
            };
            //采用手动应答
            boolean autoAck=false;
            channel.basicConsume(ACK_QUEUE_NAME,autoAck,deliverCallback,(consumerTag)->{
                System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
            });
        }
    }

睡眠工具类

    public class SleepUtils {
        public static void sleep(int second){
            try {
                Thread.sleep(1000*second);
            } catch (InterruptedException _ignored) {
                Thread.currentThread().interrupt();
            }
        }
    }

6.7、手动应答效果演示

正常情况下消息发送方发送两个消息 C1 和 C2 分别接收到消息并进行处理

202401142236201615.png

在发送者发送消息 dd，发出消息之后的把 C2 消费者停掉，按理说该 C2 来处理该消息，但是由于它处理时间较长，在还未处理完，也就是说 C2 还没有执行 ack 代码的时候， C2 被停掉了，此时会看到消息被 C1 接收到了，说明消息 dd 被重新入队，然后分配给能处理消息的 C1 处理了

202401142236207716.png

202401142236213477.png

202401142236218388.png

Java 面试宝典是大明哥全力打造的 Java 精品面试题，它是一份靠谱、强大、详细、经典的 Java 后端面试宝典。它不仅仅只是一道道面试题，而是一套完整的 Java 知识体系，一套你 Java 知识点的扫盲贴。

它的内容包括：

  • 大厂真题：Java 面试宝典里面的题目都是最近几年的高频的大厂面试真题。
  • 原创内容：Java 面试宝典内容全部都是大明哥原创，内容全面且通俗易懂，回答部分可以直接作为面试回答内容。
  • 持续更新：一次购买，永久有效。大明哥会持续更新 3+ 年，累计更新 1000+，宝典会不断迭代更新，保证最新、最全面。
  • 覆盖全面：本宝典累计更新 1000+，从 Java 入门到 Java 架构的高频面试题，实现 360° 全覆盖。
  • 不止面试：内容包含面试题解析、内容详解、知识扩展，它不仅仅只是一份面试题，更是一套完整的 Java 知识体系。
  • 宝典详情：https://www.yuque.com/chenssy/sike-java/xvlo920axlp7sf4k
  • 宝典总览：https://www.yuque.com/chenssy/sike-java/yogsehzntzgp4ly1
  • 宝典进展：https://www.yuque.com/chenssy/sike-java/en9ned7loo47z5aw

目前 Java 面试宝典累计更新 400+ 道，总字数 42w+。大明哥还在持续更新中，下图是大明哥在 2024-12 月份的更新情况：

想了解详情的小伙伴，扫描下面二维码加大明哥微信【daming091】咨询

同时，大明哥也整理一套目前市面最常见的热点面试题。微信搜[大明哥聊 Java]或扫描下方二维码关注大明哥的原创公众号[大明哥聊 Java] ，回复【面试题】 即可免费领取。

阅读全文
文章标签： MQRabbitMQ