RabbitMQ沒有延時佇列？學會這一招玩轉延時佇列-有解無憂

什么是延時佇列

延時佇列：顧名思義，是一個用于做訊息延時消費的佇列，但是它也是一個普通佇列，所以它具備普通佇列的特性，相比之下，延時的特性就是它最大的特點，所謂的延時就是將我們需要的訊息，延遲多久之后被消費，普通佇列是即時消費的，延時佇列是根據延時時間，多久之后才能消費的，

在這里插入圖片描述

延時佇列使用場景

訂單在十分鐘之內未支付則自動取消，
會員續費的定時推送
用戶注冊成功后，如果三天內沒有登陸則進行短信提醒，
預定會議后，需要在預定的時間點前十分鐘通知各個與會人員參加會議，
優惠券過期提醒

核心的應用內容基本都是基于需要設定過期時間的

RabbitMQ如何實作延時佇列

方式1、通過RabbitMQ的高級特性TTL和配合死信佇列
方式2、安裝rabbitmq_delayed_message_exchange插件

RabbitMQ中的高級特性TTL

TTL是什么呢？TTL是RabbitMQ中一個訊息或者佇列的屬性，表明一條訊息或者該佇列中的所有訊息的最大存活時間，單位是毫秒，為什么延時佇列要介紹它？TTL就是一種訊息過期策略，給我們的訊息做過期處理，當訊息在佇列中存活了指定時候之后，改佇列就會將這個訊息直接丟棄，在RabbitMQ中并沒有直接實作好的延時佇列，我們可以使用TTL這種高級特性，然后配合死信佇列，即可實作延時佇列的功能，

那么，如何設定這個TTL值呢？有兩種方式，第一種是在創建佇列的時候設定佇列的“x-message-ttl”屬性，如下：
方式一：

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

使用這種方式，訊息被設定TTL，一旦訊息過期，就會被佇列丟棄

方式二：

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());

使用這種方式，訊息即使過期，也不一定會被馬上丟棄，因為訊息是否過期是在即將投遞到消費者之前判定的，如果當前佇列有嚴重的訊息積壓情況，則已過期的訊息也許還能存活較長時間，

另外，還需要注意的一點是，如果不設定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的時間輪等，

轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/265959.html

標籤：其他

上一篇：Docker進階篇

下一篇：C語言中fun1()和fun2(void)的區別