RabbitMQ沒有延時隊列？學會這一招玩轉延時隊列

什麼是延時隊列

延時隊列：顧名思義，是一個用於作消息延時消費的隊列。可是它也是一個普通隊列，因此它具有普通隊列的特性，相比之下，延時的特性就是它最大的特色。所謂的延時就是將咱們須要的消息，延遲多久以後被消費。普通隊列是即時消費的，延時隊列是根據延時時間，多久以後才能消費的。java

延時隊列使用場景

訂單在十分鐘以內未支付則自動取消。
會員續費的定時推送
用戶註冊成功後，若是三天內沒有登錄則進行短信提醒。
預約會議後，須要在預約的時間點前十分鐘通知各個與會人員參加會議。
優惠券過時提醒

核心的應用內容基本都是基於須要設定過時時間的ui

RabbitMQ如何實現延時隊列

方式一、經過RabbitMQ的高級特性TTL和配合死信隊列
方式二、安裝rabbitmq_delayed_message_exchange插件

RabbitMQ中的高級特性TTL

TTL是什麼呢？TTL是RabbitMQ中一個消息或者隊列的屬性，代表一條消息或者該隊列中的全部消息的最大存活時間，單位是毫秒，爲何延時隊列要介紹它？TTL就是一種消息過時策略。給咱們的消息作過時處理，當消息在隊列中存活了指定時候以後，改隊列就會將這個消息直接丟棄。在RabbitMQ中並無直接實現好的延時隊列，咱們可使用TTL這種高級特性，而後配合死信隊列，便可實現延時隊列的功能。插件

那麼，如何設置這個TTL值呢？有兩種方式，第一種是在建立隊列的時候設置隊列的「x-message-ttl」屬性，以下：方式一：3d

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-message-ttl", 6000);
channel.queueDeclare(queueName, durable, exclusive, autoDelete, args);

使用這種方式，消息被設定TTL，一旦消息過時，就會被隊列丟棄code

方式二：blog

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
builder.expiration("6000");
AMQP.BasicProperties properties = builder.build();
channel.basicPublish(exchangeName, routingKey, mandatory, properties, "msg body".getBytes());

使用這種方式，消息即便過時，也不必定會被立刻丟棄，由於消息是否過時是在即將投遞到消費者以前斷定的，若是當前隊列有嚴重的消息積壓狀況，則已過時的消息也許還能存活較長時間。rabbitmq

另外，還須要注意的一點是，若是不設置TTL，表示消息永遠不會過時，若是將TTL設置爲0，則表示除非此時能夠直接投遞該消息到消費者，不然該消息將會被丟棄。隊列

RabbitMQ到底怎麼實現延時隊列

步驟一：建立一個正常的隊列，指定消息過時時間，而且指定消息過時後須要投遞的死信交換器和死信交換隊列。
步驟二：建立死信隊列和死信交換器

RabbitMQ實現延時隊列實例

package com.example.demo;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @author echo
 * @date 2021-01-14 14:35
 */
public class TopicDealProductTest {

    /**
     * 延時隊列交換機
     */
    private static final String DIRECT_EXCHANGE_DELAY = "dir_exchange_delay";
    /**
     * 死信隊列交換機
     */
    private static final String DIRECT_EXCHANGE_DEAD = "dir_exchange_dead";
    /**
     * 延時隊列
     */
    private static final String DIRECT_QUEUE_DELAY = "dir.queue.delay";
    /**
     * 死信隊列
     */
    private static final String DIRECT_QUEUE_DEAD = "dir.queue.dead";
    /**
     * 延時隊列ROUTING_KEY
     */
    private static final String DIRECT_DELAY_ROUTING_KEY = "delay.queue.routingKey";
    /**
     * 延時隊列ROUTING_KEY
     */
    private static final String DIRECT_DEAD_ROUTING_KEY = "dead.queue.routingKey";
    private static final String IP_ADDRESS = "192.168.230.131";
    private static final int PORT = 5672;


    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection = createConnection();
        // 建立一個頻道
        Channel channel = connection.createChannel();
        sendMsg(channel);
        Thread.sleep(10000);
        closeConnection(connection, channel);
    }

    private static void sendMsg(Channel channel) throws IOException {

        // 建立延時隊列和延時交換器
        channel.exchangeDeclare(DIRECT_EXCHANGE_DELAY, BuiltinExchangeType.DIRECT);
        Map<String, Object> map = new HashMap<>(16);
        // 在延時交換器上指定死信交換器
        map.put("x-dead-letter-exchange", DIRECT_EXCHANGE_DEAD);
        // 在延時交換器上指定死信隊列的routing-key
        map.put("x-dead-letter-routing-key", DIRECT_DEAD_ROUTING_KEY);
        // 設定延時隊列的延長時長 10s
        map.put("x-message-ttl", 10000);
        // 建立延時隊列
        channel.queueDeclare(DIRECT_QUEUE_DELAY, true, false, false, map);
        // 在延時交換器上綁定延時隊列
        channel.queueBind(DIRECT_QUEUE_DELAY, DIRECT_EXCHANGE_DELAY, DIRECT_DELAY_ROUTING_KEY);

        // 建立死信隊列和死信交換器
        channel.exchangeDeclare(DIRECT_EXCHANGE_DEAD, BuiltinExchangeType.TOPIC, true, false, null);
        // 建立死信隊列
        channel.queueDeclare(DIRECT_QUEUE_DEAD, true, false, false, null);
        // 在死信交換器上綁定死信隊列
        channel.queueBind(DIRECT_QUEUE_DEAD, DIRECT_EXCHANGE_DEAD, DIRECT_DEAD_ROUTING_KEY);

        channel.basicPublish(DIRECT_EXCHANGE_DELAY, DIRECT_DELAY_ROUTING_KEY, null, "hello world".getBytes());

    }

    private static void closeConnection(Connection connection, Channel channel) throws IOException, TimeoutException {
        // 關閉資源
        channel.close();
        connection.close();
    }

    private static Connection createConnection() throws IOException, TimeoutException {
        // 建立鏈接工廠
        ConnectionFactory factory = new ConnectionFactory();
        // 設置RabbitMQ的連接參數
        factory.setHost(IP_ADDRESS);
        factory.setPort(PORT);
        factory.setUsername("echo");
        factory.setPassword("123456");
        // 和RabbitMQ創建一個連接
        return factory.newConnection();
    }

}

到這裏，其實咱們不難發現，咱們無非是利用了TTL這個特性，讓消息在過時的時候丟棄到指定隊列，死信隊列其實也是一個普通隊列。圖片

執行以後，咱們來看看結果，在Exchange裏面，咱們建立了兩個交換器和兩個隊列，可是兩個隊列和交換器仍是有區別的，咱們來看圖片資源

咱們能夠看到兩個隊列的Features標誌是不同的

TTL: 消息在隊列中的過時時間
DLX: 該隊列綁定了死信交換器
DLK: 該隊列綁定的死信隊列的ROUTING_KEY

在咱們執行完成只有，咱們能夠看到，消息先被投遞到了delay，該隊列裏面的消息，到達過時時間以後就被投遞到了dead隊列中去了。

那麼咱們上面介紹了TTL和設置AMQP.BasicProperties，這兩種有必定的區別，前一個是設置隊列消息過時時間，後一個是設定每條消息的過時時間。那他們的區別在哪裏？

設置每條消息和設置TTL的區別

其實這兩種方式的區別就在於怎麼判斷該消息是否要被丟棄。TTL設定的隊列，只要消息到達過時時間，立馬就會將消息丟棄。若是是後者，可能咱們隊列裏面有不少的消息，而後每條消息的過時時間又不一致，這個時候，若是隊列出口處堵了不少沒有設定過時時間的消息又不被消費的時候，隊列後面的消息及時設定了過時時間也不會被丟棄，只有在設定了過時時間的消息到了隊列該消費的位置，纔會斷定

怎麼使用AMQP.BasicProperties？

package com.example.demo;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @author echo
 * @date 2021-01-14 14:35
 */
public class TopicDealProductTest {

    /**
     * 延時隊列交換機
     */
    private static final String DIRECT_EXCHANGE_DELAY = "dir_exchange_delay";
    /**
     * 死信隊列交換機
     */
    private static final String DIRECT_EXCHANGE_DEAD = "dir_exchange_dead";
    /**
     * 延時隊列
     */
    private static final String DIRECT_QUEUE_DELAY = "dir.queue.delay";
    /**
     * 死信隊列
     */
    private static final String DIRECT_QUEUE_DEAD = "dir.queue.dead";
    /**
     * 延時隊列ROUTING_KEY
     */
    private static final String DIRECT_DELAY_ROUTING_KEY = "delay.queue.routingKey";
    /**
     * 延時隊列ROUTING_KEY
     */
    private static final String DIRECT_DEAD_ROUTING_KEY = "dead.queue.routingKey";
    private static final String IP_ADDRESS = "192.168.230.131";
    private static final int PORT = 5672;


    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection = createConnection();
        // 建立一個頻道
        Channel channel = connection.createChannel();
        sendMsg(channel);
        Thread.sleep(10000);
        closeConnection(connection, channel);
    }

    private static void sendMsg(Channel channel) throws IOException {

        // 建立延時隊列和延時交換器
        channel.exchangeDeclare(DIRECT_EXCHANGE_DELAY, BuiltinExchangeType.DIRECT);
        Map<String, Object> map = new HashMap<>(16);
        // 在延時交換器上指定死信交換器
        map.put("x-dead-letter-exchange", DIRECT_EXCHANGE_DEAD);
        map.put("x-dead-letter-routing-key", DIRECT_DEAD_ROUTING_KEY);
        // 設定延時隊列的延長時長 10s
//        map.put("x-message-ttl", 10000);
        // 建立延時隊列
        channel.queueDeclare(DIRECT_QUEUE_DELAY, true, false, false, map);
        // 在延時交換器上綁定延時隊列
        channel.queueBind(DIRECT_QUEUE_DELAY, DIRECT_EXCHANGE_DELAY, DIRECT_DELAY_ROUTING_KEY);

        // 建立死信隊列和死信交換器
        channel.exchangeDeclare(DIRECT_EXCHANGE_DEAD, BuiltinExchangeType.TOPIC, true, false, null);
        // 建立死信隊列
        channel.queueDeclare(DIRECT_QUEUE_DEAD, true, false, false, null);
        // 在死信交換器上綁定死信隊列
        channel.queueBind(DIRECT_QUEUE_DEAD, DIRECT_EXCHANGE_DEAD, DIRECT_DEAD_ROUTING_KEY);

        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
        builder.expiration("10000");
        AMQP.BasicProperties properties = builder.build();
        channel.basicPublish(DIRECT_EXCHANGE_DELAY, DIRECT_DELAY_ROUTING_KEY, false, properties,  "hello world".getBytes());

    }

    private static void closeConnection(Connection connection, Channel channel) throws IOException, TimeoutException {
        // 關閉資源
        channel.close();
        connection.close();
    }

    private static Connection createConnection() throws IOException, TimeoutException {
        // 建立鏈接工廠
        ConnectionFactory factory = new ConnectionFactory();
        // 設置RabbitMQ的連接參數
        factory.setHost(IP_ADDRESS);
        factory.setPort(PORT);
        factory.setUsername("echo");
        factory.setPassword("123456");
        // 和RabbitMQ創建一個連接
        return factory.newConnection();
    }

}

咱們運行完成成以後，能夠看到和咱們以前那一種方式的效果是同樣的

兩種設定過時時間的方式其實區別就在於一個統一設定了過時時間，一個指定每條過時時間。可是並不影響咱們延時隊列的實現，那咱們怎麼選擇呢？

怎麼選擇TTL設定方式？

根據兩種方式的特性來選定使用場景纔是最合理的。咱們若是用來作延時隊列的，想將延時隊列的特性應用到實際場景的，而且對時時性要求比較高的，建議選擇第一種方式。

總結

延時隊列的實現並不難，關鍵是咱們要知道他的一個原理，瞭解RabbitMQ他的TTL和死信對了。掌握了它的這些特性以後，咱們就能夠很好的應用延時隊列。延時隊列在工做中對咱們的幫組也很是大，不過RabbiTMQ沒有原生延時隊列，咱們用這種方式實現了它並不意味着咱們必定要選擇它。其實還有不少的方式，好比Java中的DelayQueu、kafka的時間輪等。