快捷導(dǎo)航

RabbitMQ通過延遲插件實現(xiàn)延遲消息

更新時間：2024年10月14日 10:30:11 作者：王哲曉 tiger

在RabbitMQ中,使用延遲消息插件比死信隊列更優(yōu)化的實現(xiàn)消息的延遲發(fā)送,本文介紹了延遲插件的下載、安裝、以及如何通過設(shè)置消息頭x-delay實現(xiàn)消息的延遲投遞,特別指出,使用延遲消息可能會損耗性能,適合短時間的延遲場景

一、引言

上一章我們講到通過死信隊列組合消息過期時間來實現(xiàn)延遲消息，但相對而言這并不是比較好的方式。它的代碼實現(xiàn)相對來說比較繁瑣，而且關(guān)鍵是RabbitMQ提供死信隊列的初衷并不是讓我們用來發(fā)送延遲消息的，而是為了作為兜底方案，來接收沒有消費的死信的，以便于定位問題。那么本章節(jié)我們就開始講解通過延遲消息插件來實現(xiàn)延遲消息。

延遲消息插件可以將普通交換機改造為支持延遲消息功能的交換機，當消息投遞到交換機后可以暫存一定時間，到期后再投遞到隊列。

這個插件可以將普通交換機改造為支持延遲消息功能的交換機，當消息投遞到交換機后可以暫存一定時間，到期后再投遞到隊列。

發(fā)送消息時需要通過消息頭x-delay來設(shè)置過期時間：

二、延遲插件安裝

在Mac上安裝RabbitMQ延遲消息插件時，確保你已經(jīng)安裝了RabbitMQ并且它正在運行，操作步驟如下（Linux安裝步驟和方法類似，此處不做贅述）：

2.1. 下載插件

從RabbitMQ官方GitHub倉庫或者通過以下命令直接下載，我這里是最新版本：

wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v4.0.2/rabbitmq_delayed_message_exchange-4.0.2.ez

2.2. 安裝插件

將下載的插件拷貝到RabbitMQ的plugins目錄，使用RabbitMQ插件管理命令安裝下載的插件：

sudo rabbitmq-plugins enable rabbitmq_delayed_message_exchange

如果你的RabbitMQ是以Docker容器的方式運行的，你可以將插件下載步驟和安裝步驟合并為一個Docker命令，例如：

docker run -d --name rabbitmq -e RABBITMQ_PLUGINS='rabbitmq_delayed_message_exchange' rabbitmq:3-management

這個命令會啟動一個帶有RabbitMQ管理插件的容器，并且會自動安裝延遲消息插件。

2.3. 確認插件是否生效

安裝成功后，RabbitMQ的瀏覽器界面上，exchange交換機在創(chuàng)建時，Type多了x-delayed-message選項。

請注意，具體的RabbitMQ版本和插件版本可能會更新，因此請根據(jù)實際情況下載相應(yīng)的版本。

三、核心代碼

package com.example.publisher;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;
import java.nio.charset.StandardCharsets;
/**
 * 生產(chǎn)者
 */
@Slf4j
@SpringBootTest
class PublisherApplicationTests {
    @Resource
    private RabbitTemplate rabbitTemplate;
    @Test
    void test() {
        String content = "生活不易，所以保持足夠的努力，對自己要有信心，積極地去面對工作生活的挑戰(zhàn)！";
        Message message = MessageBuilder.withBody(content.getBytes(StandardCharsets.UTF_8)).build();
        message.getMessageProperties().setDelayLong(10000L);
        rabbitTemplate.convertAndSend("delay.direct",
                "delay", message);
    }
}

package com.example.consumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
/**
 * 消費者
 */
@Slf4j
@Component
public class SimpleListener {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "delay.queue", durable = "true"),
            exchange = @Exchange(name = "delay.direct", delayed = "true"),
            key = "delay"
    ))
    public void listener1(Message message) throws Exception {
        String msg = new String(message.getBody(), StandardCharsets.UTF_8); ;
        System.out.println("延遲消息：人生是個不斷攀登的過程【" + msg + "】");
    }
}

四、運行效果

我們可以看到消息在延時10秒后消費

五、總結(jié)

雖然延遲插件讓我們在消息延遲發(fā)送的代碼實現(xiàn)上已經(jīng)非常簡潔，但是在使用延遲消息中還有一個問題就是延遲消息比較損耗性能，我們在RabbitMQ上使用延遲消息時，它的內(nèi)部就會維護一個時鐘，每當我們定義一個新的延遲消息它就會創(chuàng)建一個新的時鐘。如果一個任務(wù)的延遲時間特別長，比如一小時甚至一天，將非常耗性能。所以延遲消息比較適用于延遲的時間比較短的場景，比如10分鐘未支付取消訂單。

但是我們?nèi)绻苯釉O(shè)定10分鐘后發(fā)送延遲消息，也不是最優(yōu)解，我們將在下一章，對延遲消息做近一步的優(yōu)化！

到此這篇關(guān)于RabbitMQ延遲消息之延遲插件的文章就介紹到這了,更多相關(guān)RabbitMQ延遲插件內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: