当前位置： 首页 > news >正文

Spring Boot 集成 RabbitMQ：普通队列、延迟队列与死信队列全解析

news 2025/7/22 7:51:47

Spring Boot 集成 RabbitMQ：普通队列、延迟队列与死信队列全解析

  • 1. 背景介绍
  • 2. RabbitMQ 及队列类型详解
  • 3. 项目依赖配置（pom.xml）
  • 4. Spring Boot RabbitMQ 配置详解（application.yml）
  • 5. 核心队列代码示例及详解
  • 6. 消息生产者实现
  • 7. 消费者设计及异常处理策略
  • 8. 死信队列消费者与告警设计
  • 9. 消息确认机制详解
    • 常见异常示例
    • 异常原因分析
    • 解决方案
  • 10. 延迟队列实现原理与RabbitMQ插件说明
  • 11. 系统容错与性能优化建议
  • 12. 常见问题及排查方法
  • 13. 总结与最佳实践

1. 背景介绍

现代分布式系统中，异步消息队列作为解耦、削峰和异步处理的重要组件，被广泛采用。RabbitMQ 是一款基于 AMQP 协议的成熟消息队列中间件，功能丰富，性能稳定。

在这里插入图片描述

本篇文章通过一个典型的业务场景讲解如何在 Spring Boot 应用中集成 RabbitMQ，实现：

  • 普通队列：用于正常业务消息处理

  • 延迟队列：实现消息的延迟投递和重试机制

  • 死信队列：捕获处理失败的消息，方便后续监控、报警或补偿处理

2. RabbitMQ 及队列类型详解

队列类型作用使用场景
普通队列存放业务正常消息，消费者消费并处理订单处理、用户通知、日志收集等实时消息处理
延迟队列支持消息延迟一定时间后再消费，用于重试或定时任务消息失败自动重试，定时提醒，延时任务执行
死信队列存放消费失败或过期的消息，避免消息丢失消息处理异常、消息TTL到期、消息被拒绝等情况捕获

关键概念

  • 交换机 (Exchange)：接收生产者发送的消息，根据类型和路由规则转发到相应队列

  • 队列 (Queue)：消息的存储容器，消费者从队列获取消息进行消费

  • 绑定 (Binding)：交换机和队列的关联关系，确定消息流向

  • 路由键 (Routing Key)：用于交换机匹配队列的关键字

  • 死信 (Dead Letter)：消息在队列中无法正常消费，被丢弃或重新路由的消息

3. 项目依赖配置（pom.xml）

   <!-- Spring Boot AMQP Starter，集成 RabbitMQ --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

4. Spring Boot RabbitMQ 配置详解（application.yml）

spring:rabbitmq:host: localhost          # RabbitMQ服务器地址port: 5672               # 端口号username: guest          # 用户名password: guest          # 密码virtual-host: /          # 虚拟主机publisher-confirm-type: correlated  # 消息确认机制publisher-returns: true  # 开启消息返回# 自定义队列和交换机配置参数config:normal:queue: normal.task.queueexchange: normal.task.exchangerouting-key: normal.task.routingKeydelay:queue: delay.retry.queueexchange: delay.retry.exchangerouting-key: delay.retry.routingKeydead-letter:queue: dead.letter.queueexchange: dead.letter.exchangerouting-key: dead.letter.routingKeyttl: 604800000 # 死信队列消息存活时间，7天（单位：毫秒）

5. 核心队列代码示例及详解

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;@Configuration
public class RabbitMQConfig {/*** 普通队列* 配置死信交换机和死信路由键，实现消息失败后自动进入死信队列*/@Beanpublic Queue normalQueue() {return QueueBuilder.durable("normal.task.queue").withArgument("x-dead-letter-exchange", "dead.letter.exchange").withArgument("x-dead-letter-routing-key", "dead.letter.routingKey").build();}/*** 普通直连交换机*/@Beanpublic DirectExchange normalExchange() {return new DirectExchange("normal.task.exchange");}/*** 普通队列绑定到普通交换机，路由键 normal.task.routingKey*/@Beanpublic Binding normalBinding() {return BindingBuilder.bind(normalQueue()).to(normalExchange()).with("normal.task.routingKey");}/*** 延迟交换机，基于 rabbitmq_delayed_message_exchange 插件* 通过 x-delayed-message 类型支持延迟投递消息*/@Beanpublic CustomExchange delayExchange() {Map<String, Object> args = new HashMap<>();args.put("x-delayed-type", "direct");  // 指定交换机类型为 directreturn new CustomExchange("delay.retry.exchange", "x-delayed-message", true, false, args);}/*** 延迟队列，存放延迟消息*/@Beanpublic Queue delayQueue() {return QueueBuilder.durable("delay.retry.queue").build();}/*** 延迟队列绑定延迟交换机，路由键 delay.retry.routingKey*/@Beanpublic Binding delayBinding() {return BindingBuilder.bind(delayQueue()).to(delayExchange()).with("delay.retry.routingKey").noargs();}/*** 死信队列，用于存放失败或过期消息* 配置消息 TTL，过期消息自动删除*/@Beanpublic Queue deadLetterQueue() {return QueueBuilder.durable("dead.letter.queue").withArgument("x-message-ttl", 604800000)  // 7天消息过期时间.build();}/*** 死信交换机*/@Beanpublic DirectExchange deadLetterExchange() {return new DirectExchange("dead.letter.exchange");}/*** 死信队列绑定死信交换机，路由键 dead.letter.routingKey*/@Beanpublic Binding deadLetterBinding() {return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with("dead.letter.routingKey");}
}

6. 消息生产者实现

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;@Component
@Slf4j
public class TaskProducer {@Resourceprivate RabbitTemplate rabbitTemplate;/*** 发送普通任务消息* @param message 业务消息内容*/public void sendTask(String message) {rabbitTemplate.convertAndSend("normal.task.exchange", "normal.task.routingKey", message);log.info("[生产者] 发送普通消息: {}", message);}/*** 发送延迟任务消息，默认延迟30分钟* @param message 消息内容*/public void sendDelayedTask(String message) {sendDelayedTask(message, 30 * 60 * 1000L);}/*** 发送指定延迟时间的延迟任务消息* @param message 消息内容* @param delayMillis 延迟时间，单位毫秒*/public void sendDelayedTask(String message, long delayMillis) {MessageProperties props = new MessageProperties();props.setContentType(MessageProperties.CONTENT_TYPE_JSON);// 设置延迟时间（单位毫秒）props.setHeader("x-delay", delayMillis);Message amqpMessage = new Message(message.getBytes(StandardCharsets.UTF_8), props);rabbitTemplate.send("delay.retry.exchange", "delay.retry.routingKey", amqpMessage);log.info("[生产者] 发送延迟消息，延迟 {} 秒后投递: {}", delayMillis / 1000, message);}
}

7. 消费者设计及异常处理策略

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.io.IOException;@Component
@Slf4j
public class TaskConsumer {@Resourceprivate TaskProducer taskProducer;/*** 监听普通队列* 消费失败时将消息发送到延迟队列进行重试*/@RabbitListener(queues = "normal.task.queue", ackMode = "MANUAL")public void handleNormalQueue(String message, Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {try {log.info("[普通队列] 处理消息: {}", message);// TODO: 业务逻辑处理// 消息成功处理，确认消息channel.basicAck(tag, false);} catch (Exception e) {log.error("[普通队列] 消息处理失败，发送延迟队列重试: {}", message, e);try {// 发送延迟队列，延迟30分钟重试taskProducer.sendDelayedTask(message);// 确认消息，防止消息重复消费channel.basicAck(tag, false);} catch (Exception ex) {log.error("[普通队列] 延迟队列发送失败，消息进入死信队列: {}", message, ex);// 拒绝消息，消息进入死信队列channel.basicReject(tag, false);}}}/*** 监听延迟队列* 消费失败时消息进入死信队列，避免死循环*/@RabbitListener(queues = "delay.retry.queue", ackMode = "MANUAL")public void handleDelayedQueue(String message, Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {try {log.info("[延迟队列] 处理消息: {}", message);// TODO: 业务逻辑处理// 确认消息channel.basicAck(tag, false);} catch (Exception e) {log.error("[延迟队列] 处理失败，消息进入死信队列: {}", message, e);// 拒绝消息，进入死信队列channel.basicReject(tag, false);}}
}

8. 死信队列消费者与告警设计

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;import java.io.IOException;@Component
@Slf4j
public class DeadLetterConsumer {/*** 监听死信队列，处理无法消费或过期消息* 业务可实现报警、日志存储或人工干预*/@RabbitListener(queues = "dead.letter.queue", ackMode = "MANUAL")public void handleDeadLetter(String message, Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {try {log.warn("[死信队列] 接收到死信消息: {}", message);// TODO: 告警处理、持久化存储等// 确认消息，避免重复消费channel.basicAck(tag, false);} catch (Exception e) {log.error("[死信队列] 处理死信消息异常，丢弃消息: {}", message, e);channel.basicReject(tag, false);}}
}

9. 消息确认机制详解

  • 自动确认（AUTO）：消息一旦被投递给消费者，RabbitMQ 直接认为消息已被消费成功，存在消息丢失风险。

  • 手动确认（MANUAL）：消费者在消息成功处理后，显式调用 basicAck 确认，失败时调用 basicReject 或 basicNack，保证消息不丢失。

  • 本文示例采用手动确认，结合死信机制，实现更高可靠性。

在上述示例中，我们采用了 手动确认 模式以保证消息的可靠消费。手动确认使得消费者在处理完成后，主动通知 RabbitMQ 消息已被正确消费，避免消息丢失或重复消费。

常见异常示例

当调用 channel.basicAck() 或 channel.basicReject() 传入了错误的 delivery tag 时，可能出现如下异常：


Shutdown Signal: channel error; protocol method: #method\<channel.close>(
reply-code=406, reply-text=PRECONDITION\_FAILED - unknown delivery tag 1, class-id=60, method-id=80)

异常原因分析

  • RabbitMQ 的 delivery tag 是每个 Channel 上递增的消息编号，用于标识该消息。
  • 如果调用确认时传入了不正确或已确认过的 delivery tag，就会出现“unknown delivery tag”的异常。
  • 多线程或异步场景中，共享 Channel 可能导致 delivery tag 不匹配。

解决方案

  1. 配置 Spring Boot 启用手动确认

    在 application.yml 中明确配置：

    spring:rabbitmq:listener:type: directdirect:acknowledge-mode: manual

    该配置保证了监听容器采用手动确认模式。

  2. 在 @RabbitListener 注解中显式声明手动确认 ackMode = "MANUAL"

    @RabbitListener(queues = "your_queue_name", ackMode = "MANUAL")
public void consumeMessage(String msg, Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {try {// 业务处理channel.basicAck(deliveryTag, false);} catch (Exception e) {channel.basicReject(deliveryTag, false);}
}

  3. 确保每条消息只调用一次确认方法

    • 严格保证 basicAck 或 basicReject 只针对当前消息调用一次，避免重复确认。

    • 尽量避免多个线程共用同一 Channel。

10. 延迟队列实现原理与RabbitMQ插件说明

  • RabbitMQ 官方不支持原生延迟队列，但提供了rabbitmq_delayed_message_exchange插件。
  • 延迟消息通过消息头 x-delay 设置延迟毫秒数，消息在交换机中等待指定时间后投递到绑定的队列。
  • 安装插件命令（RabbitMQ 服务端执行）：
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
  • 重新启动 RabbitMQ 服务后生效。

11. 系统容错与性能优化建议

  • 消息幂等性：确保消费者处理幂等，避免因消息重试造成数据重复。

  • 消息过期：合理配置死信队列消息TTL，避免死信队列堆积。

  • 重试机制：延迟队列可灵活配置延迟时间，多次重试后进入死信队列。

  • 连接池配置：合理设置 RabbitMQ 连接池大小，避免资源浪费。

  • 监控告警：对死信队列消息量、积压情况、消费者消费速率做监控，及时发现异常。

  • 异常日志：完整记录异常日志，方便问题排查。

  • 消息大小限制：避免发送过大消息，影响吞吐性能。

12. 常见问题及排查方法

问题可能原因排查建议
消费者未收到消息交换机和队列绑定不正确，路由键错误检查绑定关系和路由键配置
延迟消息无效未启用 rabbitmq_delayed_message_exchange 插件服务器启用插件，并重启 RabbitMQ
消息未进入死信队列死信交换机或死信路由键配置错误确认死信队列配置参数和绑定是否正确
消费者抛异常导致消息重试业务代码异常未捕获优化业务逻辑，捕获异常，避免无限重试
队列积压严重消费者消费慢或宕机增加消费者实例，检查消费者性能，避免阻塞

13. 总结与最佳实践

  • 利用 普通队列 + 延迟队列 + 死信队列，构建灵活的消息处理流程，提升系统可靠性和可维护性。

  • 结合手动消息确认和异常捕获，实现消息不丢失、失败消息自动重试。

  • 延迟队列利用 RabbitMQ 插件实现定时任务和重试逻辑，避免复杂业务逻辑实现。

  • 死信队列作为异常消息的存放点，配合告警和人工介入，保障业务稳定。

  • 结合监控体系和合理配置，实现高可用、可扩展的消息系统。

http://www.dtcms.com/a/290692.html

相关文章：

  • iOS WebView 调试实战 页面跳转失效与历史记录错乱的排查路径
  • 物流链上的智慧觉醒：Deepoc具身智能如何重塑搬运机器人的“空间思维”
  • 达梦数据库JSON_TABLE使用说明
  • grpc: debug: GRPC_TRACE
  • ESP32开发——基于idf框架使用NVS操作存储设备读写
  • 家庭服务具身智能机器人体系架构
  • 一次 POI 版本升级踩坑记录
  • lesson20：Python函数的标注
  • docker nginx 部署前端踩坑记录
  • WinUI3开发_Frame用法
  • MYSQL：数据库约束
  • 【PTA数据结构 | C语言版】拓扑排序
  • 通信刚需小能手，devicenet转PROFINET网关兼容物流分拣自动化
  • 自动化计算机经过加固后有什么好处？
  • OpenAI API(2) OpenAI Responses API使用
  • 设备管理系统(MMS)如何在工厂MOM功能设计和系统落地
  • 深入解析 Linux 硬链接与软链接：原理、区别及应用场景
  • 龙虎榜——20250721
  • Linux中ELF区域与文件偏移量的关系
  • 【AI论文】EXAONE 4.0：融合非推理模式与推理模式的统一大语言模型
  • Neovim 安装与解压 tar.gz 文件
  • AXI接口学习
  • Python 模块未找到？这样解决“ModuleNotFoundError”
  • Dev C++下载安装和使用教程（图文并茂，保姆级教程）
  • dolphinscheduler中sqoop无法执行
  • 机器人工程专业本科阶段的学习分析(腾讯元宝)
  • Real-World Blur Dataset for Learning and Benchmarking Deblurring Algorithms
  • 系统分析师-计算机系统-操作系统-存储器管理设备管理
  • Oracle From查看弹性域设置
  • （3）Oracle基本语法与常用函数