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面试必考的关键字的用法汇总
这篇文章主要为大家详细介绍了Java中的几种关键字相关知识，本文比较适合刚入坑Java的小白以及准备秋招的大佬阅读，需要的小伙伴快收藏起来吧
2023-06-06
Spring Cloud基于zuul实现网关过程解析
这篇文章主要介绍了Spring Cloud基于zuul实现网关过程解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2019-12-12
SpringBoot 使用 @Value 注解读取配置文件给静态变量赋值
这篇文章主要介绍了SpringBoot 使用 @Value 注解读取配置文件给静态变量赋值，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-11-11
如何对Mysql数据表查询出来的结果进行排序
这篇文章主要介绍了如何对Mysql数据表查询出来的结果进行排序问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2023-08-08
20个非常实用的Java程序代码片段
这篇文章主要为大家分享了20个非常实用的Java程序片段，对java开发项目有所帮助，感兴趣的小伙伴们可以参考一下
2016-02-02
REST架构及RESTful应用程序简介
这篇文章主要为大家介绍了REST架构及RESTful的应用程序简介，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2022-03-03
java 代理机制的实例详解
这篇文章主要介绍了java 代理机制的实例详解的相关资料,这里说明下如何实现代理机制，帮助大家理解掌握这部分内容，需要的朋友可以参考下
2017-08-08
Java实现文本编译器
这篇文章主要为大家详细介绍了Java实现文本编译器，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2022-04-04
详解Java String类常用方法有哪些
今天给大家带来的是关于Java String的相关知识,文章围绕着String类常用方法展开,文中有非常详细的介绍及代码示例,需要的朋友可以参考下
2021-06-06
快速解决List集合add元素,添加多个对象出现重复的问题
这篇文章主要介绍了快速解决List集合add元素,添加多个对象出现重复的问题，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2020-08-08