0
0

RabbitMQ系列（15）--死信队列的简介与死信队列和死信消费者的实现

1、死信的概念

死信，顾名思义就是无法被消费的消息，一般来说producer（生产者）将消息投递到broker或直接放到queue（队列）中，consumer（消费者）从queue（队列）取出消息进行消费，但某些时候由于特定的原因导致queue（队列）中的消息无法被消费，若这些消息没有后续的处理，则这些消息就变成了死信，有死信自然就有了死信队列

2、死信的应用场景

为保证订单业务的消息数据不丢失，需要使用RabbitMQ的死信队列机制，当消息发生异常时，将消息投入死信队列中

3、死信的来源

(1)消息TTL（存活时间）过期

(2)队列达到最大长度（队列满了，无法再添加数据到mq中）

(3)消息被拒绝（basic.reject或basic.nack）并且requeue=false

4、死信队列的实现

(1)我们将根据这张死信队列的代码架构图来实现死信队列

(2)新建一个名为dead的包，用于装实现死信队列的代码

效果图：

(3)新建一个名为Consumer01的类用于编写消费者的代码

代码如下：

注：RabbitMqUtils工具类的实现在我的另一篇文章里，有需要的同学可以查看参考

RabbitMQ系列（6）--RabbitMQ模式之工作队列(Work queues)的简介及实现_Ken_1115的博客-CSDN博客

package com.ken.dead;

import com.ken.utils.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;

import java.util.HashMap;
import java.util.Map;

public class Consumer01 {

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

    //普通队列的名称
    public static  final String NORMAL_QUEUE = "normal_queue";
    //死信队列的名称
    public static final String DEAD_QUEUE = "dead_queue";

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

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

        //用于在消息成为死信后，把消息转发到死信交换机dead_exchange里
        Map<String, Object> arguments = new HashMap<>();
        //正常队列设置死信交换机
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //设置死信routingkey
        arguments.put("x-dead-letter-routing-key","dead");
        //设置正常队列长度的限制
        //arguments.put("x-max-length",6);
        /**
         * 声明普通队列
         * 第一个参数：队列名称
         * 第二个参数：服务器重启后队列是否还存在，即队列是否持久化,true为是，false为否，默认false，即消息存储在内存中而不是硬盘中
         * 第三个参数：该队列是否只供一个消费者进行消费，是否进行消息共享，true为只允许一个消费者进行消费，false为允许多个消费者对队列进行消费，默认false
         * 第四个参数：是否自动删除，最后一个消费者断开连接后该队列是否自动删除，true自动删除，false不自动删除
         * 第五个参数：其他参数
         */
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
        /**
         * 声明死信队列
         * 第一个参数：队列名称
         * 第二个参数：服务器重启后队列是否还存在，即队列是否持久化,true为是，false为否，默认false，即消息存储在内存中而不是硬盘中
         * 第三个参数：该队列是否只供一个消费者进行消费，是否进行消息共享，true为只允许一个消费者进行消费，false为允许多个消费者对队列进行消费，默认false
         * 第四个参数：是否自动删除，最后一个消费者断开连接后该队列是否自动删除，true自动删除，false不自动删除
         * 第五个参数：其他参数
         */
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        //普通队列与普通交换机通过routingkey进行捆绑
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"normal");
        //死信队列与死信交换机通过routingkey进行捆绑
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"dead");

        /**
         * 声明消费者接收消息后的回调方法(由于回调方法DeliverCallback是函数式接口，所以需要给DeliverCallback赋值一个函数，为了方便我们这里使用Lambda表达式进行赋值)
         * 为什么要这样写呢，是因为basicConsume方法里的参数deliverCallback的类型DeliverCallback用 @FunctionalInterface注解规定DeliverCallback是一个函数式接口，所以要往deliverCallback参数传的值要是一个函数
         *
         * 以下是DeliverCallback接口的源代码
         *  @FunctionalInterface
         *  public interface DeliverCallback {
         *      void handle (String consumerTag, Delivery message) throws IOException;
         *  }
         */
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("Consumer01接收的消息是：" + new String(message.getBody(),"UTF-8"));
        };

        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,consumerTag -> {});
    }

}

(4)复制Consumer01类并粘贴重命名为Consumer02

​

代码如下：

package com.ken.dead;

import com.ken.utils.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.util.HashMap;
import java.util.Map;

public class Consumer02 {

    //普通队列的名称
    public static  final String NORMAL_QUEUE = "normal_queue";
    //死信队列的名称
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        /**
         * 声明消费者接收消息后的回调方法(由于回调方法DeliverCallback是函数式接口，所以需要给DeliverCallback赋值一个函数，为了方便我们这里使用Lambda表达式进行赋值)
         * 为什么要这样写呢，是因为basicConsume方法里的参数deliverCallback的类型DeliverCallback用 @FunctionalInterface注解规定DeliverCallback是一个函数式接口，所以要往deliverCallback参数传的值要是一个函数
         *
         * 以下是DeliverCallback接口的源代码
         *  @FunctionalInterface
         *  public interface DeliverCallback {
         *      void handle (String consumerTag, Delivery message) throws IOException;
         *  }
         */
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("Consumer02接收的消息是：" + new String(message.getBody(),"UTF-8"));
        };

        channel.basicConsume(DEAD_QUEUE,true,deliverCallback,consumerTag -> {});
    }

}

(5)新建一个名为Producer的类用于编写生产者的代码

​

代码如下：

package com.ken.dead;

import com.ken.utils.RabbitMqUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;

/**
 * 生产者
 */
public class Producer {

    //普通交换机的名称
    public static  final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //死信消息，设置TTL时间（存活时间），单位是ms 10000ms = 10s
        AMQP.BasicProperties properties= new AMQP.BasicProperties().builder().expiration("10000").build();
        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
            /**
             * 用信道对消息进行发布（消息持久化）
             * 第一个参数：发送到哪个交换机
             * 第二个参数：路由的Key值是哪个，本次是队列名
             * 第三个参数：其他参数信息
             * 第四个参数：发送消息的消息体
             */
            channel.basicPublish(NORMAL_EXCHANGE,"normal",properties,message.getBytes());
        }
        
    }

}

(6)先运行Consumer01，生成队列和交换机

​

​

​

(7)然后停止Consumer01，模拟消费者宕机

​

(8)运行Producer

​

(9)观察normal_queue队列和dead_queue队列消息数量的变化，一开始normal_queue队列里有10条消息，过了10s后消息都到了dead_queue队列里，证明消费者消费消息失败，消息从normal_queue队列移到了dead_queue队列里，由此可见当消息TTL过期后，死信队列成功运行

​

​

(10)启动Consumer02，可以看到Consumer02消费了dead_queue队列里的消息

(10)删除normal_queue队列

(11) 修改Produces的代码，把设置TTL时间的代码注释掉，basicPublish方法的第3个参数设置为null

代码如下：

package com.ken.dead;

import com.ken.utils.RabbitMqUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;

/**
 * 生产者
 */
public class Producer {

    //普通交换机的名称
    public static  final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //死信消息，设置TTL时间（存活时间），单位是ms 10000ms = 10s
        //AMQP.BasicProperties properties= new AMQP.BasicProperties().builder().expiration("10000").build();
        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
            /**
             * 用信道对消息进行发布（消息持久化）
             * 第一个参数：发送到哪个交换机
             * 第二个参数：路由的Key值是哪个，本次是队列名
             * 第三个参数：其他参数信息
             * 第四个参数：发送消息的消息体
             */
            channel.basicPublish(NORMAL_EXCHANGE,"normal",null,message.getBytes());
        }

    }

}

(12)修改Consumer01的代码，把之前注释的用于限制队列长度的代码放出来

代码如下：

package com.ken.dead;

import com.ken.utils.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;

import java.util.HashMap;
import java.util.Map;

public class Consumer01 {

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

    //普通队列的名称
    public static  final String NORMAL_QUEUE = "normal_queue";
    //死信队列的名称
    public static final String DEAD_QUEUE = "dead_queue";

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

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

        //用于在消息成为死信后，把消息转发到死信交换机dead_exchange里
        Map<String, Object> arguments = new HashMap<>();
        //正常队列设置死信交换机
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //设置死信routingkey
        arguments.put("x-dead-letter-routing-key","dead");
        //设置正常队列长度的限制
        arguments.put("x-max-length",6);
        /**
         * 声明普通队列
         * 第一个参数：队列名称
         * 第二个参数：服务器重启后队列是否还存在，即队列是否持久化,true为是，false为否，默认false，即消息存储在内存中而不是硬盘中
         * 第三个参数：该队列是否只供一个消费者进行消费，是否进行消息共享，true为只允许一个消费者进行消费，false为允许多个消费者对队列进行消费，默认false
         * 第四个参数：是否自动删除，最后一个消费者断开连接后该队列是否自动删除，true自动删除，false不自动删除
         * 第五个参数：其他参数
         */
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
        /**
         * 声明死信队列
         * 第一个参数：队列名称
         * 第二个参数：服务器重启后队列是否还存在，即队列是否持久化,true为是，false为否，默认false，即消息存储在内存中而不是硬盘中
         * 第三个参数：该队列是否只供一个消费者进行消费，是否进行消息共享，true为只允许一个消费者进行消费，false为允许多个消费者对队列进行消费，默认false
         * 第四个参数：是否自动删除，最后一个消费者断开连接后该队列是否自动删除，true自动删除，false不自动删除
         * 第五个参数：其他参数
         */
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        //普通队列与普通交换机通过routingkey进行捆绑
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"normal");
        //死信队列与死信交换机通过routingkey进行捆绑
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"dead");

        /**
         * 声明消费者接收消息后的回调方法(由于回调方法DeliverCallback是函数式接口，所以需要给DeliverCallback赋值一个函数，为了方便我们这里使用Lambda表达式进行赋值)
         * 为什么要这样写呢，是因为basicConsume方法里的参数deliverCallback的类型DeliverCallback用 @FunctionalInterface注解规定DeliverCallback是一个函数式接口，所以要往deliverCallback参数传的值要是一个函数
         *
         * 以下是DeliverCallback接口的源代码
         *  @FunctionalInterface
         *  public interface DeliverCallback {
         *      void handle (String consumerTag, Delivery message) throws IOException;
         *  }
         */
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("Consumer01接收的消息是：" + new String(message.getBody(),"UTF-8"));
        };

        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,consumerTag -> {});
    }

}

(13)停掉所有程序，然后重新运行Consumer01，生成队列和交换机

(14)然后停止Consumer01，模拟消费者宕机

​

(15)重新运行Producer

​

(16)观察normal_queue队列和dead_queue队列消息数量的变化，可以看到normal_queue队列只能堆积最多6条消息，而剩余的4条消息都移到了dead_queue队列里，由此可见当队列达到最大长度6条后，死信队列成功运行

(17)删除normal_queue队列和dead_queue队列

(18)修改Consumer01的代码，注释掉限制队列长度的代码，修改消息应答的部分代码

代码如下：

package com.ken.dead;

import com.ken.utils.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;

import java.util.HashMap;
import java.util.Map;

public class Consumer01 {

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

    //普通队列的名称
    public static  final String NORMAL_QUEUE = "normal_queue";
    //死信队列的名称
    public static final String DEAD_QUEUE = "dead_queue";

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

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

        //用于在消息成为死信后，把消息转发到死信交换机dead_exchange里
        Map<String, Object> arguments = new HashMap<>();
        //正常队列设置死信交换机
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //设置死信routingkey
        arguments.put("x-dead-letter-routing-key","dead");
        //设置正常队列长度的限制
        //arguments.put("x-max-length",6);
        /**
         * 声明普通队列
         * 第一个参数：队列名称
         * 第二个参数：服务器重启后队列是否还存在，即队列是否持久化,true为是，false为否，默认false，即消息存储在内存中而不是硬盘中
         * 第三个参数：该队列是否只供一个消费者进行消费，是否进行消息共享，true为只允许一个消费者进行消费，false为允许多个消费者对队列进行消费，默认false
         * 第四个参数：是否自动删除，最后一个消费者断开连接后该队列是否自动删除，true自动删除，false不自动删除
         * 第五个参数：其他参数
         */
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
        /**
         * 声明死信队列
         * 第一个参数：队列名称
         * 第二个参数：服务器重启后队列是否还存在，即队列是否持久化,true为是，false为否，默认false，即消息存储在内存中而不是硬盘中
         * 第三个参数：该队列是否只供一个消费者进行消费，是否进行消息共享，true为只允许一个消费者进行消费，false为允许多个消费者对队列进行消费，默认false
         * 第四个参数：是否自动删除，最后一个消费者断开连接后该队列是否自动删除，true自动删除，false不自动删除
         * 第五个参数：其他参数
         */
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        //普通队列与普通交换机通过routingkey进行捆绑
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"normal");
        //死信队列与死信交换机通过routingkey进行捆绑
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"dead");

        /**
         * 声明消费者接收消息后的回调方法(由于回调方法DeliverCallback是函数式接口，所以需要给DeliverCallback赋值一个函数，为了方便我们这里使用Lambda表达式进行赋值)
         * 为什么要这样写呢，是因为basicConsume方法里的参数deliverCallback的类型DeliverCallback用 @FunctionalInterface注解规定DeliverCallback是一个函数式接口，所以要往deliverCallback参数传的值要是一个函数
         *
         * 以下是DeliverCallback接口的源代码
         *  @FunctionalInterface
         *  public interface DeliverCallback {
         *      void handle (String consumerTag, Delivery message) throws IOException;
         *  }
         */
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String mes = new String(message.getBody(),"UTF-8");
            //消息被拒绝，basicReject第二个参数是false表示被拒绝后不放回队列
            if("info5".equals(mes)) {
                System.out.println("Consumer01接收的消息是：" + mes + "，此消息被拒绝");
                channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
            }else {
                System.out.println("Consumer01接收的消息是：" + mes);
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
            }
        };
        //开启手动应答
        channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,consumerTag -> {});
    }

}

(19)停掉所有程序，然后重新运行Consumer01，生成队列和交换机

(20)重新运行Producer

(21)查看Consumer01控制台的输出，观察dead_queue队列消息数量的变化，因为没有开启Consumer02消费dead_queue队列，可以看到dead_queue队列堆积了1条消息，查看这条消息，可以看出就是我们拒绝掉的info5，这证明消费者拒绝消费消息info5后，消息info5从normal_queue队列移到了dead_queue队列里，由此可见当消息被拒绝消费后，死信队列成功运行​

(22)启动Consumer02，可以看到Consumer02消费了dead_queue队列里的消息

标签： rabbitmq java-rabbitmq
本文转载自: https://blog.csdn.net/m0_64284147/article/details/130028539
版权归原作者 Ken_1115 所有， 如有侵权，请联系我们删除。
登录后发布评论

“RabbitMQ系列（15）--死信队列的简介与死信队列和死信消费者的实现”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航