碼上快樂

相關內容    簡體    繁體

RabbitMQ-延遲隊列

本文轉載自   查看原文   2021-09-16 23:34   1138    RabbitMQ/ MiddleWare/ SpringBoot

1. 簡介

我們在上一篇博文中遺留了一個小問題，就是雖然TTL + DLX能實現延遲隊列的功能，但是有兩個問題。

首先業務場景為：比如海底撈預約，每個人預約的時間段不一致，有個可能一個小時后，有的可能三個小時等，當快到預約時間點需要給用戶進行短信通知。

  1. 通過給Queue設置過期時間的方式不現實，因為很有可能每條記錄的過期時間都不一樣，不可能設置那么多的Queue。
  2. 直接給Message設置過期時間，這種方式也不好，因為這種方式是當該消息在隊列頭部時（消費時），才會單獨判斷這一消息是否過期。例：現在有兩條消息，第一條消息過期時間為30s，而第二條消息過期時間為15s，當過了15秒后，第二條消息不會立即過期，而是要等第一條消息被消費后，第二條消息被消費時，才會判斷是否過期，也就是等到第二條消息投往DLX已經過去45s了。

這也就拋出了本章主題：延遲隊列。

RabbitMQ默認沒有提供延遲隊列功能，而是要通過插件提供的x-delayed-message（延遲交換機）來實現。

延遲隊列：用戶可以使用該類型聲明一個交換，x-delayed-message然后使用自定義標頭發布消息，x-delay以毫秒為單位表示消息的延遲時間。消息將在x-delay毫秒后傳遞到相應的隊列。

2. 安裝插件

官方插件地址：https://www.rabbitmq.com/community-plugins.html

找到插件rabbitmq_delayed_message_exchange，進入GitHub下載本地RabbitMQ對應的插件版本（下載.ez文件）。

我這里下載的是3.8.9版本，如圖：

下載到本地后將文件放置RabbitMQ的plugins目錄。

我這里本地是使用docker-compose安裝的服務，image為rabbitmq:3.8.3-management（雖然版本沒對起來，但是測試能用，但是使用3.9的版本會報錯，插件安裝失敗）安裝的服務，操作步驟如下：

  1. 將下載好的文件放置RabbitMQ插件目錄

    rabbitmq：容器服務名

    $ docker cp /Users/ludangxin/Downloads/rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez rabbitmq:/opt/rabbitmq/plugins/

  2. 進入容器

    $ docker exec -it rabbitmq /bin/bash

  3. 查看現有的插件列表

    $ rabbitmq-plugins list
# 輸出部分內容如下 [E*] = 明確啟用; e = 隱式啟用
[  ] rabbitmq_amqp1_0                  3.8.3
[  ] rabbitmq_auth_backend_cache       3.8.3
[  ] rabbitmq_auth_backend_http        3.8.3
[  ] rabbitmq_auth_backend_ldap        3.8.3
[  ] rabbitmq_auth_backend_oauth2      3.8.3
[  ] rabbitmq_auth_mechanism_ssl       3.8.3
[  ] rabbitmq_consistent_hash_exchange 3.8.3
[  ] rabbitmq_event_exchange           3.8.3
[  ] rabbitmq_federation               3.8.3
[  ] rabbitmq_federation_management    3.8.3
[  ] rabbitmq_jms_topic_exchange       3.8.3
[E*] rabbitmq_management               3.8.3
[e*] rabbitmq_management_agent         3.8.3
[  ] rabbitmq_mqtt                     3.8.3

  4. 啟用插件

    $ rabbitmq-plugins enable rabbitmq_delayed_message_exchange

  5. 再次查看安裝列表就有了rabbitmq_delayed_message_exchange

安裝完畢后登陸RabbitMQ控制台查看，會發現多了個x-delayed-message類型的Exchange。

3. 實現延遲隊列

3.1 引入所需依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <optional>true</optional>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit-test</artifactId>
    <scope>test</scope>
</dependency>

3.2 application.yaml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    # rabbit 默認的虛擬主機
    virtual-host: /
    # rabbit 用戶名密碼
    username: admin
    password: admin123

3.3 RabbitConfig

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;

/**
 * 延遲隊列配置
 *
 * @author ludangxin
 * @date 2021/9/16
 */
@Configuration
public class RabbitDelayedConfig {
    public static final String QUEUE_NAME_DELAYED = "DELAY.QUEUE";
    public static final String EXCHANGE_NAME_DELAYED = "DELAY.EXCHANGE";
    public static final String ROUTING_KEY_DELAYED = "DELAY.#";

    @Bean(QUEUE_NAME_DELAYED)
    public Queue queue() {
       return QueueBuilder.durable(QUEUE_NAME_DELAYED).build();
    }

    @Bean(EXCHANGE_NAME_DELAYED)
    public CustomExchange exchange() {
       Map<String, Object> arguments = new HashMap<>(1);
       // 在這里聲明一個主題類型的延遲隊列，當然其他類型的也可以。
       arguments.put("x-delayed-type", "topic");
       return new CustomExchange(EXCHANGE_NAME_DELAYED, "x-delayed-message", true, false, arguments);
    }

    @Bean
    public Binding bindingNotify(@Qualifier(QUEUE_NAME_DELAYED) Queue queue, @Qualifier(EXCHANGE_NAME_DELAYED) CustomExchange customExchange) {
       return BindingBuilder.bind(queue).to(customExchange).with(ROUTING_KEY_DELAYED).noargs();
    }
}

3.4 Producer

import com.ldx.rabbitmq.config.RabbitDelayedConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * 延遲消息生產者
 *
 * @author ludangxin
 * @date 2021/9/9
 */
@Component
public class DelayProducer {

   @Autowired
   private RabbitTemplate rabbitTemplate;

   public void sendDelayedMsg(String msg, Integer delay) {
      MessageProperties mp = new MessageProperties();
      // 設置過期時間
      mp.setDelay(delay);
      Message message = new Message(msg.getBytes(), mp);
      rabbitTemplate.convertAndSend(RabbitDelayedConfig.EXCHANGE_NAME_DELAYED, "DELAY.MSG", message);
   }
}

3.5 Consumer

import com.ldx.rabbitmq.config.RabbitDelayedConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * 延遲消息消費者
 *
 * @author ludangxin
 * @date 2021/9/9
 */
@Slf4j
@Component
public class DelayConsumer {

    @RabbitListener(queues = {RabbitDelayedConfig.QUEUE_NAME_DELAYED})
    public void delayQueue(Message message){
        log.info(new String(message.getBody()) + "，結束時間為：" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
    }

}

3.6 測試代碼

@Autowired
private DelayProducer delayProducer;

@Test
@SneakyThrows
public void sendDelayedMsg() {
   for(int i = 16; i >= 10; i --) {
      String msg = "我將在" + i + "s后過期，開始時間為：" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
      delayProducer.sendDelayedMsg(msg,i * 1000);
   }
   // 使進程阻塞，方便Consumer監聽輸出Message
   System.in.read();
}

3.7 啟動測試

啟動測試代碼，連續發送7條消息輸出內容如下：

從日志內容可以看出，雖然我們先發送了16s的那條消息，但最終消息的過期順序還是按照10-16s的順序，符合預期。

2021-09-16 23:40:10.806  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我將在10s后過期，開始時間為：2021-09-16 23:40:00，結束時間為：2021-09-16 23:40:10
2021-09-16 23:40:11.792  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我將在11s后過期，開始時間為：2021-09-16 23:40:00，結束時間為：2021-09-16 23:40:11
2021-09-16 23:40:12.791  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我將在12s后過期，開始時間為：2021-09-16 23:40:00，結束時間為：2021-09-16 23:40:12
2021-09-16 23:40:13.791  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我將在13s后過期，開始時間為：2021-09-16 23:40:00，結束時間為：2021-09-16 23:40:13
2021-09-16 23:40:14.788  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我將在14s后過期，開始時間為：2021-09-16 23:40:00，結束時間為：2021-09-16 23:40:14
2021-09-16 23:40:15.785  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我將在15s后過期，開始時間為：2021-09-16 23:40:00，結束時間為：2021-09-16 23:40:15
2021-09-16 23:40:16.785  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我將在16s后過期，開始時間為：2021-09-16 23:40:00，結束時間為：2021-09-16 23:40:16


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 RabbitMQ-延遲隊列插件安裝 RabbitMq-更新:正常隊列+死信隊列 rabbitmq 延遲隊列插件 Spring RabbitMQ 延遲隊列 RabbitMQ延遲隊列 RabbitMQ死信隊列與延遲隊列 rabbitmq延遲隊列demo rabbitmq 延遲隊列 rabbitmq隊列延遲 RabbitMQ 延遲隊列
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM