RabbitMQ實現延遲通知的兩種方案
更新時間：2025年10月23日 08:26:21 作者：匆匆忙忙游刃有余
延遲通知是指消息在發(fā)送后不會立即被消費,而是在指定的時間延遲后才被處理的消息傳遞機制,
一、延遲通知概述
延遲通知是指消息在發(fā)送后不會立即被消費，而是在指定的時間延遲后才被處理的消息傳遞機制。常見應用場景包括：
- 訂單超時自動取消
- 定時任務調度
- 會議/活動前提醒
- 賬單到期通知
二、RabbitMQ 實現延遲通知的兩種方案
方案對比
| 實現方式 | 優(yōu)點 | 缺點 | 適用場景 |
|---|---|---|---|
| TTL + 死信隊列 | 無需安裝插件，原生支持 | 1. 隊列級TTL不支持動態(tài)延遲 2. 消息級TTL存在性能問題 | 延遲時間固定或較少變化的場景 |
| 延遲插件 | 1. 支持每條消息單獨設置延遲時間 2. 性能更好 3. 配置簡單 | 需要安裝額外插件 | 延遲時間不固定，需要靈活設置的場景 |
三、方案一：基于TTL和死信隊列實現
1. 原理
- 利用消息或隊列的TTL（Time-To-Live）特性使消息過期
- 配置死信交換機（DLX）接收過期消息
- 將死信消息路由到實際處理隊列
2. 代碼實現
2.1 配置類
package com.example.delaynotify.config;
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 TtlDelayConfig {
// 普通交換機
public static final String DELAY_EXCHANGE = "delay_exchange";
// 普通隊列
public static final String DELAY_QUEUE = "delay_queue";
// 死信交換機
public static final String DEAD_LETTER_EXCHANGE = "dead_letter_exchange";
// 死信隊列
public static final String DEAD_LETTER_QUEUE = "dead_letter_queue";
// 路由鍵
public static final String DELAY_ROUTING_KEY = "delay.key";
public static final String DEAD_LETTER_ROUTING_KEY = "dead.letter.key";
// 聲明死信交換機
@Bean
public Exchange deadLetterExchange() {
return ExchangeBuilder.directExchange(DEAD_LETTER_EXCHANGE).durable(true).build();
}
// 聲明死信隊列
@Bean
public Queue deadLetterQueue() {
return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
}
// 聲明普通交換機
@Bean
public Exchange delayExchange() {
return ExchangeBuilder.directExchange(DELAY_EXCHANGE).durable(true).build();
}
// 聲明延遲隊列并綁定死信交換機
@Bean
public Queue delayQueue() {
Map<String, Object> args = new HashMap<>();
// 設置死信交換機
args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
// 設置死信路由鍵
args.put("x-dead-letter-routing-key", DEAD_LETTER_ROUTING_KEY);
// 隊列級TTL (10秒) - 如果需要消息級TTL可以不設置此參數
args.put("x-message-ttl", 10000);
return QueueBuilder.durable(DELAY_QUEUE)
.withArguments(args)
.build();
}
// 綁定普通隊列和普通交換機
@Bean
public Binding delayBinding() {
return BindingBuilder.bind(delayQueue())
.to(delayExchange())
.with(DELAY_ROUTING_KEY)
.noargs();
}
// 綁定死信隊列和死信交換機
@Bean
public Binding deadLetterBinding() {
return BindingBuilder.bind(deadLetterQueue())
.to(deadLetterExchange())
.with(DEAD_LETTER_ROUTING_KEY)
.noargs();
}
}
2.2 生產者 - 發(fā)送延遲消息
package com.example.delaynotify.service;
import com.example.delaynotify.config.TtlDelayConfig;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class TtlDelayMessageService {
@Autowired
private RabbitTemplate rabbitTemplate;
// 發(fā)送固定延遲時間的消息（隊列級TTL）
public void sendFixedDelayMessage(String message) {
System.out.println("發(fā)送固定延遲消息: " + message + ", 時間: " + System.currentTimeMillis());
rabbitTemplate.convertAndSend(
TtlDelayConfig.DELAY_EXCHANGE,
TtlDelayConfig.DELAY_ROUTING_KEY,
message
);
}
// 發(fā)送自定義延遲時間的消息（消息級TTL）
public void sendCustomDelayMessage(String message, long delayMillis) {
System.out.println("發(fā)送自定義延遲消息: " + message + ", 延遲時間: " + delayMillis + "ms, 時間: " + System.currentTimeMillis());
rabbitTemplate.convertAndSend(
TtlDelayConfig.DELAY_EXCHANGE,
TtlDelayConfig.DELAY_ROUTING_KEY,
message,
new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
// 設置消息級TTL
message.getMessageProperties().setExpiration(String.valueOf(delayMillis));
return message;
}
}
);
}
}
2.3 消費者 - 接收延遲消息
package com.example.delaynotify.consumer;
import com.example.delaynotify.config.TtlDelayConfig;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
@Component
public class TtlDelayMessageConsumer {
@RabbitListener(queues = TtlDelayConfig.DEAD_LETTER_QUEUE)
public void receiveDelayMessage(String message, Channel channel, Message msg) throws IOException {
try {
System.out.println("接收到延遲消息: " + message + ", 時間: " + System.currentTimeMillis());
// 處理業(yè)務邏輯 - 例如發(fā)送通知、更新狀態(tài)等
processDelayMessage(message);
// 手動確認消息
channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
System.out.println("消息處理失敗: " + e.getMessage());
// 拒絕消息并丟棄
channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, false);
}
}
private void processDelayMessage(String message) {
// 模擬發(fā)送通知的業(yè)務邏輯
System.out.println("執(zhí)行通知業(yè)務: " + message);
// 這里可以調用郵件、短信、推送等服務
}
}
四、方案二：基于延遲插件實現
1. 安裝延遲插件
1.1 Docker環(huán)境安裝
# 下載插件（根據RabbitMQ版本選擇對應版本） wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v3.11.1/rabbitmq_delayed_message_exchange-3.11.1.ez # 復制插件到容器 docker cp rabbitmq_delayed_message_exchange-3.11.1.ez rabbitmq:/plugins # 進入容器啟用插件 docker exec -it rabbitmq bash rabbitmq-plugins enable rabbitmq_delayed_message_exchange exit # 重啟RabbitMQ容器 docker restart rabbitmq
1.2 驗證安裝
在RabbitMQ管理界面新建交換機時，如果能看到x-delayed-message類型，則表示插件安裝成功。
2. 代碼實現
2.1 配置類
package com.example.delaynotify.config;
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 PluginDelayConfig {
// 延遲交換機
public static final String DELAY_PLUGIN_EXCHANGE = "delay_plugin_exchange";
// 延遲隊列
public static final String DELAY_PLUGIN_QUEUE = "delay_plugin_queue";
// 路由鍵
public static final String DELAY_PLUGIN_ROUTING_KEY = "delay.plugin.key";
// 聲明延遲交換機（類型為x-delayed-message）
@Bean
public CustomExchange delayPluginExchange() {
Map<String, Object> args = new HashMap<>();
// 設置底層路由模式為direct
args.put("x-delayed-type", "direct");
return new CustomExchange(
DELAY_PLUGIN_EXCHANGE,
"x-delayed-message",
true, // 持久化
false, // 非自動刪除
args
);
}
// 聲明延遲隊列
@Bean
public Queue delayPluginQueue() {
return QueueBuilder.durable(DELAY_PLUGIN_QUEUE).build();
}
// 綁定延遲交換機和延遲隊列
@Bean
public Binding delayPluginBinding() {
return BindingBuilder.bind(delayPluginQueue())
.to(delayPluginExchange())
.with(DELAY_PLUGIN_ROUTING_KEY)
.noargs();
}
}
2.2 生產者 - 發(fā)送延遲消息
package com.example.delaynotify.service;
import com.example.delaynotify.config.PluginDelayConfig;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class PluginDelayMessageService {
@Autowired
private RabbitTemplate rabbitTemplate;
// 發(fā)送延遲消息
public void sendDelayMessage(String message, long delayMillis) {
System.out.println("使用插件發(fā)送延遲消息: " + message + ", 延遲時間: " + delayMillis + "ms, 時間: " + System.currentTimeMillis());
rabbitTemplate.convertAndSend(
PluginDelayConfig.DELAY_PLUGIN_EXCHANGE,
PluginDelayConfig.DELAY_PLUGIN_ROUTING_KEY,
message,
new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
// 設置延遲時間（毫秒）
message.getMessageProperties().setDelay((int) delayMillis);
return message;
}
}
);
}
}
2.3 消費者 - 接收延遲消息
package com.example.delaynotify.consumer;
import com.example.delaynotify.config.PluginDelayConfig;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
@Component
public class PluginDelayMessageConsumer {
@RabbitListener(queues = PluginDelayConfig.DELAY_PLUGIN_QUEUE)
public void receiveDelayMessage(String message, Channel channel, Message msg) throws IOException {
try {
System.out.println("接收到插件延遲消息: " + message + ", 時間: " + System.currentTimeMillis());
// 處理業(yè)務邏輯 - 例如發(fā)送通知、更新狀態(tài)等
processDelayMessage(message);
// 手動確認消息
channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
System.out.println("插件延遲消息處理失敗: " + e.getMessage());
// 拒絕消息并丟棄
channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, false);
}
}
private void processDelayMessage(String message) {
// 模擬發(fā)送通知的業(yè)務邏輯
System.out.println("執(zhí)行通知業(yè)務: " + message);
// 這里可以調用郵件、短信、推送等服務
}
}
五、Controller層實現
package com.example.delaynotify.controller;
import com.example.delaynotify.service.PluginDelayMessageService;
import com.example.delaynotify.service.TtlDelayMessageService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class DelayNotifyController {
@Autowired
private TtlDelayMessageService ttlDelayMessageService;
@Autowired
private PluginDelayMessageService pluginDelayMessageService;
// 基于TTL的固定延遲
@GetMapping("/ttl/fixed")
public String sendFixedTtlDelay(@RequestParam String message) {
ttlDelayMessageService.sendFixedDelayMessage(message);
return "固定延遲消息已發(fā)送 (10秒)";
}
// 基于TTL的自定義延遲
@GetMapping("/ttl/custom")
public String sendCustomTtlDelay(@RequestParam String message, @RequestParam long delayMillis) {
ttlDelayMessageService.sendCustomDelayMessage(message, delayMillis);
return "自定義延遲消息已發(fā)送 (" + delayMillis + "ms)";
}
// 基于插件的延遲
@GetMapping("/plugin/delay")
public String sendPluginDelay(@RequestParam String message, @RequestParam long delayMillis) {
pluginDelayMessageService.sendDelayMessage(message, delayMillis);
return "插件延遲消息已發(fā)送 (" + delayMillis + "ms)";
}
}
六、application.yml配置
spring:
rabbitmq:
host: localhost
port: 5672
username: admin
password: admin
virtual-host: /
# 生產者確認配置
publisher-confirm-type: correlated
publisher-returns: true
template:
mandatory: true
# 消費者配置
listener:
simple:
acknowledge-mode: manual
prefetch: 1
concurrency: 1
max-concurrency: 5
七、完整的通知場景實現示例
訂單超時通知場景
package com.example.delaynotify.service;
import com.example.delaynotify.config.PluginDelayConfig;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
@Service
public class OrderNotifyService {
@Autowired
private RabbitTemplate rabbitTemplate;
private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
/**
* 創(chuàng)建訂單并設置超時通知
* @param orderId 訂單ID
* @param notifyDelaySeconds 超時時間（秒）
*/
public void createOrderAndSetTimeout(String orderId, int notifyDelaySeconds) {
// 1. 保存訂單邏輯
System.out.println("創(chuàng)建訂單: " + orderId + " 時間: " + LocalDateTime.now().format(formatter));
// 2. 設置延遲通知
String notifyMessage = "訂單[" + orderId + "]已超時，需要取消處理";
long delayMillis = notifyDelaySeconds * 1000L;
System.out.println("設置訂單超時通知，延遲: " + notifyDelaySeconds + "秒");
// 使用延遲插件發(fā)送通知消息
rabbitTemplate.convertAndSend(
PluginDelayConfig.DELAY_PLUGIN_EXCHANGE,
PluginDelayConfig.DELAY_PLUGIN_ROUTING_KEY,
notifyMessage,
message -> {
message.getMessageProperties().setDelay((int) delayMillis);
return message;
}
);
}
}
八、兩種方案對比與選擇建議
1. 性能對比
- TTL+死信隊列：當使用消息級TTL時，RabbitMQ需要為每條消息設置過期時間，會造成額外的性能開銷
- 延遲插件：插件內部使用優(yōu)先隊列實現，性能更優(yōu)，特別適合大量不同延遲時間的消息場景
2. 靈活性對比
- TTL+死信隊列：如果要支持不同的延遲時間，需要創(chuàng)建多個不同TTL的隊列
- 延遲插件：每條消息都可以設置不同的延遲時間，更加靈活
3. 選擇建議
- 如果延遲時間固定或種類較少，可以使用TTL+死信隊列方案，無需安裝插件
- 如果延遲時間不固定或種類較多，強烈建議使用延遲插件方案
- 對于生產環(huán)境，建議使用延遲插件方案，性能更好、配置更簡潔
通過以上兩種方案，您可以根據實際需求選擇合適的方式實現RabbitMQ的延遲通知功能，滿足訂單超時、定時提醒等各種業(yè)務場景。
以上就是RabbitMQ實現延遲通知的兩種方案的詳細內容，更多關于RabbitMQ延遲通知的資料請關注腳本之家其它相關文章！
您可能感興趣的文章:
相關文章
jenv是一個用于管理?Java?環(huán)境的工具,它允許用戶在不同的Java版本之間輕松切換,這篇文章主要介紹了通過Jenv管理多JDK版本的詳細介紹,文中通過代碼介紹的非常詳細,需要的朋友可以參考下2026-05-05
這篇文章主要為大家介紹了Spring?Data?Jpa框架最佳實踐示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步早日升職加薪2022-02-02
這篇文章主要介紹了java中正則表達式實例詳解的相關資料,需要的朋友可以參考下2017-04-04
本篇文章主要介紹了Spring Boot 配置大全（小結），小編覺得挺不錯的，現在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2017-11-11
這篇文章主要介紹了ElasticSearch的安裝與基本概念,提供了一個分布式多用戶能力的全文搜索引擎，Elasticsearch是用Java開發(fā)的,需要的朋友可以參考下2023-04-04
今天小編就為大家分享一篇Java 判斷兩個字符串是否由相同的字符組成的實例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-07-07
Java+Freemarker實現根據XML模板文件生成Word文檔
這篇文章主要為大家詳細介紹了Java如何使用Freemarker實現根據XML模板文件生成Word文檔,文中的示例代碼講解詳細,感興趣的小伙伴可以學習一下2023-11-11
這篇文章主要為大家全面解析Java8觀察者模式，通過在 Java8 環(huán)境下實現觀察者模式的實例，進一步介紹了什么是觀察者模式、專業(yè)化及其命名規(guī)則，感興趣的小伙伴們可以參考一下2016-02-02
Spring?boot事務無效報錯:Transaction?not?enabled問題排查解決
在業(yè)務代碼中經常需要保證事務的原子性,但是有的時候確實是出現事務沒有生效,這篇文章主要給大家介紹了關于Spring?boot事務無效報錯:Transaction?not?enabled問題排查的相關資料,需要的朋友可以參考下2023-11-11
Java動態(tài)字節(jié)碼注入技術是一種在運行時修改Java字節(jié)碼的技術,本文主要介紹了Java動態(tài)字節(jié)碼注入技術的實現,具有一定的參考價值,感興趣的可以了解一下2023-08-08