SpringBoot教程（十五） | SpringBoot集成RabbitMq（死信队列、延迟队列）

2024-09-06 15:20
文章标签 java springboot 教程 spring 队列 集成 rabbitmq 十五 延迟 死信

本文主要是介绍SpringBoot教程（十五） | SpringBoot集成RabbitMq（死信队列、延迟队列），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

SpringBoot教程（十五） | SpringBoot集成RabbitMq（死信队列、延迟队列）

  • （一）死信队列
    • 使用场景
    • 具体用法
    • 前提
    • 示例:
  • （二）延迟队列
    • 使用场景
    • 方法一：通过死亡队列实现
    • 方法二：通过延迟消息插件（rabbitmq_delayed_message_exchange）实现

（一）死信队列

死信队列是一个重要的概念，用于处理那些因各种原因无法被正常消费的消息。
它不是RabbitMQ直接提供的一个现成的方法或工具，而是通过特定的配置和机制来实现的。

使用场景

死信队列在多种场景下都非常有用，包括但不限于：

  1. 消息重试机制：当消息处理失败时，可以将其发送到死信队列进行重试。
  2. 异常消息处理：对于无法被正常处理的异常消息，可以将其存储在死信队列中，以便后续分析处理。
  3. 延迟消息处理：通过结合消息的TTL（Time-To-Live，生存时间）和死信队列，可以实现消息的延迟处理。
  4. 确保消息不丢失：在消息处理过程中，如果发生消费者崩溃或网络故障等情况，消息可能会丢失。通过死信队列，可以确保这些消息得到保留，并在系统恢复后重新处理。

具体用法

要在RabbitMQ中设置和使用死信队列，通常需要按照以下步骤进行：

  1. 定义死信交换机（DLX）：首先，需要定义一个交换机作为死信交换机，它可以是任何类型的交换机（如direct、fanout、topic等）。
  2. 配置原队列：在声明原队列时，需要指定两个参数：x-dead-letter-exchange和x-dead-letter-routing-key。前者指定了当消息变成死信时应该发送到的交换机（即死信交换机），后者指定了发送到该交换机的路由键。
  3. 声明死信队列：接着，需要声明一个或多个死信队列，并将它们绑定到死信交换机上。这样，当死信消息被发送到死信交换机时，就可以根据路由键将其路由到相应的死信队列中。
  4. 处理死信消息：最后，需要编写消费者代码来监听死信队列中的消息，并对这些消息进行相应的处理。

前提

要想进入死信队列，得出现异常，出现异常后，会根据你的配置帮你放到死信队列中 所以异常不要被捕获。
如果实在要捕获的话，就得你在消费者这边去做“发送消息的”操作，自己把发送过来消息塞到死信队列中

示例:

消费者 mq的yml配置（重试机制）

spring:rabbitmq:host: 127.0.0.1port: 5672username: guestpassword: guestlistener:simple:# 重试机制retry:enabled: true #是否开启消费者重试

配置类：

package com.example.reactboot.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;@Configuration
public class DirectExchangeConfig {//===========================普通===========================//定义队列的名称常量public static final String DIRECT_QUEUE = "directQueue";public static final String DIRECT_QUEUE2 = "directQueue2";//定义直接交换机的名称常量public static final String DIRECT_EXCHANGE = "directExchange";//定义路由键常量，用于交换机和队列之间的绑定public static final String DIRECT_ROUTING_KEY = "direct";//定义路由键常量，用于交换机和队列之间的绑定public static final String DIRECT_ROUTING_KEY_2= "direct2";//定义队列，名称为DIRECT_QUEUE//为普通队列 绑设置 死信参数@Beanpublic Queue directQueue() {//return new Queue(DIRECT_QUEUE, true);Map<String, Object> args = new HashMap<>();// 设置死信交换机args.put("x-dead-letter-exchange", DLX_EXCHANGE);// 设置发送到死信交换机的路由键args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);// 创建队列，设置为持久化、非排他、非自动删除，并附带死信参数return new Queue(DIRECT_QUEUE, true, false, false, args);}//定义直接交换机@Beanpublic DirectExchange directExchange() {return new DirectExchange(DIRECT_EXCHANGE, true, false);}//定义队列，名称为DIRECT_QUEUE2@Beanpublic Queue directQueue2() {return new Queue(DIRECT_QUEUE2, true);}//定义一个绑定，将directQueue队列绑定到directExchange交换机上，//使用direct作为路由键@Beanpublic Binding bindingDirectExchange(Queue directQueue, DirectExchange directExchange) {return BindingBuilder.bind(directQueue).to(directExchange).with(DIRECT_ROUTING_KEY);}// 定义一个绑定Bean，将directQueue2队列也绑定到directExchange交换机上，@Beanpublic Binding bindingDirectExchange2(Queue directQueue2, DirectExchange directExchange) {return BindingBuilder.bind(directQueue2).to(directExchange).with(DIRECT_ROUTING_KEY_2);}//===========================死信===========================// 定义死信交换机的名称public static final String DLX_EXCHANGE = "dlx_exchange";// 定义发送到死信交换机的路由键public static final String DLX_ROUTING_KEY = "dlx.routing.key";// 定义死信队列的名称public static final String DLX_QUEUE = "dlx_queue";/*** 声明死信交换机，这里使用Direct类型。* @return 返回一个配置好的DirectExchange对象。*/@BeanDirectExchange dlxExchange() {// 创建并返回Direct类型的交换机return new DirectExchange(DLX_EXCHANGE,true, false);}/*** 声明死信队列。* @return 返回一个配置好的Queue对象，用作死信队列。*/@BeanQueue dlxQueue() {// 创建并返回死信队列，设置为持久化return new Queue(DLX_QUEUE, true);}/*** 绑定死信队列到死信交换机，使用指定的路由键。*/@BeanBinding binding(Queue dlxQueue,DirectExchange dlxExchange) {return BindingBuilder.bind(dlxQueue).to(dlxExchange).with(DLX_ROUTING_KEY);}}

生产者发送消息：

package com.example.reactboot.controller;import com.example.reactboot.config.DirectExchangeConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.nio.charset.StandardCharsets;@RestController
public class RabbitMqTest {@AutowiredRabbitTemplate rabbitTemplate;@RequestMapping("/sendMQ")public String sendMessage() {rabbitTemplate.convertAndSend(DirectExchangeConfig.DIRECT_EXCHANGE,DirectExchangeConfig.DIRECT_ROUTING_KEY, "发送一条测试消息：direct");return "direct消息发送成功！！";}}

消费者消费消息：

package com.example.reactboot.queueListener;import com.example.reactboot.config.DirectExchangeConfig;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;/*** @className: DirectQueueListener* @description: 直连交换机的监听器* @author: sh.Liu* @date: 2021-08-23 16:03*/
@Slf4j
@Component
public class DirectQueueListener {//监听普通队列@RabbitHandler@RabbitListener(queues = DirectExchangeConfig.DIRECT_QUEUE)public void process(String xx){SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("DirectReceiver消费者收到消息1  : " + xx + " 接收时间：" + sdf.format(new Date()) + "\n");//先执行业务代码int i = 1 / 0;}//监听死信队列@RabbitHandler@RabbitListener(queues = DirectExchangeConfig.DLX_QUEUE)public void process3(String testMessage) {System.out.println("死信得列里面的  : " + testMessage + "\n");}}

（二）延迟队列

延迟队列是一种特殊的消息队列，其内部消息是有序的，并且具有延时属性。
在RabbitMQ中，虽然AMQP协议本身没有直接支持延迟队列，但可以通过一些变通的方法（如使用死信队列配合消息的TTL属性，或者使用RabbitMQ的延迟消息插件）来实现延迟队列的功能。

使用场景

延迟队列在多种业务场景中都有广泛的应用，包括但不限于：

  1. 订单超时未支付自动取消：用户下单后，如果在规定时间内未完成支付，系统可以自动取消订单。
  2. 退款超时通知：用户申请退款后，如果长时间未得到处理，系统可以自动通知相关运营人员介入。
  3. 新用户注册后的引导邮件：用户注册账号后，系统可以在一段时间后发送欢迎邮件或引导邮件。
  4. 会议提醒：在预定的会议开始前一段时间，系统自动发送提醒给参会人员。
  5. 任务调度：在指定时间后执行某项任务，如定时清理日志、执行批处理任务等。

方法一：通过死亡队列实现

以下是使用死信队列配合TTL属性实现延迟队列的基本步骤：

  1. 定义死信交换机（DLX, Dead-Letter Exchange）和死信队列（DLQ, Dead-Letter Queue）
  2. 设置普通队列的TTL和死信交换机：在创建普通队列时，可以为其设置TTL属性，指定消息在该队列中的最大存活时间。同时，需要将该队列的死信交换机设置为前面定义的DLX，以便消息在过期后能够被发送到DLQ。
  3. 生产者发送消息：生产者将消息发送到普通队列，并指定消息的TTL。消息在队列中等待，直到TTL过期。
  4. 消息过期并发送到死信队列：当消息的TTL过期后，RabbitMQ会自动将该消息发送到其配置的死信交换机，再由死信交换机根据路由键将其发送到DLQ。
  5. 消费者从死信队列消费消息：消费者监听DLQ，当有新消息到达时，进行消费处理。

就是：把普通队列的消息设置存活时间，目前有两者方式：
1.在队列上面设置消息的过期时间
2.直接在消息上面设置过期时间。

方式一（队列上面设置消息过期时间）：

上面的关于 死信示例 完全可以复用进行测试

在以下的方法里面多加一行 args.put(“x-message-ttl”, 10000);

    //定义队列，名称为DIRECT_QUEUE//为普通队列 绑设置 死信参数@Beanpublic Queue directQueue() {//return new Queue(DIRECT_QUEUE, true);Map<String, Object> args = new HashMap<>();// 设置消息TTL为10秒args.put("x-message-ttl", 10000);// 设置死信交换机args.put("x-dead-letter-exchange", DLX_EXCHANGE);// 设置发送到死信交换机的路由键args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);// 创建队列，设置为持久化、非排他、非自动删除，并附带死信参数return new Queue(DIRECT_QUEUE, true, false, false, args);}

你可以把 DirectQueueListener 里面的 process 方法注释掉（以免被消费掉）。
再执行生产者的 sendMessage 方法。
这个时候你就可以看到下面关于 监听死信队列 的方法 ，等10秒后就会打印你发的消息了

方式二（消息上面设置过期时间）：

上面的关于 死信示例 完全可以复用进行测试

改一下 这个 生产者发送消息：

package com.example.reactboot.controller;import com.example.reactboot.config.DirectExchangeConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.nio.charset.StandardCharsets;@RestController
public class RabbitMqTest {@AutowiredRabbitTemplate rabbitTemplate;@RequestMapping("/sendMQ")public String sendMessage() {SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");rabbitTemplate.convertAndSend(DirectExchangeConfig.DIRECT_EXCHANGE, DirectExchangeConfig.DIRECT_ROUTING_KEY, "发送一条测试消息：direct! "+sdf.format(new Date()), new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {//设置过期时间，超过5秒消息就会消失message.getMessageProperties().setExpiration("5000");//设置编码格式message.getMessageProperties().setContentEncoding("UTF-8");return message;}});     return "direct消息发送成功！！";}}

你可以把 DirectQueueListener 里面的 process 方法注释掉（以免被消费掉）。
再执行生产者的 sendMessage 方法。
这个时候你就可以看到下面关于 监听死信队列 的方法 ，等5秒后就会打印你发的消息了

到这里其实就结束了，剩下的就是监听到死信队列里面的消息后的业务操作了

方法二：通过延迟消息插件（rabbitmq_delayed_message_exchange）实现

后续在说

这篇关于SpringBoot教程（十五） | SpringBoot集成RabbitMq（死信队列、延迟队列）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1142377。

相关文章

Java实现检查多个时间段是否有重合

Java实现检查多个时间段是否有重合

《Java实现检查多个时间段是否有重合》这篇文章主要为大家详细介绍了如何使用Java实现检查多个时间段是否有重合,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录流程概述步骤详解China编程步骤1：定义时间段类步骤2：添加时间段步骤3：检查时间段是否有重合步骤4：输出结果示例代码结语作
阅读更多...
Java中String字符串使用避坑指南

Java中String字符串使用避坑指南

《Java中String字符串使用避坑指南》Java中的String字符串是我们日常编程中用得最多的类之一,看似简单的String使用,却隐藏着不少“坑”,如果不注意,可能会导致性能问题、意外的错误容... 目录8个避坑点如下：1. 字符串的不可变性：每次修改都创建新对象2. 使用 == 比较字符串，陷阱满
阅读更多...
Java判断多个时间段是否重合的方法小结

Java判断多个时间段是否重合的方法小结

《Java判断多个时间段是否重合的方法小结》这篇文章主要为大家详细介绍了Java中判断多个时间段是否重合的方法,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录判断多个时间段是否有间隔判断时间段集合是否与某时间段重合判断多个时间段是否有间隔实体类内容public class D
阅读更多...
IDEA编译报错“java: 常量字符串过长”的原因及解决方法

IDEA编译报错“java: 常量字符串过长”的原因及解决方法

《IDEA编译报错“java:常量字符串过长”的原因及解决方法》今天在开发过程中,由于尝试将一个文件的Base64字符串设置为常量,结果导致IDEA编译的时候出现了如下报错java:常量字符串过长,... 目录一、问题描述二、问题原因2.1 理论角度2.2 源码角度三、解决方案解决方案①：StringBui
阅读更多...
Java覆盖第三方jar包中的某一个类的实现方法

Java覆盖第三方jar包中的某一个类的实现方法

《Java覆盖第三方jar包中的某一个类的实现方法》在我们日常的开发中,经常需要使用第三方的jar包,有时候我们会发现第三方的jar包中的某一个类有问题,或者我们需要定制化修改其中的逻辑,那么应该如何... 目录一、需求描述二、示例描述三、操作步骤四、验证结果五、实现原理一、需求描述需求描述如下：需要在
阅读更多...
Debezium 与 Apache Kafka 的集成方式步骤详解

Debezium 与 Apache Kafka 的集成方式步骤详解

《Debezium与ApacheKafka的集成方式步骤详解》本文详细介绍了如何将Debezium与ApacheKafka集成,包括集成概述、步骤、注意事项等,通过KafkaConnect,D... 目录一、集成概述二、集成步骤1. 准备 Kafka 环境2. 配置 Kafka Connect3. 安装 D
阅读更多...
Java中ArrayList和LinkedList有什么区别举例详解

Java中ArrayList和LinkedList有什么区别举例详解

《Java中ArrayList和LinkedList有什么区别举例详解》：本文主要介绍Java中ArrayList和LinkedList区别的相关资料,包括数据结构特性、核心操作性能、内存与GC影... 目录一、底层数据结构二、核心操作性能对比三、内存与 GC 影响四、扩容机制五、线程安全与并发方案六、工程
阅读更多...
JavaScript中的reduce方法执行过程、使用场景及进阶用法

JavaScript中的reduce方法执行过程、使用场景及进阶用法

《JavaScript中的reduce方法执行过程、使用场景及进阶用法》：本文主要介绍JavaScript中的reduce方法执行过程、使用场景及进阶用法的相关资料,reduce是JavaScri... 目录1. 什么是reduce2. reduce语法2.1 语法2.2 参数说明3. reduce执行过程
阅读更多...
如何使用Java实现请求deepseek

如何使用Java实现请求deepseek

《如何使用Java实现请求deepseek》这篇文章主要为大家详细介绍了如何使用Java实现请求deepseek功能,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录1.deepseek的api创建2.Java实现请求deepseek2.1 pom文件2.2 json转化文件2.2
阅读更多...
Java调用DeepSeek API的最佳实践及详细代码示例

Java调用DeepSeek API的最佳实践及详细代码示例

《Java调用DeepSeekAPI的最佳实践及详细代码示例》：本文主要介绍如何使用Java调用DeepSeekAPI,包括获取API密钥、添加HTTP客户端依赖、创建HTTP请求、处理响应、... 目录1. 获取API密钥2. 添加HTTP客户端依赖3. 创建HTTP请求4. 处理响应5. 错误处理6.
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2