RabbitMQ延時隊列詳解與Java代碼實(shí)現(xiàn)
作者：小威要向諸佬學(xué)習(xí)呀
RabbitMQ 延時隊列介紹
RabbitMQ 延時隊列是指消息在發(fā)送到隊列后，并不立即被消費(fèi)者消費(fèi)，而是等待一段時間后再被消費(fèi)者消費(fèi)。這種隊列通常用于實(shí)現(xiàn)定時任務(wù)，例如，訂單超時未支付系統(tǒng)取消訂單釋放所占庫存等。
RabbitMQ實(shí)現(xiàn)延時隊列的方法有多種，其中比較常見的是使用插件或者通過DLX（Dead Letter Exchange）機(jī)制實(shí)現(xiàn)。
使用插件實(shí)現(xiàn)延時隊列
RabbitMQ提供了rabbitmq_delayed_message_exchange插件，可以通過該插件實(shí)現(xiàn)延時隊列。該插件的原理是在消息發(fā)送時，將消息發(fā)送到一個特定的Exchange中，然后該Exchange會根據(jù)消息中的延時時間將消息轉(zhuǎn)發(fā)到指定的隊列中，從而實(shí)現(xiàn)延時隊列的功能。
使用該插件需要先安裝插件，然后創(chuàng)建一個Exchange，并將該Exchange的類型設(shè)置為x-delayed-message，然后將該Exchange與隊列綁定即可。
使用DLX機(jī)制實(shí)現(xiàn)延時隊列
消息的TTL就是消息的存活時間。RabbitMQ可以對隊列和消息分別設(shè)置TTL。而對隊列設(shè)置就是隊列沒有消費(fèi)者連著的保留時間，也可以對每一個單獨(dú)的消息做單獨(dú)的 設(shè)置。超過了這個時間，我們認(rèn)為這個消息就死了，稱之為死信。如果隊列設(shè)置了，消息也設(shè)置了，那么會取小的。所以一個消息如果被路由到不同的隊 列中，這個消息死亡的時間有可能不一樣（不同的隊列設(shè)置）。這里單講單個消息的TTL，因?yàn)樗攀菍?shí)現(xiàn)延遲任務(wù)的關(guān)鍵。可以通過設(shè)置消息的expiration字段或者x- message-ttl屬性來設(shè)置時間，兩者是一樣的效果。
DLX機(jī)制是RabbitMQ提供的一種消息轉(zhuǎn)發(fā)機(jī)制，它可以將無法被處理的消息轉(zhuǎn)發(fā)到指定的Exchange中，從而實(shí)現(xiàn)消息的延時處理。具體實(shí)現(xiàn)步驟如下：
- 創(chuàng)建一個普通的Exchange和Queue，并將它們綁定在一起。
- 創(chuàng)建一個DLX Exchange，并將普通Exchange綁定到該DLX Exchange上。
- 將Queue設(shè)置為具有TTL（Time To Live）屬性，并設(shè)置消息過期時間。
- 將Queue綁定到DLX Exchange上。
當(dāng)消息過期后，會被發(fā)送到DLX Exchange中，然后再由DLX Exchange將消息轉(zhuǎn)發(fā)到指定的Exchange中，從而實(shí)現(xiàn)延時隊列的功能。
使用DLX機(jī)制實(shí)現(xiàn)延時隊列的優(yōu)點(diǎn)是不需要安裝額外的插件，但是需要對消息的過期時間進(jìn)行精確控制，否則可能會出現(xiàn)消息過期時間不準(zhǔn)確的情況。
Java語言設(shè)置延時隊列
下面是使用 Java 語言通過 RabbitMQ 設(shè)置延時隊列的步驟：
安裝插件
首先，需要安裝 rabbitmq_delayed_message_exchange 插件?？梢酝ㄟ^以下命令安裝：
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
創(chuàng)建延時交換機(jī)
延時隊列需要使用延時交換機(jī)?？梢允褂?x-delayed-message 類型創(chuàng)建一個延時交換機(jī)。以下是創(chuàng)建延時交換機(jī)的示例代碼：
Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");
channel.exchangeDeclare("delayed-exchange", "x-delayed-message", true, false, args);
創(chuàng)建延時隊列
創(chuàng)建延時隊列時，需要將隊列綁定到延時交換機(jī)上，并設(shè)置隊列的 TTL（Time To Live）參數(shù)。以下是創(chuàng)建延時隊列的示例代碼：
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "delayed-exchange");
args.put("x-dead-letter-routing-key", "delayed-queue");
args.put("x-message-ttl", 5000);
channel.queueDeclare("delayed-queue", true, false, false, args);
channel.queueBind("delayed-queue", "delayed-exchange", "delayed-queue");
在上述代碼中，將隊列綁定到延時交換機(jī)上，并設(shè)置了隊列的 TTL 參數(shù)為 5000 毫秒，即消息在發(fā)送到隊列后，如果在 5000 毫秒內(nèi)沒有被消費(fèi)者消費(fèi)，則會被轉(zhuǎn)發(fā)到 delayed-exchange 交換機(jī)上，并發(fā)送到 delayed-queue 隊列中。
發(fā)送延時消息
發(fā)送延時消息時，需要設(shè)置消息的 expiration 屬性，該屬性表示消息的過期時間。以下是發(fā)送延時消息的示例代碼：
Map<String, Object> headers = new HashMap<>();
headers.put("x-delay", 5000);
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
.headers(headers)
.expiration("5000")
.build();
channel.basicPublish("delayed-exchange", "delayed-queue", properties, "Hello, delayed queue!".getBytes());
在上述代碼中，設(shè)置了消息的 expiration 屬性為 5000 毫秒，并將消息發(fā)送到 delayed-exchange 交換機(jī)上，路由鍵為 delayed-queue，消息內(nèi)容為 “Hello, delayed queue!”。
消費(fèi)延時消息
消費(fèi)延時消息時，需要設(shè)置消費(fèi)者的 QOS（Quality of Service）參數(shù)，以控制消費(fèi)者的并發(fā)處理能力。以下是消費(fèi)延時消息的示例代碼：
channel.basicQos(1);
channel.basicConsume("delayed-queue", false, (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
System.out.println("Received message: " + message);
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
});
在上述代碼中，設(shè)置了 QOS 參數(shù)為 1，即每次只處理一個消息。然后使用 basicConsume 方法消費(fèi) delayed-queue 隊列中的消息，并在消費(fèi)完成后，使用 basicAck 方法確認(rèn)消息已被消費(fèi)。
通過上述步驟，就可以實(shí)現(xiàn) RabbitMQ 延時隊列，用于實(shí)現(xiàn)定時任務(wù)等功能。
RabbitMQ延時隊列是一種常見的消息隊列應(yīng)用場景，它可以在消息發(fā)送后指定一定的時間后才能被消費(fèi)者消費(fèi)，通常用于實(shí)現(xiàn)一些延時任務(wù)，例如訂單超時未支付自動取消等。
RabbitMQ延時隊列具體代碼
下面是具體代碼（附注釋）：
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;
public class DelayedQueueExample {
private static final String EXCHANGE_NAME = "delayed_exchange";
private static final String QUEUE_NAME = "delayed_queue";
private static final String ROUTING_KEY = "delayed_routing_key";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
/*
Exchange.DeclareOk exchangeDeclare(String exchange,
String type,
boolean durable,
boolean autoDelete,
boolean internal,
Map<String, Object> arguments) throws IOException;
*/
// 創(chuàng)建一個支持延時隊列的Exchange
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-delayed-type", "direct");
channel.exchangeDeclare(EXCHANGE_NAME, "x-delayed-message", true, false, arguments);
// 創(chuàng)建一個延時隊列，設(shè)置x-dead-letter-exchange和x-dead-letter-routing-key參數(shù)
Map<String, Object> queueArguments = new HashMap<>();
queueArguments.put("x-dead-letter-exchange", "");
queueArguments.put("x-dead-letter-routing-key", QUEUE_NAME);
queueArguments.put("x-message-ttl", 5000);
channel.queueDeclare(QUEUE_NAME, true, false, false, queueArguments);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
// 發(fā)送消息到延時隊列中，設(shè)置expiration參數(shù)
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
.expiration("10000")
.build();
String message = "Hello, delayed queue!";
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, properties, message.getBytes());
System.out.println("Sent message to delayed queue: " + message);
channel.close();
connection.close();
}
}
在上面的代碼中，我們創(chuàng)建了一個支持延時隊列的Exchange，并創(chuàng)建了一個延時隊列，設(shè)置了x-dead-letter-exchange和x-dead-letter-routing-key參數(shù)。然后，我們發(fā)送了一條消息到延時隊列中，設(shè)置了expiration參數(shù)，表示這條消息延時10秒后才能被消費(fèi)。
注意，如果我們想要消費(fèi)延時隊列中的消息，需要創(chuàng)建一個消費(fèi)者，并監(jiān)聽這個隊列。當(dāng)消息被消費(fèi)時，需要發(fā)送ack確認(rèn)消息已經(jīng)被消費(fèi)，否則消息會一直留在隊列中。
到此這篇關(guān)于RabbitMQ延時隊列詳解與Java代碼實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)Java實(shí)現(xiàn)RabbitMQ延時隊列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！