SpringBoot整合RabbitMQ消息中间件，实现延迟队列和死信队列

2024-06-09 16:36
文章标签 java 实现 springboot spring 整合 队列 rabbitmq 消息中间件 延迟 死信

本文主要是介绍SpringBoot整合RabbitMQ消息中间件，实现延迟队列和死信队列，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

在现代的分布式系统中，消息队列作为一种重要的中间件，广泛应用于系统解耦、流量削峰、异步处理等场景。而RabbitMQ作为其中一款流行的消息队列中间件，因其高性能和丰富的功能受到众多开发者的青睐。本文将详细介绍如何在SpringBoot项目中整合RabbitMQ，实现延迟队列和死信队列，以满足复杂业务需求。

一、RabbitMQ简介

RabbitMQ是基于AMQP（Advanced Message Queuing Protocol）的开源消息代理系统，主要由以下几个部分组成：

  1. Producer（生产者）：消息的发送者。
  2. Consumer（消费者）：消息的接收者。
  3. Queue（队列）：存储消息的容器。
  4. Exchange（交换机）：接收生产者发送的消息，并根据绑定规则（Binding）将消息路由到队列。
  5. Binding（绑定）：将交换机与队列绑定的规则。

RabbitMQ支持多种交换机类型，如Direct、Fanout、Topic、Headers等，灵活性极高。

二、SpringBoot整合RabbitMQ

2.1 引入依赖

在SpringBoot项目中，我们可以通过引入Spring AMQP（Spring与RabbitMQ的集成框架）来快速整合RabbitMQ。在pom.xml中添加如下依赖：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2.2 配置RabbitMQ

在application.yml或application.properties中配置RabbitMQ连接信息：

spring:rabbitmq:host: localhostport: 5672username: guestpassword: guest

2.3 定义配置类

创建RabbitMQ的配置类，定义交换机、队列和绑定关系。

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {// 定义交换机@Beanpublic DirectExchange directExchange() {return new DirectExchange("directExchange");}// 定义队列@Beanpublic Queue queue() {return new Queue("queue");}// 定义绑定关系@Beanpublic Binding binding(Queue queue, DirectExchange directExchange) {return BindingBuilder.bind(queue).to(directExchange).with("routingKey");}
}

2.4 生产者

定义消息生产者，将消息发送到指定的交换机和路由键。

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;@Service
public class MessageProducer {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendMessage(String message) {rabbitTemplate.convertAndSend("directExchange", "routingKey", message);}
}

2.5 消费者

定义消息消费者，从队列中接收并处理消息。

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Service
public class MessageConsumer {@RabbitListener(queues = "queue")public void receiveMessage(String message) {System.out.println("Received message: " + message);}
}

三、实现延迟队列

延迟队列的需求在很多场景下非常常见，例如订单超时处理、消息重试等。RabbitMQ本身并不直接支持延迟队列功能，但我们可以通过TTL（Time-To-Live）和DLX（Dead Letter Exchange）机制来实现。

3.1 配置延迟队列

首先，我们需要定义一个用于存储延迟消息的队列，并配置其TTL和死信交换机：

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class DelayedQueueConfig {// 定义死信交换机@Beanpublic DirectExchange deadLetterExchange() {return new DirectExchange("deadLetterExchange");}// 定义死信队列@Beanpublic Queue deadLetterQueue() {return new Queue("deadLetterQueue");}// 定义死信队列与死信交换机的绑定关系@Beanpublic Binding deadLetterBinding() {return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with("deadLetterRoutingKey");}// 定义延迟队列，并设置其TTL和死信交换机@Beanpublic Queue delayedQueue() {return QueueBuilder.durable("delayedQueue").withArgument("x-dead-letter-exchange", "deadLetterExchange").withArgument("x-dead-letter-routing-key", "deadLetterRoutingKey").withArgument("x-message-ttl", 60000) // 60秒TTL.build();}// 定义延迟队列的交换机@Beanpublic DirectExchange delayedExchange() {return new DirectExchange("delayedExchange");}// 定义延迟队列与延迟交换机的绑定关系@Beanpublic Binding delayedBinding() {return BindingBuilder.bind(delayedQueue()).to(delayedExchange()).with("delayedRoutingKey");}
}

3.2 发送延迟消息

在生产者中，发送消息到延迟队列：

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;@Service
public class DelayedMessageProducer {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendDelayedMessage(String message) {rabbitTemplate.convertAndSend("delayedExchange", "delayedRoutingKey", message);}
}

3.3 消费延迟消息

定义消费者，从死信队列中接收并处理延迟后的消息：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Service
public class DelayedMessageConsumer {@RabbitListener(queues = "deadLetterQueue")public void receiveDelayedMessage(String message) {System.out.println("Received delayed message: " + message);}
}

四、实现死信队列

死信队列用于处理无法正常消费的消息。通常情况下，消息在以下情况会进入死信队列：

  1. 消息被拒绝（basic.reject或basic.nack）并且requeue参数设置为false。
  2. 消息在队列中的TTL过期。
  3. 队列的最大长度限制被超出。

4.1 配置死信队列

我们在前面已经定义了死信队列和死信交换机，这里我们进一步探讨如何将普通队列配置为支持死信消息：

@Configuration
public class DeadLetterQueueConfig {// 定义普通队列，并配置其死信交换机和死信路由键@Beanpublic Queue normalQueue() {return QueueBuilder.durable("normalQueue").withArgument("x-dead-letter-exchange", "deadLetterExchange").withArgument("x-dead-letter-routing-key", "deadLetterRoutingKey").build();}// 定义普通交换机@Beanpublic DirectExchange normalExchange() {return new DirectExchange("normalExchange");}// 定义普通队列与普通交换机的绑定关系@Beanpublic Binding normalBinding() {return BindingBuilder.bind(normalQueue()).to(normalExchange()).with("normalRoutingKey");}
}

4.2 生产消息

在生产者中，将消息发送到普通队列：

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;@Service
public class NormalMessageProducer {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendNormalMessage(String message) {rabbitTemplate.convertAndSend("normalExchange", "normalRoutingKey", message);}
}

4.3 消费消息

定义消费者，从普通队列中接收消息，如果出现问题则将消息转移到死信队列：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Service
public class NormalMessageConsumer {@RabbitListener(queues = "normalQueue")public void receiveNormalMessage(String message) {try {// 模拟处理逻辑System.out.println("Processing message: " + message);// 模拟异常情况if ("error".equals(message)) {throw new RuntimeException("Processing error");}} catch (Exception e) {// 消息处理失败，拒绝并不重新入队System.out.println("Message processing failed: " + message);throw new AmqpRejectAndDontRequeueException("Message rejected");}}
}

4.4 消费死

信消息

定义死信消息消费者，从死信队列中接收并处理无法正常消费的消息：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Service
public class DeadLetterMessageConsumer {@RabbitListener(queues = "deadLetterQueue")public void receiveDeadLetterMessage(String message) {System.out.println("Received dead letter message: " + message);// 处理死信消息的逻辑}
}

五、总结

通过本文，我们详细介绍了如何在SpringBoot项目中整合RabbitMQ，并实现延迟队列和死信队列的功能。我们先介绍了RabbitMQ的基本概念，然后逐步讲解了如何配置RabbitMQ、定义生产者和消费者，最后重点介绍了延迟队列和死信队列的实现方式。希望本文能够帮助开发者更好地理解和应用RabbitMQ，实现更加健壮和灵活的消息处理系统。

在实际开发中，消息队列的配置和使用可能会因具体业务需求而有所不同，开发者应根据自身需求进行调整和优化。同时，RabbitMQ提供了丰富的功能，如消息优先级、消息确认、集群部署等，开发者可以深入学习和应用这些功能，以构建高性能和高可用的分布式系统。

这篇关于SpringBoot整合RabbitMQ消息中间件，实现延迟队列和死信队列的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

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
阅读更多...
hdu1043（八数码问题，广搜 + hash（实现状态压缩） ）

hdu1043（八数码问题，广搜 + hash（实现状态压缩） ）

利用康拓展开将一个排列映射成一个自然数，然后就变成了普通的广搜题。 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<queue>#include<map>#include<stdio.h>#include<stdlib.h>#include<ctype.h>#inclu
阅读更多...
hdu1180（广搜+优先队列）

hdu1180（广搜+优先队列）

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

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2