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。

相关文章

JVM 的类初始化机制

JVM 的类初始化机制

前言 当你在 Java 程序中new对象时，有没有考虑过 JVM 是如何把静态的字节码（byte code）转化为运行时对象的呢，这个问题看似简单，但清楚的同学相信也不会太多，这篇文章首先介绍 JVM 类初始化的机制，然后给出几个易出错的实例来分析，帮助大家更好理解这个知识点。 JVM 将字节码转化为运行时对象分为三个阶段，分别是：loading 、Linking、initialization
阅读更多...
Spring Security 基于表达式的权限控制

Spring Security 基于表达式的权限控制

前言 spring security 3.0已经可以使用spring el表达式来控制授权，允许在表达式中使用复杂的布尔逻辑来控制访问的权限。 常见的表达式 Spring Security可用表达式对象的基类是SecurityExpressionRoot。 表达式描述hasRole([role])用户拥有制定的角色时返回true （Spring security默认会带有ROLE_前缀）,去
阅读更多...
浅析Spring Security认证过程

浅析Spring Security认证过程

类图 为了方便理解Spring Security认证流程，特意画了如下的类图，包含相关的核心认证类 概述 核心验证器 AuthenticationManager 该对象提供了认证方法的入口，接收一个Authentiaton对象作为参数; public interface AuthenticationManager {Authentication authenticate(Authenti
阅读更多...
Spring Security--Architecture Overview

Spring Security--Architecture Overview

1 核心组件 这一节主要介绍一些在Spring Security中常见且核心的Java类，它们之间的依赖，构建起了整个框架。想要理解整个架构，最起码得对这些类眼熟。 1.1 SecurityContextHolder SecurityContextHolder用于存储安全上下文（security context）的信息。当前操作的用户是谁，该用户是否已经被认证，他拥有哪些角色权限…这些都被保
阅读更多...
Spring Security基于数据库验证流程详解

Spring Security基于数据库验证流程详解

Spring Security 校验流程图 相关解释说明（认真看哦） AbstractAuthenticationProcessingFilter 抽象类 /*** 调用 #requiresAuthentication(HttpServletRequest, HttpServletResponse) 决定是否需要进行验证操作。* 如果需要验证，则会调用 #attemptAuthentica
阅读更多...
Spring Security 从入门到进阶系列教程

Spring Security 从入门到进阶系列教程

Spring Security 入门系列 《保护 Web 应用的安全》 《Spring-Security-入门（一）：登录与退出》 《Spring-Security-入门（二）：基于数据库验证》 《Spring-Security-入门（三）：密码加密》 《Spring-Security-入门（四）：自定义-Filter》 《Spring-Security-入门（五）：在 Sprin
阅读更多...
Java架构师知识体认识

Java架构师知识体认识

源码分析 常用设计模式 Proxy代理模式Factory工厂模式Singleton单例模式Delegate委派模式Strategy策略模式Prototype原型模式Template模板模式 Spring5 beans 接口实例化代理Bean操作 Context Ioc容器设计原理及高级特性Aop设计原理Factorybean与Beanfactory Transaction 声明式事物
阅读更多...
Java进阶13讲__第12讲_1/2

Java进阶13讲__第12讲_1/2

多线程、线程池 1.  线程概念 1.1  什么是线程 1.2  线程的好处 2.   创建线程的三种方式 注意事项 2.1  继承Thread类 2.1.1 认识  2.1.2  编码实现  package cn.hdc.oop10.Thread;import org.slf4j.Logger;import org.slf4j.LoggerFactory
阅读更多...
Makefile简明使用教程

Makefile简明使用教程

文章目录 规则makefile文件的基本语法：加在命令前的特殊符号：.PHONY伪目标: Makefilev1 直观写法v2 加上中间过程v3 伪目标v4 变量 make 选项-f-n-C Make 是一种流行的构建工具，常用于将源代码转换成可执行文件或者其他形式的输出文件(如库文件、文档等)。Make 可以自动化地执行编译、链接等一系列操作。 规则 makefile文件
阅读更多...
hdu1180（广搜+优先队列）

hdu1180（广搜+优先队列）

此题要求最少到达目标点T的最短时间，所以我选择了广度优先搜索，并且要用到优先队列。 另外此题注意点较多，比如说可以在某个点停留，我wa了好多两次，就是因为忽略了这一点，然后参考了大神的思想，然后经过反复修改才AC的 这是我的代码 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2