延遲隊列介紹

延遲隊列概念：
延時隊列,隊列內(nèi)部是有序的，最重要的特性就體現(xiàn)在它的延時屬性上，延時隊列中的元素是希望 在指定時間到了以后或之前取出和處理，簡單來說，延時隊列就是用來存放需要在指定時間被處理的 元素的隊列。

延遲隊列使用場景：
1.訂單在十分鐘之內(nèi)未支付則自動取消 2.新創(chuàng)建的店鋪，如果在十天內(nèi)都沒有上傳過商品，則自動發(fā)送消息提醒。 3.用戶注冊成功后，如果三天內(nèi)沒有登陸則進(jìn)行短信提醒。 4.用戶發(fā)起退款，如果三天內(nèi)沒有得到處理則通知相關(guān)運營人員。 5.預(yù)定會議后，需要在預(yù)定的時間點前十分鐘通知各個與會人員參加會議

這些場景都有一個特點，需要在某個事件發(fā)生之后或者之前的指定時間點完成某一項任務(wù)，如： 發(fā)生訂單生成事件，在十分鐘之后檢查該訂單支付狀態(tài)，然后將未支付的訂單進(jìn)行關(guān)閉；那我們一直輪詢數(shù)據(jù)，每秒查一次，取出需要被處理的數(shù)據(jù)，然后處理不就完事了嗎？

如果數(shù)據(jù)量比較少，確實可以這樣做，比如：對于“如果賬單一周內(nèi)未支付則進(jìn)行自動結(jié)算”這樣的需求， 如果對于時間不是嚴(yán)格限制，而是寬松意義上的一周，那么每天晚上跑個定時任務(wù)檢查一下所有未支付的賬單，確實也是一個可行的方案。但對于數(shù)據(jù)量比較大，并且時效性較強(qiáng)的場景，如：“訂單十分鐘內(nèi)未支付則關(guān)閉“，短期內(nèi)未支付的訂單數(shù)據(jù)可能會有很多，活動期間甚至?xí)_(dá)到百萬甚至千萬級別，對這么龐大的數(shù)據(jù)量仍舊使用輪詢的方式顯然是不可取的，很可能在一秒內(nèi)無法完成所有訂單的檢查，同時會給數(shù)據(jù)庫帶來很大壓力，無法滿足業(yè)務(wù)要求而且性能低下。

RabbitMQ 中的 TTL

TTL 是什么呢？TTL 是 RabbitMQ 中一個消息或者隊列的屬性，表明一條消息或者該隊列中的所有消息的最大存活時間，單位是毫秒。

換句話說，如果一條消息設(shè)置了 TTL 屬性或者進(jìn)入了設(shè)置TTL 屬性的隊列，那么這條消息如果在 TTL 設(shè)置的時間內(nèi)沒有被消費，則會成為"死信"。如果同時配置了隊列的TTL 和消息的 TTL，那么較小的那個值將會被使用，有兩種方式設(shè)置 TTL。

隊列設(shè)置TTL
在創(chuàng)建隊列的時候設(shè)置隊列的“x-message-ttl”屬性

消息設(shè)置TTL
是針對每條消息設(shè)置TTL

兩者的區(qū)別

如果設(shè)置了隊列的 TTL 屬性，那么一旦消息過期，就會被隊列丟棄(如果配置了死信隊列被丟到死信隊列中)，而第二種方式，消息即使過期，也不一定會被馬上丟棄，因為消息是否過期是在即將投遞到消費者之前判定的，如果當(dāng)前隊列有嚴(yán)重的消息積壓情況，則已過期的消息也許還能存活較長時間；

另外，還需要注意的一點是，如果不設(shè)置 TTL，表示消息永遠(yuǎn)不會過期，如果將 TTL 設(shè)置為 0，則表示除非此時可以直接投遞該消息到消費者，否則該消息將會被丟棄。

整合 springboot

前一小節(jié)我們介紹了死信隊列，剛剛又介紹了 TTL，至此利用 RabbitMQ 實現(xiàn)延時隊列的兩大要素已經(jīng)集齊，接下來只需要將它們進(jìn)行融合，再加入一點點調(diào)味料，延時隊列就可以新鮮出爐了。想想看，延時隊列，不就是想要消息延遲多久被處理嗎，TTL 則剛好能讓消息在延遲多久之后成為死信，另一方面， 成為死信的消息都會被投遞到死信隊列里，這樣只需要消費者一直消費死信隊列里的消息就完事了，因為里面的消息都是希望被立即處理的消息。

1、創(chuàng)建一個空項目：

2、添加依賴：

<dependencies>
   <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <!--RabbitMQ 依賴-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.47</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
    <!--swagger-->
    <dependency>
        <groupId>io.springfox</groupId>
        <artifactId>springfox-swagger2</artifactId>
        <version>3.0.0</version>
    </dependency>
    <dependency>
        <groupId>io.springfox</groupId>
        <artifactId>springfox-swagger-ui</artifactId>
        <version>3.0.0</version>
    </dependency>
    <!--RabbitMQ 測試依賴-->
    <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

3、修改配置文件

spring.rabbitmq.host=42.192.149.71
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456

4、添加Swagger 配置類

package com.oddfar.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import springfox.documentation.builders.ApiInfoBuilder;
import springfox.documentation.service.ApiInfo;
import springfox.documentation.service.Contact;
import springfox.documentation.spi.DocumentationType;
import springfox.documentation.spring.web.plugins.Docket;
import springfox.documentation.swagger2.annotations.EnableSwagger2;


@Configuration
@EnableSwagger2
public class SwaggerConfig {

    @Bean
    public Docket webApiConfig() {
        return new Docket(DocumentationType.SWAGGER_2)
                .groupName("webApi")
                .apiInfo(webApiInfo())
                .select()
                .build();
    }

    private ApiInfo webApiInfo() {
        return new ApiInfoBuilder()
                .title("rabbitmq 接口文檔")
                .description("本文檔描述了 rabbitmq 微服務(wù)接口定義")
                .version("1.0")
                .contact(new Contact("zhiyuan", "http://oddfar.com", "test@qq.com"))
                .build();
    }

}

隊列 TTL

代碼架構(gòu)圖
創(chuàng)建兩個隊列 QA 和 QB，兩者隊列 TTL 分別設(shè)置為 10S 和 40S，然后在創(chuàng)建一個交換機(jī) X 和死信交 換機(jī) Y，它們的類型都是direct，創(chuàng)建一個死信隊列 QD，它們的綁定關(guān)系如下：

原先配置隊列信息，寫在了生產(chǎn)者和消費者代碼中，現(xiàn)在可寫咋配置類中，生產(chǎn)者只發(fā)消息，消費者只接受消息

1、配置文件類代碼 ：

package com.oddfar.config;

import org.springframework.amqp.core.*;
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;

@Configuration
public class TtlQueueConfig {
    public static final String X_EXCHANGE = "X";
    public static final String QUEUE_A = "QA";
    public static final String QUEUE_B = "QB";
    //死信交換機(jī)
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    //死信隊列
    public static final String DEAD_LETTER_QUEUE = "QD";

    // 聲明 xExchange
    @Bean("xExchange")
    public DirectExchange xExchange() {
        return new DirectExchange(X_EXCHANGE);
    }

    // 聲明 死信隊列交換機(jī)
    @Bean("yExchange")
    public DirectExchange yExchange() {
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }

    //聲明隊列 A ttl 為 10s 并綁定到對應(yīng)的死信交換機(jī)
    @Bean("queueA")
    public Queue queueA() {
        Map<String, Object> args = new HashMap<>(3);
        //聲明當(dāng)前隊列綁定的死信交換機(jī)
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //聲明當(dāng)前隊列的死信路由 key
        args.put("x-dead-letter-routing-key", "YD");
        //聲明隊列的 TTL
        args.put("x-message-ttl", 10000);
        return QueueBuilder.durable(QUEUE_A).withArguments(args).build();
    }

    // 聲明隊列 A 綁定 X 交換機(jī)
    @Bean
    public Binding queueaBindingX(@Qualifier("queueA") Queue queueA,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }

    //聲明隊列 B ttl 為 40s 并綁定到對應(yīng)的死信交換機(jī)
    @Bean("queueB")
    public Queue queueB() {
        Map<String, Object> args = new HashMap<>(3);
        //聲明當(dāng)前隊列綁定的死信交換機(jī)
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //聲明當(dāng)前隊列的死信路由 key
        args.put("x-dead-letter-routing-key", "YD");
        //聲明隊列的 TTL
        args.put("x-message-ttl", 40000);
        return QueueBuilder.durable(QUEUE_B).withArguments(args).build();
    }

    //聲明隊列 B 綁定 X 交換機(jī)
    @Bean
    public Binding queuebBindingX(@Qualifier("queueB") Queue queue1B,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queue1B).to(xExchange).with("XB");
    }

    //聲明死信隊列 QD
    @Bean("queueD")
    public Queue queueD() {
        return new Queue(DEAD_LETTER_QUEUE);
    }

    //聲明死信隊列 QD 綁定關(guān)系
    @Bean
    public Binding deadLetterBindingQAD(@Qualifier("queueD") Queue queueD,
                                        @Qualifier("yExchange") DirectExchange yExchange) {
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");
    }

}

2、消息生產(chǎn)者代碼

package com.oddfar.contorller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;

@Slf4j
@RequestMapping("ttl")
@RestController
public class SendMsgController {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("sendMsg/{message}")
    public void sendMsg(@PathVariable String message) {
        log.info("當(dāng)前時間：{},發(fā)送一條信息給兩個 TTL 隊列:{}", new Date(), message);
        rabbitTemplate.convertAndSend("X", "XA", "消息來自 ttl 為 10S 的隊列: " + message);
        rabbitTemplate.convertAndSend("X", "XB", "消息來自 ttl 為 40S 的隊列: " + message);
    }
    
}

3、消息消費者代碼

package com.oddfar.consumer;

import com.rabbitmq.client.Channel;
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.io.IOException;
import java.util.Date;

/**
 * 消費者 - 死信隊列
 */
@Slf4j
@Component
public class DeadLetterQueueConsumer {

    @RabbitListener(queues = "QD")
    public void receiveD(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info("當(dāng)前時間：{},收到死信隊列信息{}", new Date().toString(), msg);
    }

}

發(fā)起一個請求 http://localhost:8080/ttl/sendMsg/嘻嘻嘻

第一條消息在 10S 后變成了死信消息，然后被消費者消費掉，第二條消息在 40S 之后變成了死信消息， 然后被消費掉，這樣一個延時隊列就打造完成了。

不過，如果這樣使用的話，豈不是每增加一個新的時間需求，就要新增一個隊列，這里只有 10S 和 40S 兩個時間選項，如果需要一個小時后處理，那么就需要增加TTL 為一個小時的隊列，如果是預(yù)定會議室然后提前通知這樣的場景，豈不是要增加無數(shù)個隊列才能滿足需求？

延時隊列TTL優(yōu)化

在這里新增了一個隊列 QC,綁定關(guān)系如下,該隊列不設(shè)置TTL 時間

配置文件類代碼：

@Configuration
public class MsgTtlQueueConfig {
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    public static final String QUEUE_C = "QC";

    //聲明隊列 C 死信交換機(jī)
    @Bean("queueC")
    public Queue queueB() {
        Map<String, Object> args = new HashMap<>(3);
        //聲明當(dāng)前隊列綁定的死信交換機(jī)
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //聲明當(dāng)前隊列的死信路由 key
        args.put("x-dead-letter-routing-key", "YD");
        //沒有聲明 TTL 屬性
        return QueueBuilder.durable(QUEUE_C).withArguments(args).build();
    }

    //聲明隊列 B 綁定 X 交換機(jī)
    @Bean
    public Binding queuecBindingX(@Qualifier("queueC") Queue queueC,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueC).to(xExchange).with("XC");
    }
}

生產(chǎn)者代碼：

/**
 * 延時隊列優(yōu)化
 * @param message 消息
 * @param ttlTime 延時的毫秒
 */
@GetMapping("sendExpirationMsg/{message}/{ttlTime}")
public void sendMsg(@PathVariable String message, @PathVariable String ttlTime) {
    rabbitTemplate.convertAndSend("X", "XC", message, correlationData -> {
        correlationData.getMessageProperties().setExpiration(ttlTime);
        return correlationData;
    });
    log.info("當(dāng)前時間：{},發(fā)送一條時長{}毫秒 TTL 信息給隊列 C:{}", new Date(), ttlTime, message);
}

發(fā)起請求

http://localhost:8080/ttl/sendExpirationMsg/你好1/20000

http://localhost:8080/ttl/sendExpirationMsg/你好2/2000

看起來似乎沒什么問題，但是在最開始的時候，就介紹過如果使用在消息屬性上設(shè)置 TTL 的方式，消息可能并不會按時“死亡“

因為 RabbitMQ 只會檢查第一個消息是否過期，如果過期則丟到死信隊列， 如果第一個消息的延時時長很長，而第二個消息的延時時長很短，第二個消息并不會優(yōu)先得到執(zhí)行。

這也就是為什么第二個延時2秒，卻后執(zhí)行。

Rabbitmq 插件實現(xiàn)延遲隊列

上文中提到的問題，確實是一個問題，如果不能實現(xiàn)在消息粒度上的 TTL，并使其在設(shè)置的TTL 時間及時死亡，就無法設(shè)計成一個通用的延時隊列。那如何解決呢，接下來我們就去解決該問題。

安裝延時隊列插件

可去官網(wǎng)下載 rabbitmq_delayed_message_exchange 插件，放置到 RabbitMQ 的插件目錄。

進(jìn)入 RabbitMQ 的安裝目錄下的 plgins 目錄，執(zhí)行下面命令讓該插件生效，然后重啟 RabbitMQ

# ls
erlang-21.3.8.21-1.el7.x86_64.rpm  rabbitmq_delayed_message_exchange-3.8.0.ez  rabbitmq-server-3.8.8-1.el7.noarch.rpm
#移動
cp rabbitmq_delayed_message_exchange-3.8.0.ez /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins
#安裝
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
#重啟服務(wù)
systemctl restart rabbitmq-server

代碼

在這里新增了一個隊列delayed.queue,一個自定義交換機(jī) delayed.exchange，綁定關(guān)系如下:

1、配置文件類代碼：

在我們自定義的交換機(jī)中，這是一種新的交換類型，該類型消息支持延遲投遞機(jī)制消息傳遞后并不會立即投遞到目標(biāo)隊列中，而是存儲在 mnesia(一個分布式數(shù)據(jù)系統(tǒng))表中，當(dāng)達(dá)到投遞時間時，才投遞到目標(biāo)隊列中。

@Configuration
public class DelayedQueueConfig {
    public static final String DELAYED_QUEUE_NAME = "delayed.queue";
    public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
    public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";

    @Bean
    public Queue delayedQueue() {
        return new Queue(DELAYED_QUEUE_NAME);
    }

    //自定義交換機(jī) 我們在這里定義的是一個延遲交換機(jī)
    @Bean
    public CustomExchange delayedExchange() {
        Map<String, Object> args = new HashMap<>();
        //自定義交換機(jī)的類型
        args.put("x-delayed-type", "direct");
        return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false, args);
    }

    @Bean
    public Binding bindingDelayedQueue(@Qualifier("delayedQueue") Queue queue,
                                       @Qualifier("delayedExchange") CustomExchange delayedExchange) {
        return BindingBuilder.bind(queue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
    }

}

2、生產(chǎn)者代碼

@GetMapping("sendDelayMsg/{message}/{delayTime}")
public void sendMsg(@PathVariable String message, @PathVariable Integer delayTime) {
    rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, message,
            correlationData -> {
                correlationData.getMessageProperties().setDelay(delayTime);
                return correlationData;
            });
    log.info(" 當(dāng) 前 時 間 ： {}, 發(fā) 送 一 條 延 遲 {} 毫秒的信息給隊列 delayed.queue:{}", new Date(), delayTime, message);
}

3、消費者代碼

/**
 * 消費者 - 基于插件的延時隊列
 */
@Slf4j
@ComponentScan
public class DelayQueueConsumer {

    public static final String DELAYED_QUEUE_NAME = "delayed.queue";

    @RabbitListener(queues = DELAYED_QUEUE_NAME)
    public void receiveDelayedQueue(Message message) {
        String msg = new String(message.getBody());
        log.info("當(dāng)前時間：{},收到延時隊列的消息：{}", new Date().toString(), msg);
    }
}

發(fā)送請求：

http://localhost:8080/ttl/sendDelayMsg/hello1/20000
http://localhost:8080/ttl/sendDelayMsg/hello2/2000

第二個消息被先消費掉了，符合預(yù)期

總結(jié)

延時隊列在需要延時處理的場景下非常有用，使用 RabbitMQ 來實現(xiàn)延時隊列可以很好的利用 RabbitMQ 的特性，如：消息可靠發(fā)送、消息可靠投遞、死信隊列來保障消息至少被消費一次以及未被正確處理的消息不會被丟棄。另外，通過 RabbitMQ 集群的特性，可以很好的解決單點故障問題，不會因為 單個節(jié)點掛掉導(dǎo)致延時隊列不可用或者消息丟失。

當(dāng)然，延時隊列還有很多其它選擇，比如利用 Java 的 DelayQueue，利用 Redis 的 zset，利用 Quartz 或者利用 kafka 的時間輪，這些方式各有特點,看需要適用的場景文章來源地址http://www.zghlxwxcb.cn/news/detail-487482.html

到了這里，關(guān)于RabbitMQ - 延遲隊列的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！