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