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死信队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Javaweb项目启动Tomcat常见的报错解决方案
Java Web项目启动Tomcat时可能会遇到各种错误,本文就来介绍一下Javaweb项目启动Tomcat常见的报错解决方案,具有一定的参考价值,感兴趣的可以了解一下
2024-02-02
springboot使用JdbcTemplate完成对数据库的增删改查功能
这篇文章主要介绍了springboot使用JdbcTemplate完成对数据库的增删改查功能,需要的朋友可以参考下
2017-12-12
Protostuff序列化和反序列化的使用说明
今天小编就为大家分享一篇关于Protostuff序列化和反序列化的使用说明，小编觉得内容挺不错的，现在分享给大家，具有很好的参考价值，需要的朋友一起跟随小编来看看吧
2019-04-04
java多文件压缩下载的解决方法
这篇文章主要为大家详细介绍了java多文件压缩下载的解决方法，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2019-07-07
java交换排序之奇偶排序实现方法
这篇文章主要介绍了java交换排序之奇偶排序实现方法,实例分析了奇偶排序的原理与具体实现技巧,非常具有实用价值,需要的朋友可以参考下
2015-02-02
Java并发工具辅助类代码实例
这篇文章主要介绍了Java并发工具辅助类代码实例,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2020-04-04
Java调用python代码的五种方式总结
这篇文章主要给大家介绍了关于Java调用python代码的五种方式,在Java中调用Python函数的方法有很多种,文中通过代码介绍的非常详细,需要的朋友可以参考下
2023-09-09
MyBatisPlus批量添加的优化与报错解决
MybatisPlus是一个高效的java持久层框架,它在Mybatis的基础上增加了一些便捷的功能,提供了更加易用的API,可以大幅度提高开发效率,这篇文章主要给大家介绍了关于MyBatisPlus批量添加的优化与报错解决的相关资料,需要的朋友可以参考下
2023-05-05
java中实体类和JSON对象之间相互转化
Java中关于Json格式转化Object,Map,Collection类型和String类型之间的转化在我们实际项目中应用的很是普遍和广泛。最近工作的过程中也是经常有，因此,自己封装了一个类分享给大家。
2015-05-05
Kafka在Spring Boot生态中的浅析与应用场景分析
文章主要介绍了Apache Kafka的核心概念、主要业务场景以及在Spring Boot项目中的集成方式,文章还详细介绍了Kafka的事务支持,感兴趣的朋友跟随小编一起看看吧
2025-10-10