【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列

这篇具有很好参考价值的文章主要介绍了【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列

                                                                  💧 【 R a b b i t M Q 教 程 】 第 五 章 — — R a b b i t M Q − 死 信 队 列 \color{#FF1493}{【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列} RabbitMQRabbitMQ💧          


🌷 仰望天空,妳我亦是行人.✨
🦄 个人主页——微风撞见云的博客🎐
🐳 《数据结构与算法》专栏的文章图文并茂🦕生动形象🦖简单易学!欢迎大家来踩踩~🌺
💧 《Java学习笔记》专栏的文章是本人在Java学习中总结的一些知识点~ 💐
🥣 《每天一点小知识》专栏的文章可以丰富你的知识库,滴水成河~ 🌊
🥕 《RabbitMQ》专栏的文章是在学习尚硅谷课程时整理的笔记,方便复习巩固~ 🍑
🪁 希望本文能够给读者带来一定的帮助~🌸文章粗浅,敬请批评指正!🐥



🌊RabbitMQ - 死信队列

死信的概念

先从概念解释上搞清楚这个定义,死信,顾名思义就是无法被消费的消息,字面意思可以这样理 解,一般来说,producer 将消息投递到 broker 或者直接到queue 里了,consumer 从 queue 取出消息 进行消费,但某些时候由于特定的原因导致 queue 中的某些消息无法被消费,这样的消息如果没有后续的处理,就变成了死信,有死信自然就有了死信队列。

应用场景:为了保证订单业务的消息数据不丢失,需要使用到 RabbitMQ 的死信队列机制,当消息消费发生异常时,将消息投入死信队列中。还有比如说:用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

#死信的来源

死信的来源

  • 消息 TTL 过期

    TTL是Time To Live的缩写, 也就是生存时间

  • 队列达到最大长度

    队列满了,无法再添加数据到 mq 中

  • 消息被拒绝

    (basic.reject 或 basic.nack) 并且 requeue=false.

死信实战

【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列

死信之 T T l

消费者 C1 代码:

/**
 * 死信队列 - 消费者01
 */
public class Consumer01 {

    //普通交换机名称
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    //死信交换机名称
    private static final String DEAD_EXCHANGE = "dead_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();

        //声明死信和普通交换机 类型为 direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        //声明死信队列
        String deadQueue = "dead-queue";
        channel.queueDeclare(deadQueue, false, false, false, null);
        //死信队列绑定:队列、交换机、路由键(routingKey)
        channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");

        //正常队列绑定死信队列信息
        Map<String, Object> params = new HashMap<>();
        //正常队列设置死信交换机 参数 key 是固定值
        params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //正常队列设置死信 routing-key 参数 key 是固定值
        params.put("x-dead-letter-routing-key", "lisi");

        //正常队列
        String normalQueue = "normal-queue";
        channel.queueDeclare(normalQueue, false, false, false, params);
        channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");

        System.out.println("等待接收消息........... ");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Consumer01 接收到消息" + message);
        };
        channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {
        });
    }

}

生产者代码

public class Producer {
    private static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] argv) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();

        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        //设置消息的 TTL 时间 10s
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
        //该信息是用作演示队列个数限制
        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
            channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes());
            System.out.println("生产者发送消息:" + message);
        }

    }
}

启动 C1 ,之后关闭消费者,模拟其接收不到消息。再启动 Producer
【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列
消费者 C2 代码:

以上步骤完成后,启动 C2 消费者,它消费死信队列里面的消息

public class Consumer02 {
    //死信交换机名称
    private static final String DEAD_EXCHANGE = "dead_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();

        //声明交换机
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        //声明队列
        String deadQueue = "dead-queue";
        channel.queueDeclare(deadQueue, false, false, false, null);
        channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");

        System.out.println("等待接收死信消息........... ");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Consumer02 接收到消息" + message);
        };
        channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> {
        });
    }
}

【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列

死信之最大长度

1、消息生产者代码去掉 TTL 属性
【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列
2、C1 消费者修改以下代码**(启动之后关闭该消费者 模拟其接收不到消息)**
【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列

//设置正常队列的长度限制,例如发10个,4个则为死信
params.put("x-max-length",6);

注意此时需要把原先队列删除 因为参数改变了

3、C2 消费者代码不变(启动 C2 消费者)
【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列

死信之消息被拒

1、消息生产者代码同上生产者一致

2、C1 消费者代码(启动之后关闭该消费者 模拟其接收不到消息)

拒收消息 “info5”

public class Consumer01 {

    //普通交换机名称
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    //死信交换机名称
    private static final String DEAD_EXCHANGE = "dead_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();

        //声明死信和普通交换机 类型为 direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        //声明死信队列
        String deadQueue = "dead-queue";
        channel.queueDeclare(deadQueue, false, false, false, null);
        //死信队列绑定:队列、交换机、路由键(routingKey)
        channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
        
        //正常队列绑定死信队列信息
        Map<String, Object> params = new HashMap<>();
        //正常队列设置死信交换机 参数 key 是固定值
        params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //正常队列设置死信 routing-key 参数 key 是固定值
        params.put("x-dead-letter-routing-key", "lisi");
//        //设置正常队列的长度限制,例如发10个,4个则为死信
//        params.put("x-max-length",6);
        
        //正常队列
        String normalQueue = "normal-queue";
        channel.queueDeclare(normalQueue, false, false, false, params);
        channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");

        System.out.println("等待接收消息........... ");
        
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            if (message.equals("info5")) {
                System.out.println("Consumer01 接收到消息" + message + "并拒绝签收该消息");
                //requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中
                channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);
            } else {
                System.out.println("Consumer01 接收到消息" + message);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }

        };
        //开启手动应答
        channel.basicConsume(normalQueue, false, deliverCallback, consumerTag -> {
        });
    }

}

【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列
3、C2 消费者代码不变

启动消费者 1 然后再启动消费者 2

【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列


【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列


🐳结语

🐬初学一门技术时,总有些许的疑惑,别怕,它们是我们学习路上的点点繁星,帮助我们不断成长。

🐟文章粗浅,希望对大家有帮助!

💧下一篇 -->【RabbitMQ教程】第四章 —— RabbitMQ - 延迟队列文章来源地址https://www.toymoban.com/news/detail-487167.html

到了这里,关于【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Rabbitmq死信队列及延时队列实现

    问题:什么是延迟队列 我们常说的延迟队列是指消息进入队列后不会被立即消费,只有达到指定时间后才能被消费。 但RabbitMq中并 没有提供延迟队列功能 。那么RabbitMQ如何实现延迟队列 通过:死信队列 + RabbitMQ的TTL特性实现。 实现原理 给一个普通带有过期功能的队列绑定一

    2024年02月15日
    浏览(48)
  • 【RabbitMQ笔记10】消息队列RabbitMQ之死信队列的介绍

    这篇文章,主要介绍消息队列RabbitMQ之死信队列。 目录 一、RabbitMQ死信队列 1.1、什么是死信队列 1.2、设置过期时间TTL 1.3、配置死信交换机和死信队列(代码配置) (1)设置队列过期时间 (2)设置单条消息过期时间 (3)队列设置死信交换机 (4)配置的基本思路 1.4、配置

    2024年02月16日
    浏览(82)
  • 【RabbitMQ学习日记】——死信队列与延迟队列

    死信,顾名思义就是无法被消费的消息,字面意思可以这样理解,一般来说, producer 将消息投递到 broker 或者直接到 queue 里了, consumer 从 queue 取出消息进行消费,但某些时候 由于特定的原因导致 queue 中的某些消息无法被消费 ,这样的消息如果没有后续的处理,就变成了死

    2024年02月06日
    浏览(56)
  • RabbitMQ:死信队列

    📃个人主页:不断前进的皮卡丘 🌞博客描述:梦想也许遥不可及,但重要的是追梦的过程,用博客记录自己的成长,记录自己一步一步向上攀登的印记 🔥个人专栏:消息中间件 队列中不能被消费的消息称为死信队列 有时候因为特殊原因,可能导致队列中的某些信息无法被消费

    2024年02月02日
    浏览(44)
  • RabbitMQ 死信队列实现

    死信队列使用注解实现 报错: 可以使用注解的方式来绑定 死信队列,但是还是会报上面的错误,继续修改 参数试试 java - How to set x-dead-letter-exchange in Rabbit? - Stack Overflow 但是使用注解绑定的话好像又不生效了,问题原因,tmd将死信参数绑到交换机上了,c 修改代码  至于问题

    2024年02月02日
    浏览(34)
  • rabbitmq的死信队列

    目录 成为死信的条件  消息TTL过期   队列达到最大长度  消息被拒 延迟队列  延迟队列使用场景  消息设置 TTL 队列设置 TTL  两者区别   producer 将消息投递到 broker 或者直接到 queue 里了, consumer 从 queue 取出消息 进行消费,但某些时候由于特定的 原因导致 queue 中的某些消

    2024年02月12日
    浏览(36)
  • RabbitMQ: 死信队列

    其实就是一个普通的队列,绑定号私信交换机,不给ttl,给上匹配的路由,等待交换机发送消息。 1.在消费者里的RabbitMQConfig配置类里,创建队列,给它加参数 第四个参数,就是放入这个队列,的一些属性参数 也就是这两个位置 对应Java代码里好像少个参数,排他性,是指,

    2024年02月09日
    浏览(44)
  • RabbitMQ进阶——死信队列

    在消息队列中,执行异步任务时,通常是将消息生产者发布的消息存储在队列中,由消费者从队列中获取并处理这些消息。但是,在某些情况下,消息可能无法正常地被处理和消耗,例如:格式错误、设备故障等,这些未成功处理的消息就被称为“死信”。 为了避免这些未成

    2024年04月13日
    浏览(32)
  • rabbitMQ引入死信队列

            指的是,从队列当中取出来的消息,到达消费方后,因为某些原因导致消息并没有被正常消费掉,这些没有被后续处理的消息就是“死信”,而保存死信的队列,就是死信队列。         为了保证订单业务的消息数据不丢失,需要使用死信队列机制,在消息消费发生

    2024年02月02日
    浏览(45)
  • RabbitMQ——死信队列

    死信队列(Dead Letter Queue,DLQ)是 RabbitMQ 中的一种重要特性,用于处理无法被消费的消息,防止消息丢失。 死信的来源 在消息队列中,当消息满足一定条件而无法被正常消费时,这些消息会被发送到死信队列。满足条件的情况包括但不限于: 消息被拒绝( basic.reject 或 bas

    2024年03月14日
    浏览(55)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包