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。

相关文章

Spring boot整合dubbo+zookeeper的详细过程

Spring boot整合dubbo+zookeeper的详细过程

《Springboot整合dubbo+zookeeper的详细过程》本文讲解SpringBoot整合Dubbo与Zookeeper实现API、Provider、Consumer模式,包含依赖配置、... 目录Spring boot整合dubbo+zookeeper1.创建父工程2.父工程引入依赖3.创建ap
阅读更多...
SpringBoot结合Docker进行容器化处理指南

SpringBoot结合Docker进行容器化处理指南

《SpringBoot结合Docker进行容器化处理指南》在当今快速发展的软件工程领域,SpringBoot和Docker已经成为现代Java开发者的必备工具,本文将深入讲解如何将一个SpringBo... 目录前言一、为什么选择 Spring Bootjavascript + docker1. 快速部署与
阅读更多...
Spring Boot spring-boot-maven-plugin 参数配置详解(最新推荐)

Spring Boot spring-boot-maven-plugin 参数配置详解(最新推荐)

《SpringBootspring-boot-maven-plugin参数配置详解(最新推荐)》文章介绍了SpringBootMaven插件的5个核心目标（repackage、run、start... 目录一 spring-boot-maven-plugin 插件的5个Goals二 应用场景1 重新打包应用
阅读更多...
SpringBoot+EasyExcel实现自定义复杂样式导入导出

SpringBoot+EasyExcel实现自定义复杂样式导入导出

《SpringBoot+EasyExcel实现自定义复杂样式导入导出》这篇文章主要为大家详细介绍了SpringBoot如何结果EasyExcel实现自定义复杂样式导入导出功能,文中的示例代码讲解详细,... 目录安装处理自定义导出复杂场景1、列不固定，动态列2、动态下拉3、自定义锁定行/列，添加密码4、合并
阅读更多...
Spring Boot集成Druid实现数据源管理与监控的详细步骤

Spring Boot集成Druid实现数据源管理与监控的详细步骤

《SpringBoot集成Druid实现数据源管理与监控的详细步骤》本文介绍如何在SpringBoot项目中集成Druid数据库连接池,包括环境搭建、Maven依赖配置、SpringBoot配置文件... 目录1. 引言1.1 环境准备1.2 Druid介绍2. 配置Druid连接池3. 查看Druid监控
阅读更多...
Java中读取YAML文件配置信息常见问题及解决方法

Java中读取YAML文件配置信息常见问题及解决方法

《Java中读取YAML文件配置信息常见问题及解决方法》：本文主要介绍Java中读取YAML文件配置信息常见问题及解决方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要... 目录1 使用Spring Boot的@ConfigurationProperties2. 使用@Valu
阅读更多...
创建Java keystore文件的完整指南及详细步骤

创建Java keystore文件的完整指南及详细步骤

《创建Javakeystore文件的完整指南及详细步骤》本文详解Java中keystore的创建与配置,涵盖私钥管理、自签名与CA证书生成、SSL/TLS应用,强调安全存储及验证机制,确保通信加密和... 目录1. 秘密键（私钥）的理解与管理私钥的定义与重要性私钥的管理策略私钥的生成与存储2. 证书的创建与
阅读更多...
浅析Spring如何控制Bean的加载顺序

浅析Spring如何控制Bean的加载顺序

《浅析Spring如何控制Bean的加载顺序》在大多数情况下,我们不需要手动控制Bean的加载顺序,因为Spring的IoC容器足够智能,但在某些特殊场景下,这种隐式的依赖关系可能不存在,下面我们就来... 目录核心原则：依赖驱动加载手动控制 Bean 加载顺序的方法方法 1：使用@DependsOn(最直
阅读更多...
SpringBoot中如何使用Assert进行断言校验

SpringBoot中如何使用Assert进行断言校验

《SpringBoot中如何使用Assert进行断言校验》Java提供了内置的assert机制,而Spring框架也提供了更强大的Assert工具类来帮助开发者进行参数校验和状态检查,下... 目录前言一、Java 原生assert简介1.1 使用方式1.2 示例代码1.3 优缺点分析二、Spring Fr
阅读更多...
java使用protobuf-maven-plugin的插件编译proto文件详解

java使用protobuf-maven-plugin的插件编译proto文件详解

《java使用protobuf-maven-plugin的插件编译proto文件详解》：本文主要介绍java使用protobuf-maven-plugin的插件编译proto文件,具有很好的参考价... 目录protobuf文件作为数据传输和存储的协议主要介绍在Java使用maven编译proto文件的插件
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2