最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

RabbitMQ實現(xiàn)延遲通知的完整方案

 更新時間：2026年01月27日 08:40:48   作者：左Python右Java  
本文主要介紹了RabbitMQ實現(xiàn)延遲通知的完整方案,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧

一、延遲通知概述

延遲通知是指消息在發(fā)送后不會立即被消費，而是在指定的時間延遲后才被處理的消息傳遞機制。常見應用場景包括：

  • 訂單超時自動取消
  • 定時任務調(diào)度
  • 會議/活動前提醒
  • 賬單到期通知

二、RabbitMQ 實現(xiàn)延遲通知的兩種方案

方案對比

實現(xiàn)方式優(yōu)點缺點適用場景
TTL + 死信隊列無需安裝插件，原生支持1. 隊列級TTL不支持動態(tài)延遲
2. 消息級TTL存在性能問題		延遲時間固定或較少變化的場景
延遲插件1. 支持每條消息單獨設置延遲時間
2. 性能更好
3. 配置簡單		需要安裝額外插件延遲時間不固定，需要靈活設置的場景

三、方案一：基于TTL和死信隊列實現(xiàn)

1. 原理

  • 利用消息或隊列的TTL（Time-To-Live）特性使消息過期
  • 配置死信交換機（DLX）接收過期消息
  • 將死信消息路由到實際處理隊列

2. 代碼實現(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 {
    // 普通交換機
    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可以不設置此參數(shù)
        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 生產(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ā)送固定延遲時間的消息（隊列級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);
        // 這里可以調(diào)用郵件、短信、推送等服務
    }
}

四、方案二：基于延遲插件實現(xiàn)

1. 安裝延遲插件

1.1 Docker環(huán)境安裝

# 下載插件（根據(jù)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. 代碼實現(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 {
    // 延遲交換機
    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 生產(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 + ", 延遲時間: " + 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);
        // 這里可以調(diào)用郵件、短信、推送等服務
    }
}

五、Controller層實現(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)者確認配置
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true
    # 消費者配置
    listener:
      simple:
        acknowledge-mode: manual
        prefetch: 1
        concurrency: 1
        max-concurrency: 5

七、完整的通知場景實現(xiàn)示例

訂單超時通知場景

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需要為每條消息設置過期時間，會造成額外的性能開銷
  • 延遲插件：插件內(nèi)部使用優(yōu)先隊列實現(xiàn)，性能更優(yōu)，特別適合大量不同延遲時間的消息場景

2. 靈活性對比

  • TTL+死信隊列：如果要支持不同的延遲時間，需要創(chuàng)建多個不同TTL的隊列
  • 延遲插件：每條消息都可以設置不同的延遲時間，更加靈活

3. 選擇建議

  • 如果延遲時間固定或種類較少，可以使用TTL+死信隊列方案，無需安裝插件
  • 如果延遲時間不固定或種類較多，強烈建議使用延遲插件方案
  • 對于生產(chǎn)環(huán)境，建議使用延遲插件方案，性能更好、配置更簡潔

通過以上兩種方案，您可以根據(jù)實際需求選擇合適的方式實現(xiàn)RabbitMQ的延遲通知功能，滿足訂單超時、定時提醒等各種業(yè)務場景。

到此這篇關(guān)于RabbitMQ實現(xiàn)延遲通知的完整方案的文章就介紹到這了,更多相關(guān)RabbitMQ 延遲通知內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

相關(guān)文章

  • SpringBoot整合Minio實現(xiàn)文件上傳和讀取功能

    SpringBoot整合Minio實現(xiàn)文件上傳和讀取功能

    最近有一個需求是關(guān)于視頻上傳播放的,需要設計一個方案,中間談到了Minio這個技術(shù),于是來學習一下,所以本文給大家介紹了SpringBoot整合Minio實現(xiàn)文件上傳和讀取功能,文中有詳細的代碼示例供大家參考,需要的朋友可以參考下
    2024-07-07
  • Java如何向Word模板中插入Base64圖片和數(shù)據(jù)信息

    Java如何向Word模板中插入Base64圖片和數(shù)據(jù)信息

    這篇文章主要介紹了Java如何向Word模板中插入Base64圖片和數(shù)據(jù)信息問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教
    2024-07-07
  • Spring MVC過濾器-登錄過濾的代碼實現(xiàn)

    Spring MVC過濾器-登錄過濾的代碼實現(xiàn)

    本篇文章主要介紹了Spring MVC過濾器-登錄過濾，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧。
    2017-01-01
  • Springboot+ElementUi實現(xiàn)評論、回復、點贊功能

    Springboot+ElementUi實現(xiàn)評論、回復、點贊功能

    這篇文章主要介紹了通過Springboot ElementUi實現(xiàn)評論、回復、點贊功能。如果是自己評論的還可以刪除，刪除的規(guī)則是如果該評論下還有回復，也一并刪除。需要的可以參考一下
    2022-01-01
  • Java中閉包簡單代碼示例

    Java中閉包簡單代碼示例

    這篇文章主要介紹了Java中閉包簡單代碼示例，具有一定借鑒價值,需要的朋友可以參考下
    2018-01-01
  • java多線程學習筆記之自定義線程池

    java多線程學習筆記之自定義線程池

    本篇文章主要介紹了java多線程學習筆記之自定義線程池 ，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧
    2017-09-09
  • Java時間輪調(diào)度算法的代碼實現(xiàn)

    Java時間輪調(diào)度算法的代碼實現(xiàn)

    時間輪是一種高效的定時調(diào)度算法,主要用于管理延時任務或周期性任務,它通過一個環(huán)形數(shù)組（時間輪）和指針來實現(xiàn),將大量定時任務分攤到固定的時間槽中,極大地降低了時間復雜度和資源開銷,本文給大家介紹了Java時間輪調(diào)度算法的代碼實現(xiàn),需要的朋友可以參考下
    2025-03-03
  • Java線程池Executor用法詳解

    Java線程池Executor用法詳解

    本文主要為大家詳細介紹了Java線程池Executor的用法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
    2022-08-08
  • SpringBoot實現(xiàn)微服務通信的多種方式

    SpringBoot實現(xiàn)微服務通信的多種方式

    微服務通信是指在分布式系統(tǒng)中,各個微服務之間進行數(shù)據(jù)交互和通信的過程,今天我們將探討在Spring Boot中實現(xiàn)微服務通信的多種方式,文章通過代碼示例給大家介紹的非常詳細,需要的朋友可以參考下
    2024-07-07
  • Java如何連接數(shù)據(jù)庫圖文教程

    Java如何連接數(shù)據(jù)庫圖文教程

    關(guān)于java數(shù)據(jù)庫的文章早已是非常多了,本文是對我個人過往學習java數(shù)據(jù)庫,理解及應用java數(shù)據(jù)庫的一個總結(jié),下面這篇文章主要給大家介紹了關(guān)于Java如何連接數(shù)據(jù)庫的相關(guān)資料,需要的朋友可以參考下
    2023-04-04

最新評論

石棉县| 广元市| 苍南县| 闽侯县| 双峰县| 三江| 汕头市| 沂南县| 上犹县| 山西省| 博罗县| 汪清县| 华亭县| 江源县| 肇源县| 长泰县| 镇安县| 龙井市| 太仓市| 望谟县| 青龙| 合水县| 新田县| 津市市| 丹棱县| 龙泉市| 奈曼旗| 临洮县| 高州市| 泰来县| 辛集市| 揭东县| 辽阳县| 瑞丽市| 清新县| 北宁市| 蓬安县| 维西| 凤山县| 济南市| 道孚县|