延時佇列

延時佇列的使用場景：

1.訂單業務：在電商中，使用者下單後30分鐘後未付款則取消訂單。

2.簡訊通知：使用者下單並付款後，1分鐘後發簡訊給使用者。

延時佇列實現思路

AMQP協議和RabbitMQ佇列本身沒有直接支援延遲佇列功能，但是我們可以通過RabbitMQ的兩個特性來曲線實現延遲佇列：

特性一：Time To Live(TTL)

RabbitMQ可以針對Queue設定x-expires 或者 針對Message設定 x-message-ttl，來控制訊息的生存時間，如果超時(兩者同時設定以最先到期的時間為準)，則訊息變為dead letter(死信)
RabbitMQ針對佇列中的訊息過期時間有兩種方法可以設定。
A: 通過佇列屬性設定，佇列中所有訊息都有相同的過期時間。
B: 對訊息進行單獨設定，每條訊息TTL可以不同。

如果同時使用，則訊息的過期時間以兩者之間TTL較小的那個數值為準。訊息在佇列的生存時間一旦超過設定的TTL值，就成為dead letter

特性二：Dead Letter Exchanges（DLX）

RabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key（可選）兩個引數，如果佇列內出現了dead letter，則按照這兩個引數重新路由轉發到指定的佇列。
x-dead-letter-exchange：出現dead letter之後將dead letter重新發送到指定exchange
x-dead-letter-routing-key：出現dead letter之後將dead letter重新按照指定的routing-key傳送
隊列出現dead letter的情況有：
訊息或者佇列的TTL過期
佇列達到最大長度

訊息被消費端拒絕（basic.reject or basic.nack）並且requeue=false

SpringBoot整合RabbitMQ

在 pom.xml 中新增 spring-boot-starter-amqp的依賴

<dependencies>
		<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.projectlombok</groupId>
			<artifactId>lombok</artifactId>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
	</dependencies>
 

在 application.yml檔案中配置rabbitmq相關內容

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

具體編碼實現

1.配置佇列

package com.lzc.rabbitmq.config;

import lombok.extern.slf4j.Slf4j;
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
@Slf4j
public class DelayRabbitConfig {


    /**
     * 延遲佇列 TTL 名稱
     */
    private static final String ORDER_DELAY_QUEUE = "user.order.delay.queue";
    /**
     * DLX，dead letter傳送到的 exchange
     * 延時訊息就是傳送到該交換機的
     */
    public static final String ORDER_DELAY_EXCHANGE = "user.order.delay.exchange";
    /**
     * routing key 名稱
     * 具體訊息傳送在該 routingKey 的
     */
    public static final String ORDER_DELAY_ROUTING_KEY = "order_delay";

    public static final String ORDER_QUEUE_NAME = "user.order.queue";
    public static final String ORDER_EXCHANGE_NAME = "user.order.exchange";
    public static final String ORDER_ROUTING_KEY = "order";

    /**
     * 延遲佇列配置
     * <p>
     * 1、params.put("x-message-ttl", 5 * 1000);
     * 第一種方式是直接設定 Queue 延遲時間 但如果直接給佇列設定過期時間,這種做法不是很靈活,（當然二者是相容的,預設是時間小的優先）
     * 2、rabbitTemplate.convertAndSend(book, message -> {
     * message.getMessageProperties().setExpiration(2 * 1000 + "");
     * return message;
     * });
     * 第二種就是每次傳送訊息動態設定延遲時間,這樣我們可以靈活控制
     **/
    @Bean
    public Queue delayOrderQueue() {
        Map<String, Object> params = new HashMap<>();
        // x-dead-letter-exchange 聲明瞭佇列裡的死信轉發到的DLX名稱，
        params.put("x-dead-letter-exchange", ORDER_EXCHANGE_NAME);
        // x-dead-letter-routing-key 聲明瞭這些死信在轉發時攜帶的 routing-key 名稱。
        params.put("x-dead-letter-routing-key", ORDER_ROUTING_KEY);
        return new Queue(ORDER_DELAY_QUEUE, true, false, false, params);
    }
    /**
     * 需要將一個佇列繫結到交換機上，要求該訊息與一個特定的路由鍵完全匹配。
     * 這是一個完整的匹配。如果一個佇列繫結到該交換機上要求路由鍵 “dog”，則只有被標記為“dog”的訊息才被轉發，
     * 不會轉發dog.puppy，也不會轉發dog.guard，只會轉發dog。
     * @return DirectExchange
     */
    @Bean
    public DirectExchange orderDelayExchange() {
        return new DirectExchange(ORDER_DELAY_EXCHANGE);
    }
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(delayOrderQueue()).to(orderDelayExchange()).with(ORDER_DELAY_ROUTING_KEY);
    }

    @Bean
    public Queue orderQueue() {
        return new Queue(ORDER_QUEUE_NAME, true);
    }
    /**
     * 將路由鍵和某模式進行匹配。此時佇列需要繫結要一個模式上。
     * 符號“#”匹配一個或多個詞，符號“*”匹配不多不少一個詞。因此“audit.#”能夠匹配到“audit.irs.corporate”，但是“audit.*” 只會匹配到“audit.irs”。
     **/
    @Bean
    public TopicExchange orderTopicExchange() {
        return new TopicExchange(ORDER_EXCHANGE_NAME);
    }

    @Bean
    public Binding orderBinding() {
        // TODO 如果要讓延遲佇列之間有關聯,這裡的 routingKey 和 繫結的交換機很關鍵
        return BindingBuilder.bind(orderQueue()).to(orderTopicExchange()).with(ORDER_ROUTING_KEY);
    }

}

2.建立一個Order實體類

package com.lzc.rabbitmq.dataobject;

import lombok.Data;

import java.io.Serializable;

@Data
public class Order implements Serializable {


    private static final long serialVersionUID = -2221214252163879885L;
    
    private String orderId; // 訂單id

    private Integer orderStatus; // 訂單狀態 0：未支付，1：已支付，2：訂單已取消

    private String orderName; // 訂單名字
}

3.接收者

package com.lzc.rabbitmq.config;

import com.lzc.rabbitmq.dataobject.Order;
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.util.Date;

@Component
@Slf4j
public class DelayReceiver {

    @RabbitListener(queues = {DelayRabbitConfig.ORDER_QUEUE_NAME})
    public void orderDelayQueue(Order order, Message message, Channel channel) {
        log.info("###########################################");
        log.info("【orderDelayQueue 監聽的訊息】 - 【消費時間】 - [{}]- 【訂單內容】 - [{}]",  new Date(), order.toString());
        if(order.getOrderStatus() == 0) {
            order.setOrderStatus(2);
            log.info("【該訂單未支付，取消訂單】" + order.toString());
        } else if(order.getOrderStatus() == 1) {
            log.info("【該訂單已完成支付】");
        } else if(order.getOrderStatus() == 2) {
            log.info("【該訂單已取消】");
        }
        log.info("###########################################");
    }
}

4.傳送者

package com.lzc.rabbitmq.config;

import com.lzc.rabbitmq.dataobject.Order;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;

@Component
@Slf4j
public class DelaySender {

    @Autowired
    private AmqpTemplate amqpTemplate;

    public void sendDelay(Order order) {
        log.info("【訂單生成時間】" + new Date().toString() +"【1分鐘後檢查訂單是否已經支付】" + order.toString() );
        this.amqpTemplate.convertAndSend(DelayRabbitConfig.ORDER_DELAY_EXCHANGE, DelayRabbitConfig.ORDER_DELAY_ROUTING_KEY, order, message -> {
            // 如果配置了 params.put("x-message-ttl", 5 * 1000); 那麼這一句也可以省略,具體根據業務需要是宣告 Queue 的時候就指定好延遲時間還是在傳送自己控制時間
            message.getMessageProperties().setExpiration(1 * 1000 * 60 + "");
            return message;
        });
    }
}

5.測試，訪問http://localhost:8080/sendDelay，檢視日誌輸出

package com.lzc.rabbitmq.controller;

import com.lzc.rabbitmq.config.DelaySender;
import com.lzc.rabbitmq.config.Sender;
import com.lzc.rabbitmq.dataobject.Order;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.DelayQueue;

@RestController
public class TestController {

    @Autowired
    private DelaySender delaySender;

    @GetMapping("/sendDelay")
    public Object sendDelay() {
        Order order1 = new Order();
        order1.setOrderStatus(0);
        order1.setOrderId("123456");
        order1.setOrderName("小米6");

        Order order2 = new Order();
        order2.setOrderStatus(1);
        order2.setOrderId("456789");
        order2.setOrderName("小米8");

        delaySender.sendDelay(order1);
        delaySender.sendDelay(order2);
        return "ok";
    }
}

6.日誌輸出

【訂單生成時間】Mon Jun 18 11:55:36 CST 2018【1分鐘後檢查訂單是否已經支付】Order(orderId=123456, orderStatus=0, orderName=小米6)
【訂單生成時間】Mon Jun 18 11:55:36 CST 2018【1分鐘後檢查訂單是否已經支付】Order(orderId=456789, orderStatus=1, orderName=小米8)
 ###########################################
【orderDelayQueue 監聽的訊息】 - 【消費時間】 - [Mon Jun 18 11:56:36 CST 2018]- 【訂單內容】 - [Order(orderId=123456, orderStatus=0, orderName=小米6)]
【該訂單未支付，取消訂單】Order(orderId=123456, orderStatus=2, orderName=小米6)
 ###########################################
 ###########################################
【orderDelayQueue 監聽的訊息】 - 【消費時間】 - [Mon Jun 18 11:56:36 CST 2018]- 【訂單內容】 - [Order(orderId=456789, orderStatus=1, orderName=小米8)]
【該訂單已完成支付】
 ###########################################