Spring Boot 3 集成 RabbitMQ 實踐指南(原理解析)
更新時間：2025年02月24日 10:48:13 作者：翱翔-藍天
本文介紹了SpringBoot 3集成RabbitMQ的實踐指南,涵蓋了RabbitMQ的核心原理、核心概念、高級特性、應用場景、環(huán)境搭建、核心配置類、消息生產(chǎn)者、消息消費者、接口控制器、監(jiān)控與運維、最佳實踐以及常見問題與解決方案等內容,感興趣的朋友一起看看吧
Spring Boot 3 集成 RabbitMQ 實踐指南
1. RabbitMQ 核心原理
1.1 什么是RabbitMQ
RabbitMQ是一個開源的消息代理和隊列服務器，使用Erlang語言開發(fā)，基于AMQP（Advanced Message Queuing Protocol）協(xié)議實現(xiàn)。它支持多種消息傳遞模式，具有高可用性、可擴展性和可靠性等特點。
1.2 核心概念
1.2.1 基礎組件
Producer（生產(chǎn)者）
- 消息的發(fā)送者
- 負責創(chuàng)建消息并發(fā)布到RabbitMQ中
Consumer（消費者）
- 消息的接收者
- 連接到RabbitMQ服務器并訂閱隊列
Exchange（交換機）
- 接收生產(chǎn)者發(fā)送的消息并根據(jù)路由規(guī)則轉發(fā)到隊列
- 類型：
- Direct Exchange：根據(jù)routing key精確匹配
- Topic Exchange：根據(jù)routing key模式匹配
- Fanout Exchange：廣播到所有綁定隊列
- Headers Exchange：根據(jù)消息屬性匹配
Queue（隊列）
- 消息存儲的地方
- 支持持久化、臨時、自動刪除等特性
Binding（綁定）
- 交換機和隊列之間的虛擬連接
- 定義消息路由規(guī)則
1.2.2 高級特性
消息持久化
- 交換機持久化：創(chuàng)建時設置durable=true
- 隊列持久化：創(chuàng)建時設置durable=true
- 消息持久化：設置delivery-mode=2
消息確認機制
- 生產(chǎn)者確認：Publisher Confirm和Return機制
- 消費者確認：自動確認、手動確認、批量確認
死信隊列（DLX）
- 消息被拒絕且不重新入隊
- 消息過期（TTL）
- 隊列達到最大長度
1.3 應用場景
異步處理
- 發(fā)送郵件、短信通知
- 日志處理、報表生成
- 文件處理、圖片處理
應用解耦
- 系統(tǒng)間通信
- 服務解耦
- 流程分離
流量控制
- 削峰填谷
- 請求緩沖
- 流量整形
定時任務
- 延遲隊列
- 定時處理
- 任務調度
2. 環(huán)境搭建
2.1 基礎環(huán)境
- Spring Boot: 3.x
- Java: 17+
- RabbitMQ: 3.12+
- Maven/Gradle
2.2 依賴配置
<dependencies>
<!-- Spring Boot Starter AMQP -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- Spring Boot Starter Web -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- Lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<!-- Jackson -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
</dependencies>2.3 基礎配置
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
virtual-host: /
# 消息確認配置
publisher-confirm-type: correlated # 開啟發(fā)布確認
publisher-returns: true # 開啟發(fā)布返回
template:
mandatory: true # 消息路由失敗返回
# 消費者配置
listener:
simple:
acknowledge-mode: manual # 手動確認
prefetch: 1 # 每次獲取消息數(shù)量
retry:
enabled: true # 開啟重試
initial-interval: 1000 # 重試間隔時間
max-attempts: 3 # 最大重試次數(shù)
multiplier: 1.0 # 重試時間乘數(shù)
# SSL配置（可選）
ssl:
enabled: false
key-store: classpath:keystore.p12
key-store-password: password
trust-store: classpath:truststore.p12
trust-store-password: password3. 核心配置類
3.1 RabbitMQ配置類
@Configuration
@EnableRabbit
public class RabbitMQConfig {
// 交換機名稱
public static final String BUSINESS_EXCHANGE = "business.exchange";
public static final String DEAD_LETTER_EXCHANGE = "dead.letter.exchange";
// 隊列名稱
public static final String BUSINESS_QUEUE = "business.queue";
public static final String DEAD_LETTER_QUEUE = "dead.letter.queue";
// 路由鍵
public static final String BUSINESS_KEY = "business.key";
public static final String DEAD_LETTER_KEY = "dead.letter.key";
// 業(yè)務交換機
@Bean
public DirectExchange businessExchange() {
return ExchangeBuilder.directExchange(BUSINESS_EXCHANGE)
.durable(true)
.build();
}
// 死信交換機
@Bean
public DirectExchange deadLetterExchange() {
return ExchangeBuilder.directExchange(DEAD_LETTER_EXCHANGE)
.durable(true)
.build();
}
// 業(yè)務隊列
@Bean
public Queue businessQueue() {
Map<String, Object> args = new HashMap<>(3);
// 消息過期時間
args.put("x-message-ttl", 60000);
// 隊列最大長度
args.put("x-max-length", 1000);
// 死信交換機
args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
args.put("x-dead-letter-routing-key", DEAD_LETTER_KEY);
return QueueBuilder.durable(BUSINESS_QUEUE)
.withArguments(args)
.build();
}
// 死信隊列
@Bean
public Queue deadLetterQueue() {
return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
}
// 業(yè)務綁定
@Bean
public Binding businessBinding() {
return BindingBuilder.bind(businessQueue())
.to(businessExchange())
.with(BUSINESS_KEY);
}
// 死信綁定
@Bean
public Binding deadLetterBinding() {
return BindingBuilder.bind(deadLetterQueue())
.to(deadLetterExchange())
.with(DEAD_LETTER_KEY);
}
// 消息轉換器
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
// RabbitTemplate配置
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
rabbitTemplate.setMessageConverter(messageConverter());
return rabbitTemplate;
}
}3.2 消息確認配置
@Configuration
@Slf4j
public class RabbitConfirmConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void init() {
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnsCallback(this);
}
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (ack) {
log.info("消息發(fā)送到交換機成功: correlationData={}", correlationData);
} else {
log.error("消息發(fā)送到交換機失敗: correlationData={}, cause={}", correlationData, cause);
// 處理失敗邏輯，如重試、告警等
}
}
@Override
public void returnedMessage(ReturnedMessage returned) {
log.error("消息路由到隊列失敗: exchange={}, routingKey={}, replyCode={}, replyText={}, message={}",
returned.getExchange(),
returned.getRoutingKey(),
returned.getReplyCode(),
returned.getReplyText(),
new String(returned.getMessage().getBody()));
// 處理失敗邏輯，如重試、告警等
}
}4. 消息生產(chǎn)者
4.1 消息發(fā)送服務
@Service
@Slf4j
public class MessageProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendMessage(Object message, String exchange, String routingKey) {
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
try {
rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
log.info("消息發(fā)送成功: message={}, exchange={}, routingKey={}, correlationData={}",
message, exchange, routingKey, correlationData);
} catch (Exception e) {
log.error("消息發(fā)送異常: message={}, exchange={}, routingKey={}, correlationData={}, error={}",
message, exchange, routingKey, correlationData, e.getMessage());
throw new RuntimeException("消息發(fā)送失敗", e);
}
}
public void sendDelayMessage(Object message, String exchange, String routingKey, long delayMillis) {
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
MessagePostProcessor messagePostProcessor = msg -> {
msg.getMessageProperties().setDelay((int) delayMillis);
return msg;
};
try {
rabbitTemplate.convertAndSend(exchange, routingKey, message, messagePostProcessor, correlationData);
log.info("延遲消息發(fā)送成功: message={}, exchange={}, routingKey={}, delay={}, correlationData={}",
message, exchange, routingKey, delayMillis, correlationData);
} catch (Exception e) {
log.error("延遲消息發(fā)送異常: message={}, exchange={}, routingKey={}, delay={}, correlationData={}, error={}",
message, exchange, routingKey, delayMillis, correlationData, e.getMessage());
throw new RuntimeException("延遲消息發(fā)送失敗", e);
}
}
}5. 消息消費者
5.1 消息處理服務
@Service
@Slf4j
public class MessageConsumer {
@RabbitListener(queues = RabbitMQConfig.BUSINESS_QUEUE)
public void handleMessage(Message message, Channel channel) throws IOException {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
// 獲取消息內容
String messageBody = new String(message.getBody());
log.info("收到消息: message={}, deliveryTag={}", messageBody, deliveryTag);
// 業(yè)務處理
processMessage(messageBody);
// 手動確認消息
channel.basicAck(deliveryTag, false);
log.info("消息處理成功: deliveryTag={}", deliveryTag);
} catch (Exception e) {
log.error("消息處理異常: deliveryTag={}, error={}", deliveryTag, e.getMessage());
// 判斷是否重新投遞
if (message.getMessageProperties().getRedelivered()) {
log.error("消息已重試，拒絕消息: deliveryTag={}", deliveryTag);
channel.basicReject(deliveryTag, false);
} else {
log.info("消息首次處理失敗，重新投遞: deliveryTag={}", deliveryTag);
channel.basicNack(deliveryTag, false, true);
}
}
}
private void processMessage(String message) {
// 實現(xiàn)具體的業(yè)務邏輯
log.info("處理消息: {}", message);
}
}5.2 死信消息處理
@Service
@Slf4j
public class DeadLetterConsumer {
@RabbitListener(queues = RabbitMQConfig.DEAD_LETTER_QUEUE)
public void handleDeadLetter(Message message, Channel channel) throws IOException {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
String messageBody = new String(message.getBody());
log.info("收到死信消息: message={}, deliveryTag={}", messageBody, deliveryTag);
// 死信消息處理邏輯
processDeadLetter(messageBody);
channel.basicAck(deliveryTag, false);
log.info("死信消息處理成功: deliveryTag={}", deliveryTag);
} catch (Exception e) {
log.error("死信消息處理異常: deliveryTag={}, error={}", deliveryTag, e.getMessage());
channel.basicReject(deliveryTag, false);
}
}
private void processDeadLetter(String message) {
// 實現(xiàn)死信消息處理邏輯
log.info("處理死信消息: {}", message);
}
}6. 接口控制器
@RestController
@RequestMapping("/api/mq")
@Slf4j
public class MessageController {
@Autowired
private MessageProducer messageProducer;
@PostMapping("/send")
public ResponseEntity<String> sendMessage(@RequestBody MessageDTO message) {
try {
messageProducer.sendMessage(message.getContent(),
RabbitMQConfig.BUSINESS_EXCHANGE,
RabbitMQConfig.BUSINESS_KEY);
return ResponseEntity.ok("消息發(fā)送成功");
} catch (Exception e) {
log.error("消息發(fā)送失敗: {}", e.getMessage());
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
.body("消息發(fā)送失敗: " + e.getMessage());
}
}
@PostMapping("/send/delay")
public ResponseEntity<String> sendDelayMessage(
@RequestBody MessageDTO message,
@RequestParam long delayMillis) {
try {
messageProducer.sendDelayMessage(message.getContent(),
RabbitMQConfig.BUSINESS_EXCHANGE,
RabbitMQConfig.BUSINESS_KEY,
delayMillis);
return ResponseEntity.ok("延遲消息發(fā)送成功");
} catch (Exception e) {
log.error("延遲消息發(fā)送失敗: {}", e.getMessage());
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
.body("延遲消息發(fā)送失敗: " + e.getMessage());
}
}
}7. 監(jiān)控與運維
7.1 RabbitMQ管理界面
- 訪問地址：http://localhost:15672
- 默認賬號：guest/guest
- 主要功能：
- 隊列監(jiān)控
- 交換機管理
- 連接狀態(tài)
- 消息追蹤
7.2 Prometheus + Grafana監(jiān)控
# prometheus.yml
scrape_configs:
- job_name: 'rabbitmq'
static_configs:
- targets: ['localhost:15692']7.3 日志配置
logging:
level:
org.springframework.amqp: INFO
com.your.package: DEBUG
pattern:
console: "%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n"7.4 告警配置
@Configuration
public class RabbitMQAlertConfig {
@Value("${alert.dingtalk.webhook}")
private String webhookUrl;
@Bean
public AlertService alertService() {
return new DingTalkAlertService(webhookUrl);
}
}8. 最佳實踐
8.1 消息冪等性處理
@Service
public class MessageIdempotentHandler {
@Autowired
private RedisTemplate<String, String> redisTemplate;
public boolean isProcessed(String messageId) {
String key = "mq:processed:" + messageId;
return Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(key, "1", 24, TimeUnit.HOURS));
}
}8.2 消息重試策略
@Configuration
public class RetryConfig {
@Bean
public RetryTemplate retryTemplate() {
RetryTemplate retryTemplate = new RetryTemplate();
FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
backOffPolicy.setBackOffPeriod(1000);
retryTemplate.setBackOffPolicy(backOffPolicy);
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
retryPolicy.setMaxAttempts(3);
retryTemplate.setRetryPolicy(retryPolicy);
return retryTemplate;
}
}8.3 消息序列化
@Configuration
public class MessageConverterConfig {
@Bean
public MessageConverter jsonMessageConverter() {
Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
converter.setCreateMessageIds(true);
return converter;
}
}8.4 消息追蹤
@Aspect
@Component
@Slf4j
public class MessageTraceAspect {
@Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitListener)")
public Object traceMessage(ProceedingJoinPoint joinPoint) throws Throwable {
String messageId = MDC.get("messageId");
log.info("開始處理消息: messageId={}", messageId);
try {
Object result = joinPoint.proceed();
log.info("消息處理完成: messageId={}", messageId);
return result;
} catch (Exception e) {
log.error("消息處理異常: messageId={}, error={}", messageId, e.getMessage());
throw e;
}
}
}9. 常見問題與解決方案
9.1 消息丟失問題
- 生產(chǎn)者確認機制
- 消息持久化
- 手動確認模式
- 集群高可用
9.2 消息重復消費
- 冪等性處理
- 消息去重
- 業(yè)務檢查
9.3 消息堆積問題
- 增加消費者數(shù)量
- 提高處理效率
- 隊列分片
- 死信隊列處理
9.4 性能優(yōu)化
- 合理設置預取數(shù)量
- 批量確認消息
- 消息壓縮
- 連接池優(yōu)化
10. 高可用部署
10.1 集群配置
spring:
rabbitmq:
addresses: rabbit1:5672,rabbit2:5672,rabbit3:5672
username: admin
password: password
virtual-host: /10.2 鏡像隊列
# 設置鏡像策略
rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'10.3 負載均衡
# nginx.conf
upstream rabbitmq_cluster {
server rabbit1:15672;
server rabbit2:15672;
server rabbit3:15672;
}11. 參考資源
到此這篇關于Spring Boot 3 集成 RabbitMQ 實踐指南的文章就介紹到這了,更多相關Spring Boot 3 集成 RabbitMQ內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
- RabbitMQ 3.9.7 鏡像模式集群與Springboot 2.5.5 整合
- SpringAMQP消息隊列(SpringBoot集成RabbitMQ方式)
- 一文掌握Springboot集成RabbitMQ的方法
- springboot2.5.6集成RabbitMq實現(xiàn)Topic主題模式(推薦)
- Springboot集成RabbitMQ死信隊列的實現(xiàn)
- SpringBoot集成RabbitMQ的方法(死信隊列)
- springboot2.0集成rabbitmq的示例代碼
- Spring Boot系列教程之7步集成RabbitMQ的方法
- springboot集成rabbitMQ之對象傳輸?shù)姆椒?/a>
- spring boot集成rabbitmq的實例教程
- 詳解spring boot集成RabbitMQ
相關文章
今天帶大家學習Java并發(fā)編程的相關知識,文中對Java ReentrantLock作了非常詳細的圖文示例,對正在學習java的小伙伴們有很好地幫助,需要的朋友可以參考下2021-05-05
TransactionSynchronization的invokeAfterCompletion事務源碼解析
這篇文章主要為大家介紹了TransactionSynchronization的invokeAfterCompletion事務源碼解析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2023-09-09
Spring @Cacheable自定義緩存過期時間的實現(xiàn)示例
本文主要介紹了Spring @Cacheable自定義緩存過期時間的實現(xiàn)示例,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2024-05-05
Spring Boot引入swagger-ui 后swagger-ui.html無法訪問404的問題
這篇文章主要介紹了Spring Boot引入swagger-ui 后swagger-ui.html無法訪問404的問題及解決方法，本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2020-09-09
之前工作需要,查了挺多種Excel轉PDF的方法,下面這篇文章主要給大家介紹了關于java將excel轉為pdf的相關資料,文中通過代碼介紹的非常詳細,需要的朋友可以參考下2024-01-01
反射是框架的靈魂，Java框架底層都是用反射機制+xml配置等來實現(xiàn)的，本文將通過示例詳細講解Java中的反射機制，感興趣的小伙伴可以跟隨小編學習一下2022-06-06
這篇文章主要為大家介紹了Kotlin內存陷阱inline使用技巧示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-10-10
這篇文章主要介紹了Java實現(xiàn)字符串倒序輸出的四種方法匯總，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-06-06
java中字符串轉整數(shù)及MyAtoi方法的實現(xiàn)
這篇文章主要介紹了java中字符串轉整數(shù)及MyAtoi方法的實現(xiàn)的相關資料,需要的朋友可以參考下2017-05-05
詳談hibernate,jpa與spring?data?jpa三者之間的關系
這篇文章主要介紹了hibernate,jpa與spring?data?jpa三者之間的關系，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-11-11