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

手把手帶你掌握SpringBoot RabbitMQ延遲隊(duì)列

 更新時(shí)間:2021年09月17日 15:18:16   作者:張鐵牛  
RabbitMQ 是一個(gè)由Erlang語(yǔ)言開發(fā)的AMQP的開源實(shí)現(xiàn),支持多種客戶端。用于在分布式系統(tǒng)中存儲(chǔ)轉(zhuǎn)發(fā)消息,在易用性、擴(kuò)展性、高可用性等方面表現(xiàn)不俗,下文將帶你深入了解 RabbitMQ 延遲隊(duì)列

1. 簡(jiǎn)介

我們?cè)谏弦黄┪闹羞z留了一個(gè)小問題,就是雖然TTL + DLX能實(shí)現(xiàn)延遲隊(duì)列的功能,但是有兩個(gè)問題。

首先業(yè)務(wù)場(chǎng)景為:比如海底撈預(yù)約,每個(gè)人預(yù)約的時(shí)間段不一致,有個(gè)可能一個(gè)小時(shí)后,有的可能三個(gè)小時(shí)等,當(dāng)快到預(yù)約時(shí)間點(diǎn)需要給用戶進(jìn)行短信通知。

通過給Queue設(shè)置過期時(shí)間的方式不現(xiàn)實(shí),因?yàn)楹苡锌赡苊織l記錄的過期時(shí)間都不一樣,不可能設(shè)置那么多的Queue。直接給Message設(shè)置過期時(shí)間,這種方式也不好,因?yàn)檫@種方式是當(dāng)該消息在隊(duì)列頭部時(shí)(消費(fèi)時(shí)),才會(huì)單獨(dú)判斷這一消息是否過期。例:現(xiàn)在有兩條消息,第一條消息過期時(shí)間為30s,而第二條消息過期時(shí)間為15s,當(dāng)過了15秒后,第二條消息不會(huì)立即過期,而是要等第一條消息被消費(fèi)后,第二條消息被消費(fèi)時(shí),才會(huì)判斷是否過期,也就是等到第二條消息投往DLX已經(jīng)過去45s了。

這也就拋出了本章主題:延遲隊(duì)列。

RabbitMQ默認(rèn)沒有提供延遲隊(duì)列功能,而是要通過插件提供的x-delayed-message(延遲交換機(jī))來實(shí)現(xiàn)。

延遲隊(duì)列:用戶可以使用該類型聲明一個(gè)交換,x-delayed-message然后使用自定義標(biāo)頭發(fā)布消息,x-delay以毫秒為單位表示消息的延遲時(shí)間。消息將在x-delay毫秒后傳遞到相應(yīng)的隊(duì)列。

2. 安裝插件

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

找到插件rabbitmq_delayed_message_exchange,進(jìn)入GitHub下載本地RabbitMQ對(duì)應(yīng)的插件版本(下載.ez文件)。

我這里下載的是3.8.9版本,如圖:

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

我這里本地是使用docker-compose安裝的服務(wù),imagerabbitmq:3.8.3-management(雖然版本沒對(duì)起來,但是測(cè)試能用,但是使用3.9的版本會(huì)報(bào)錯(cuò),插件安裝失?。┌惭b的服務(wù),操作步驟如下:

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

rabbitmq:容器服務(wù)名

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

2.進(jìn)入容器

$ docker exec -it rabbitmq /bin/bash

3.查看現(xiàn)有的插件列表

$ rabbitmq-plugins list
# 輸出部分內(nèi)容如下 [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

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

安裝完畢后登陸RabbitMQ控制臺(tái)查看,會(huì)發(fā)現(xiàn)多了個(gè)x-delayed-message類型的Exchange。

3. 實(shí)現(xiàn)延遲隊(duì)列

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 默認(rèn)的虛擬主機(jī)
    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;

/**
 * 延遲隊(duì)列配置
 *
 * @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);
       // 在這里聲明一個(gè)主題類型的延遲隊(duì)列,當(dāng)然其他類型的也可以。
       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;

/**
 * 延遲消息生產(chǎn)者
 *
 * @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();
      // 設(shè)置過期時(shí)間
      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;

/**
 * 延遲消息消費(fèi)者
 *
 * @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()) + ",結(jié)束時(shí)間為:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
    }

}

3.6 測(cè)試代碼

@Autowired
private DelayProducer delayProducer;

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

3.7 啟動(dòng)測(cè)試

啟動(dòng)測(cè)試代碼,輸出內(nèi)容如下:

從日志內(nèi)容可以看出,消息存活了30s,符合預(yù)期。

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

到此這篇關(guān)于手把手帶你掌握SpringBoot RabbitMQ延遲隊(duì)列的文章就介紹到這了,更多相關(guān)SpringBoot RabbitMQ 延遲隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

相關(guān)文章

  • 如何使用ActiveMQ中間件方式發(fā)送郵件

    如何使用ActiveMQ中間件方式發(fā)送郵件

    這篇文章主要介紹了如何使用ActiveMQ中間件方式發(fā)送郵件的操作,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教
    2021-07-07
  • java中類和對(duì)象的詳細(xì)介紹

    java中類和對(duì)象的詳細(xì)介紹

    這篇文章主要給大家介紹了關(guān)于java中類和對(duì)象的相關(guān)資料,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
    2021-03-03
  • Java中的鍵盤事件處理及監(jiān)聽機(jī)制解析

    Java中的鍵盤事件處理及監(jiān)聽機(jī)制解析

    這篇文章主要介紹了Java中的鍵盤事件處理及監(jiān)聽機(jī)制解析,Java事件處理采用了委派事件模型,在這個(gè)模型中,當(dāng)事件發(fā)生時(shí),產(chǎn)生事件的對(duì)象將事件信息傳遞給事件的監(jiān)聽者進(jìn)行處理,在Java中,事件源是產(chǎn)生事件的對(duì)象,比如窗口、按鈕等,需要的朋友可以參考下
    2023-10-10
  • Java實(shí)現(xiàn)高效隨機(jī)數(shù)算法的示例代碼

    Java實(shí)現(xiàn)高效隨機(jī)數(shù)算法的示例代碼

    這篇文章主要介紹了Java實(shí)現(xiàn)高效隨機(jī)數(shù)算法的示例代碼,小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,也給大家做個(gè)參考。一起跟隨小編過來看看吧
    2019-02-02
  • Java讀取.properties配置文件方法示例

    Java讀取.properties配置文件方法示例

    這篇文章主要介紹了Java讀取.properties配置文件,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
    2019-03-03
  • Mybatis中的@Select、foreach用法

    Mybatis中的@Select、foreach用法

    這篇文章主要介紹了Mybatis中的@Select、foreach用法,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教
    2022-08-08
  • IDEA EasyCode 一鍵幫你生成所需代碼

    IDEA EasyCode 一鍵幫你生成所需代碼

    這篇文章主要介紹了IDEA EasyCode 一鍵幫你生成所需代碼,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
    2020-08-08
  • java中線程中斷的實(shí)現(xiàn)示例

    java中線程中斷的實(shí)現(xiàn)示例

    中斷只是一種協(xié)作協(xié)商機(jī)制,Java沒有給中斷增加任何語(yǔ)法,中斷的過程完全需要程序員自己實(shí)現(xiàn),本文就來介紹一下線程中斷的實(shí)現(xiàn),具有一定的參考價(jià)值,感興趣的可以了解一下
    2023-09-09
  • 解決JSON.toJSONString首字母大小寫的問題

    解決JSON.toJSONString首字母大小寫的問題

    這篇文章主要介紹了解決JSON.toJSONString首字母大小寫的問題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教
    2022-02-02
  • 詳解Mybatis是如何解析配置文件的

    詳解Mybatis是如何解析配置文件的

    這篇文章主要介紹了詳解Mybatis是如何解析配置文件的,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
    2020-12-12

最新評(píng)論

寻甸| 迁安市| 秭归县| 新干县| 大埔区| 辽源市| 秦皇岛市| 梅河口市| 新邵县| 新兴县| 通州市| 乌审旗| 含山县| 宁都县| 苍山县| 承德县| 长顺县| 漳浦县| 当涂县| 桃园县| 老河口市| 长春市| 都匀市| 奉新县| 新竹市| 乐业县| 泊头市| 南召县| 平泉县| 苍山县| 观塘区| 商河县| 彭山县| 涟水县| 潜江市| 高台县| 武平县| 江门市| 景洪市| 读书| 呼伦贝尔市|