RabbitMQ 中的消息長(zhǎng)期未被消費(fèi)會(huì)過(guò)期嗎？用過(guò) RabbitMQ 的小伙伴可能都有這樣的疑問(wèn)，今天松哥就來(lái)和大家捋一捋這個(gè)問(wèn)題。

1. 默認(rèn)情況

首先我們來(lái)看看默認(rèn)情況。

默認(rèn)情況下，消息是不會(huì)過(guò)期的，也就是我們平日里在消息發(fā)送時(shí)，如果不設(shè)置任何消息過(guò)期的相關(guān)參數(shù)，那么消息是不會(huì)過(guò)期的，即使消息沒(méi)被消費(fèi)掉，也會(huì)一直存儲(chǔ)在隊(duì)列中。

這種情況具體代碼就不用我再演示了吧，松哥之前的文章凡是涉及到 RabbitMQ 的，基本上都是這樣的。

2. TTL

TTL（Time-To-Live），消息存活的時(shí)間，即消息的有效期。如果我們希望消息能夠有一個(gè)存活時(shí)間，那么我們可以通過(guò)設(shè)置 TTL 來(lái)實(shí)現(xiàn)這一需求。如果消息的存活時(shí)間超過(guò)了 TTL 并且還沒(méi)有被消息，此時(shí)消息就會(huì)變成死信，關(guān)于死信以及死信隊(duì)列，松哥后面再和大家介紹。

TTL 的設(shè)置有兩種不同的方式：

• 在聲明隊(duì)列的時(shí)候，我們可以在隊(duì)列屬性中設(shè)置消息的有效期，這樣所有進(jìn)入該隊(duì)列的消息都會(huì)有一個(gè)相同的有效期。
• 在發(fā)送消息的時(shí)候設(shè)置消息的有效期，這樣不同的消息就具有不同的有效期。

那如果兩個(gè)都設(shè)置了呢？

以時(shí)間短的為準(zhǔn)。

當(dāng)我們?cè)O(shè)置了消息有效期后，消息過(guò)期了就會(huì)被從隊(duì)列中刪除了（進(jìn)入到死信隊(duì)列，后文一樣，不再標(biāo)注），但是兩種方式對(duì)應(yīng)的刪除時(shí)機(jī)有一些差異：

對(duì)于第一種方式，當(dāng)消息隊(duì)列設(shè)置過(guò)期時(shí)間的時(shí)候，那么消息過(guò)期了就會(huì)被刪除，因?yàn)橄⑦M(jìn)入 RabbitMQ 后是存在一個(gè)消息隊(duì)列中，隊(duì)列的頭部是最早要過(guò)期的消息，所以 RabbitMQ 只需要一個(gè)定時(shí)任務(wù)，從頭部開(kāi)始掃描是否有過(guò)期消息，有的話就直接刪除。對(duì)于第二種方式，當(dāng)消息過(guò)期后并不會(huì)立馬被刪除，而是當(dāng)消息要投遞給消費(fèi)者的時(shí)候才會(huì)去刪除，因?yàn)榈诙N方式，每條消息的過(guò)期時(shí)間都不一樣，想要知道哪條消息過(guò)期，必須要遍歷隊(duì)列中的所有消息才能實(shí)現(xiàn)，當(dāng)消息比較多時(shí)這樣就比較耗費(fèi)性能，因此對(duì)于第二種方式，當(dāng)消息要投遞給消費(fèi)者的時(shí)候才去刪除。

介紹完 TTL 之后，接下來(lái)我們來(lái)看看具體用法。

接下來(lái)所有代碼松哥都以 Spring Boot 中封裝的 AMPQ 為例來(lái)講解。

2.1 單條消息過(guò)期

我們先來(lái)看單條消息的過(guò)期時(shí)間。

首先創(chuàng)建一個(gè) Spring Boot 項(xiàng)目，引入 Web 和 RabbitMQ 依賴(lài)，如下：

然后在 application.properties 中配置一下 RabbitMQ 的連接信息，如下：

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

接下來(lái)稍微配置一下消息隊(duì)列：

@Configuration
public class QueueConfig {

    public static final String JAVABOY_QUEUE_DEMO = "javaboy_queue_demo";
    public static final String JAVABOY_EXCHANGE_DEMO = "javaboy_exchange_demo";
    public static final String HELLO_ROUTING_KEY = "hello_routing_key";

    @Bean
    Queue queue() {
        return new Queue(JAVABOY_QUEUE_DEMO, true, false, false);
    }

    @Bean
    DirectExchange directExchange() {
        return new DirectExchange(JAVABOY_EXCHANGE_DEMO, true, false);
    }

    @Bean
    Binding binding() {
        return BindingBuilder.bind(queue())
                .to(directExchange())
                .with(HELLO_ROUTING_KEY);
    }
}

這個(gè)配置類(lèi)主要干了三件事：配置消息隊(duì)列、配置交換機(jī)以及將兩者綁定在一起。

• 首先配置一個(gè)消息隊(duì)列，new 一個(gè) Queue：第一個(gè)參數(shù)是消息隊(duì)列的名字；第二個(gè)參數(shù)表示消息是否持久化；第三個(gè)參數(shù)表示消息隊(duì)列是否排他，一般我們都是設(shè)置為 false，即不排他；第四個(gè)參數(shù)表示如果該隊(duì)列沒(méi)有任何訂閱的消費(fèi)者的話，該隊(duì)列會(huì)被自動(dòng)刪除，一般適用于臨時(shí)隊(duì)列。
• 配置一個(gè) DirectExchange 交換機(jī)。
• 將交換機(jī)和隊(duì)列綁定到一起。

這段配置應(yīng)該很簡(jiǎn)單，沒(méi)啥好解釋的，有一個(gè)排他性，松哥這里稍微多說(shuō)兩句：

關(guān)于排他性，如果設(shè)置為 true，則該消息隊(duì)列只有創(chuàng)建它的 Connection 才能訪問(wèn)，其他的 Connection 都不能訪問(wèn)該消息隊(duì)列，如果試圖在不同的連接中重新聲明或者訪問(wèn)排他性隊(duì)列，那么系統(tǒng)會(huì)報(bào)一個(gè)資源被鎖定的錯(cuò)誤。另一方面，對(duì)于排他性隊(duì)列而言，當(dāng)連接斷掉的時(shí)候，該消息隊(duì)列也會(huì)自動(dòng)刪除（無(wú)論該隊(duì)列是否被聲明為持久性隊(duì)列都會(huì)被刪除）。

接下來(lái)提供一個(gè)消息發(fā)送接口，如下：

@RestController
public class HelloController {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @GetMapping("/hello")
    public void hello() {
        Message message = MessageBuilder.withBody("hello javaboy".getBytes())
                .setExpiration("10000")
                .build();
        rabbitTemplate.convertAndSend(QueueConfig.JAVABOY_QUEUE_DEMO, message);
    }
}

在創(chuàng)建 Message 對(duì)象的時(shí)候我們可以設(shè)置消息的過(guò)期時(shí)間，這里設(shè)置消息的過(guò)期時(shí)間為 10 秒。

這就可以啦！

接下來(lái)我們啟動(dòng)項(xiàng)目，進(jìn)行消息發(fā)送測(cè)試。當(dāng)消息發(fā)送成功之后，由于沒(méi)有消費(fèi)者，所以這條消息并不會(huì)被消費(fèi)。打開(kāi) RabbitMQ 管理頁(yè)面，點(diǎn)擊到 Queues 選項(xiàng)卡，10s 之后，我們會(huì)發(fā)現(xiàn)消息已經(jīng)不見(jiàn)了：

很簡(jiǎn)單吧！

單條消息設(shè)置過(guò)期時(shí)間，就是在消息發(fā)送的時(shí)候設(shè)置一下消息有效期即可。

2.2 隊(duì)列消息過(guò)期

給隊(duì)列設(shè)置消息過(guò)期時(shí)間，方式如下：

@Bean
Queue queue() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-message-ttl", 10000);
    return new Queue(JAVABOY_QUEUE_DEMO, true, false, false, args);
}

設(shè)置完成后，我們修改消息的發(fā)送邏輯，如下：

@RestController
public class HelloController {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @GetMapping("/hello")
    public void hello() {
        Message message = MessageBuilder.withBody("hello javaboy".getBytes())
                .build();
        rabbitTemplate.convertAndSend(QueueConfig.JAVABOY_QUEUE_DEMO, message);
    }
}

可以看到，消息正常發(fā)送即可，不用設(shè)置消息過(guò)期時(shí)間。

OK，啟動(dòng)項(xiàng)目，發(fā)送一條消息進(jìn)行測(cè)試。查看 RabbitMQ 管理頁(yè)面，如下：

可以看到，消息隊(duì)列的 Features 屬性為 D 和 TTL，D 表示消息隊(duì)列中消息持久化，TTL 則表示消息會(huì)過(guò)期。

10s 之后刷新頁(yè)面，發(fā)現(xiàn)消息數(shù)量已經(jīng)恢復(fù)為 0。

這就是給消息隊(duì)列設(shè)置消息過(guò)期時(shí)間，一旦設(shè)置了，所有進(jìn)入到該隊(duì)列的消息都有一個(gè)過(guò)期時(shí)間了。

2.3 特殊情況

還有一種特殊情況，就是將消息的過(guò)期時(shí)間 TTL 設(shè)置為 0，這表示如果消息不能立馬消費(fèi)則會(huì)被立即丟掉，這個(gè)特性可以部分替代 RabbitMQ3.0 以前支持的 immediate 參數(shù)，之所以所部分代替，是因?yàn)?immediate 參數(shù)在投遞失敗會(huì)有 basic.return 方法將消息體返回（這個(gè)功能可以利用死信隊(duì)列來(lái)實(shí)現(xiàn)）。

具體代碼松哥就不演示了，這個(gè)應(yīng)該比較容易。

3. 死信隊(duì)列

有小伙伴不禁要問(wèn)，被刪除的消息去哪了？真的被刪除了嗎？非也非也！這就涉及到死信隊(duì)列了，接下來(lái)我們來(lái)看看死信隊(duì)列。

3.1 死信交換機(jī)

死信交換機(jī)，Dead-Letter-Exchange 即 DLX。

死信交換機(jī)用來(lái)接收死信消息（Dead Message）的，那什么是死信消息呢？一般消息變成死信消息有如下幾種情況：

• 消息被拒絕(Basic.Reject/Basic.Nack) ，井且設(shè)置requeue 參數(shù)為false
• 消息過(guò)期
• 隊(duì)列達(dá)到最大長(zhǎng)度

當(dāng)消息在一個(gè)隊(duì)列中變成了死信消息后，此時(shí)就會(huì)被發(fā)送到 DLX，綁定 DLX 的消息隊(duì)列則稱(chēng)為死信隊(duì)列。

DLX 本質(zhì)上也是一個(gè)普普通通的交換機(jī)，我們可以為任意隊(duì)列指定 DLX，當(dāng)該隊(duì)列中存在死信時(shí)，RabbitMQ 就會(huì)自動(dòng)的將這個(gè)死信發(fā)布到 DLX 上去，進(jìn)而被路由到另一個(gè)綁定了 DLX 的隊(duì)列上（即死信隊(duì)列）。

3.2 死信隊(duì)列

這個(gè)好理解，綁定了死信交換機(jī)的隊(duì)列就是死信隊(duì)列。

3.3 實(shí)踐

我們來(lái)看一個(gè)簡(jiǎn)單的例子。

首先我們來(lái)創(chuàng)建一個(gè)死信交換機(jī)，接著創(chuàng)建一個(gè)死信隊(duì)列，再將死信交換機(jī)和死信隊(duì)列綁定到一起：

public static final String DLX_EXCHANGE_NAME = "dlx_exchange_name";
public static final String DLX_QUEUE_NAME = "dlx_queue_name";
public static final String DLX_ROUTING_KEY = "dlx_routing_key";

/**
 * 配置死信交換機(jī)
 *
 * @return
 */
@Bean
DirectExchange dlxDirectExchange() {
    return new DirectExchange(DLX_EXCHANGE_NAME, true, false);
}
/**
 * 配置死信隊(duì)列
 * @return
 */
@Bean
Queue dlxQueue() {
    return new Queue(DLX_QUEUE_NAME);
}
/**
 * 綁定死信隊(duì)列和死信交換機(jī)
 * @return
 */
@Bean
Binding dlxBinding() {
    return BindingBuilder.bind(dlxQueue())
            .to(dlxDirectExchange())
            .with(DLX_ROUTING_KEY);
}

這其實(shí)跟普通的交換機(jī)，普通的消息隊(duì)列沒(méi)啥兩樣。

接下來(lái)為消息隊(duì)列配置死信交換機(jī)，如下：

@Bean
Queue queue() {
    Map<String, Object> args = new HashMap<>();
    //設(shè)置消息過(guò)期時(shí)間
    args.put("x-message-ttl", 0);
    //設(shè)置死信交換機(jī)
    args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);
    //設(shè)置死信 routing_key
    args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);
    return new Queue(JAVABOY_QUEUE_DEMO, true, false, false, args);
}

就兩個(gè)參數(shù)：

• x-dead-letter-exchange：配置死信交換機(jī)。
• x-dead-letter-routing-key：配置死信 routing_key。

這就配置好了。

將來(lái)發(fā)送到這個(gè)消息隊(duì)列上的消息，如果發(fā)生了 nack、reject 或者過(guò)期等問(wèn)題，就會(huì)被發(fā)送到 DLX 上，進(jìn)而進(jìn)入到與 DLX 綁定的消息隊(duì)列上。

死信消息隊(duì)列的消費(fèi)和普通消息隊(duì)列的消費(fèi)并無(wú)二致：

@RabbitListener(queues = QueueConfig.DLX_QUEUE_NAME)
public void dlxHandle(String msg) {
    System.out.println("dlx msg = " + msg);
}

很容易吧～

4. 小結(jié)

好啦，今天就和小伙伴們聊一聊 RabbitMQ 中的消息過(guò)期問(wèn)題，感興趣的小伙伴可以去試試哦～

參考資料：

Java RabbitMQ的TTL和DLX全面精解

到此這篇關(guān)于Java RabbitMQ 中的消息長(zhǎng)期不消費(fèi)會(huì)過(guò)期嗎的文章就介紹到這了,更多相關(guān)Java RabbitMQ 消息過(guò)期內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論