springboot中RabbitMQ死信队列的实现示例

更新时间：2024年01月23日 10:06:30 作者：Java小生不才

死信队列是一种特殊的消息队列,用来存储无法被正常消费的消息,常被用来实现延迟处理,异常消息处理等,本文主要介绍了springboot中RabbitMQ死信队列的实现示例,感兴趣的可以了解一下

1. 死信队列

死信队列是一种特殊的消息队列,用来存储无法被正常消费的消息，常被用来实现延迟处理，异常消息处理等，提高了系统的可伸缩性和容错性，能够应对高并发和异常消息。
死信队列中的消息被称为死信消息,用来分发死信消息的交换机被称为死信交换机(Dead Letter Exchange,DLX)。
死信队列在实际项目中的应用场景有很多如:

订单超时未支付，将此消息放入死信队列中，等待后续处理(延迟等待)
消息消费失败将消息放入死信队列中进行重试(消息重试机制)

在这里插入图片描述

2.正常消息成为死信消息的条件

消息到了过期时间仍然未被消费者消费
队列已满无法保存新消息
消息被拒绝消费且未设置重新放入队列

在这里插入图片描述

3.消费者1

package com.hong.rabbitmq9;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * @Description: 死信队列消费者1
 * @Author: hong
 * @Date: 2024-01-17 21:04
 * @Version: 1.0
 **/
public class Consumer1 {
    //正常交换机名称
    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 = RabbitMQUtil.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");

        //正常队列绑定死信队列信息
        Map<String,Object> map = new HashMap<>();
        map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        map.put("x-dead-letter-routing-key", "lisi");
        //声明正常队列
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,map);
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");

        System.out.println("Consumer1等待接收消息:");
        DeliverCallback deliverCallback = (comsumerTag, message) -> {
            System.out.println( "routingKey:" + message.getEnvelope().getRoutingKey() + ",消息:" + new String(message.getBody(), StandardCharsets.UTF_8));
        };

        CancelCallback cancelCallback = var -> {
        };

        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback);

    }
}

在这里插入图片描述

4.生产者

package com.hong.rabbitmq9;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

/**
 * @Description: 死信队列消息生产者
 * @Author: hong
 * @Date: 2024-01-17 20:49
 * @Version: 1.0
 **/
public class Producer {
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.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);
        }
    }

}

启动消费者1后马上关闭消费者1,模拟消费者1接收不到消息,再启动生产者

在这里插入图片描述

生产者发送10条正常队列中有10条消息

在这里插入图片描述

10s后正常队列中的消息由于没有消费者消费进入死信队列中

在这里插入图片描述

5.消费者2

package com.hong.rabbitmq9;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * @Description: 死信队列-死信消费者
 * @Author: hong
 * @Date: 2024-01-17 21:31
 * @Version: 1.0
 **/
public class Consumer2 {
    public static final String DEAD_EXCHANGE = "dead_exchange";
    public static final String DEAD_QUEUE = "dead_queue";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();

        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");


        System.out.println("Consumer2等待接收死信消息:");
        DeliverCallback deliverCallback = (comsumerTag, message) -> {
            System.out.println( "routingKey:" + message.getEnvelope().getRoutingKey() + ",消息:" + new String(message.getBody(), StandardCharsets.UTF_8));
        };

        CancelCallback cancelCallback = var -> {
        };

        channel.basicConsume(DEAD_QUEUE,true,deliverCallback,cancelCallback);
    }
}

启动消费者2,使其消费死信队列中的消息

在这里插入图片描述

6.队列达到最大长度

以上代码是TTL,队列达到最大长度只要将上述代码稍微改动一下即可

6.1.注释掉生产者代码中的ttl部分

package com.hong.rabbitmq9;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

/**
 * @Description: 死信队列消息生产者
 * @Author: hong
 * @Date: 2024-01-17 20:49
 * @Version: 1.0
 **/
public class Producer {
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.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", null, message.getBytes());
            System.out.println("生产者发送消息:" + message);
        }
    }
}

6.2.消费者1代码中加最大长度

package com.hong.rabbitmq9;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * @Description: 死信队列消费者1
 * @Author: hong
 * @Date: 2024-01-17 21:04
 * @Version: 1.0
 **/
public class Consumer1 {
    //正常交换机名称
    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 = RabbitMQUtil.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");

        //正常队列绑定死信队列信息
        Map<String,Object> map = new HashMap<>();
        map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        map.put("x-dead-letter-routing-key", "lisi");
        map.put("x-max-length",8);
        //声明正常队列
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,map);
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");

        System.out.println("Consumer1等待接收消息:");
        DeliverCallback deliverCallback = (comsumerTag, message) -> {
            System.out.println( "routingKey:" + message.getEnvelope().getRoutingKey() + ",消息:" + new String(message.getBody(), StandardCharsets.UTF_8));
        };

        CancelCallback cancelCallback = var -> {
        };

        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback);

    }
}

启动消费者1后立马关闭，模拟队列已满

在这里插入图片描述

到此这篇关于springboot中RabbitMQ死信队列的实现示例的文章就介绍到这了,更多相关springboot RabbitMQ死信队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Java日志框架之logback使用详解
这篇文章主要介绍了Java日志框架之logback使用详解，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-07-07
SpringCloud Gateway过滤器StripPrefix=1的作用及说明
SpringCloudGateway过滤器StripPrefix=1用于去除请求的前缀路径,具体配置如下：当收到形如/api/users的请求时,StripPrefix=1将会去除/api前缀,将最终的请求路径转发到http://example.com/users,需要注意的是StripPrefix过滤器可以根据实际需要进行配置
2026-01-01
Spring Boot 集成 MyBatis 全面讲解(最新推荐)
MyBatis 是一款优秀的持久层框架,与 Spring Boot 集成后可以大大简化开发流程,本文将全面讲解如何在 Spring Boot 中集成 MyBatis,包括环境配置、基础操作、高级功能和最佳实践,需要的朋友可以参考下
2024-12-12
Java使用多线程批次查询大量数据(Callable返回数据)方式
今天给大家分享Java使用多线程批次查询大量数据(Callable返回数据)方式,多线程有好几种方式,今天说的方式比较好,实现Callable<> 这种方式能返回查询的数据,加上Future异步获取方式,查询效率大大加快,感兴趣的朋友一起看看吧
2023-11-11
mybatis的association传递参数问题示例
这篇文章主要介绍了mybatis的association传递参数问题,本文通过示例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2022-12-12
Java日期格式化实现过程
本文介绍了在Java中处理日期格式yyyyMMddHHmmss的方法,包括使用java.time包（Java8+）和SimpleDateFormat（Java7及以下）,详细说明了格式化和解析时间的示例代码,并强调了线程安全和异常处理的重要性,还讨论了常见错误及其解决方法,以及如何生成唯一标识符
2025-11-11
详解Springboot对多线程的支持
Spring是通过任务执行器(TaskExecutor)来实现多线程和并发编程，使用ThreadPoolTaskExecutor来创建一个基于线城池的TaskExecutor。这篇文章给大家介绍Springboot对多线程的支持，感兴趣的朋友一起看看吧
2018-07-07
Java线程池新手避坑指南
在高并发编程中,线程池是一个非常重要的组件,它不仅能够有效地管理和复用线程资源,还可以提升应用程序的性能和稳定性,本文将详细介绍Java线程池避坑指南,从Executors到自定义线程池,一文搞懂,需要的朋友可以参考下
2025-07-07
Java数据类型与变量详解
文章主要介绍了Java中的字面常量、数据类型、变量、类型转换和字符串类型,涵盖了基本数据类型和引用数据类型,以及不同类型之间的转换和运算规则,感兴趣的朋友跟随小编一起看看吧
2026-01-01
SpringBoot之ApplicationRunner解析(spring容器启动完成执行的类)
这篇文章主要介绍了SpringBoot之ApplicationRunner解析(spring容器启动完成执行的类)，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2023-05-05