RabbitMQ延時隊列詳解與Java代碼實現
作者：小威要向諸佬學習呀
RabbitMQ 延時隊列介紹
RabbitMQ 延時隊列是指消息在發(fā)送到隊列后，并不立即被消費者消費，而是等待一段時間后再被消費者消費。這種隊列通常用于實現定時任務，例如，訂單超時未支付系統(tǒng)取消訂單釋放所占庫存等。
RabbitMQ實現延時隊列的方法有多種，其中比較常見的是使用插件或者通過DLX（Dead Letter Exchange）機制實現。
使用插件實現延時隊列
RabbitMQ提供了rabbitmq_delayed_message_exchange插件，可以通過該插件實現延時隊列。該插件的原理是在消息發(fā)送時，將消息發(fā)送到一個特定的Exchange中，然后該Exchange會根據消息中的延時時間將消息轉發(fā)到指定的隊列中，從而實現延時隊列的功能。
使用該插件需要先安裝插件，然后創(chuàng)建一個Exchange，并將該Exchange的類型設置為x-delayed-message，然后將該Exchange與隊列綁定即可。
使用DLX機制實現延時隊列
消息的TTL就是消息的存活時間。RabbitMQ可以對隊列和消息分別設置TTL。而對隊列設置就是隊列沒有消費者連著的保留時間，也可以對每一個單獨的消息做單獨的 設置。超過了這個時間，我們認為這個消息就死了，稱之為死信。如果隊列設置了，消息也設置了，那么會取小的。所以一個消息如果被路由到不同的隊 列中，這個消息死亡的時間有可能不一樣（不同的隊列設置）。這里單講單個消息的TTL，因為它才是實現延遲任務的關鍵。可以通過設置消息的expiration字段或者x- message-ttl屬性來設置時間，兩者是一樣的效果。
DLX機制是RabbitMQ提供的一種消息轉發(fā)機制，它可以將無法被處理的消息轉發(fā)到指定的Exchange中，從而實現消息的延時處理。具體實現步驟如下：
- 創(chuàng)建一個普通的Exchange和Queue，并將它們綁定在一起。
- 創(chuàng)建一個DLX Exchange，并將普通Exchange綁定到該DLX Exchange上。
- 將Queue設置為具有TTL（Time To Live）屬性，并設置消息過期時間。
- 將Queue綁定到DLX Exchange上。
當消息過期后，會被發(fā)送到DLX Exchange中，然后再由DLX Exchange將消息轉發(fā)到指定的Exchange中，從而實現延時隊列的功能。
使用DLX機制實現延時隊列的優(yōu)點是不需要安裝額外的插件，但是需要對消息的過期時間進行精確控制，否則可能會出現消息過期時間不準確的情況。
Java語言設置延時隊列
下面是使用 Java 語言通過 RabbitMQ 設置延時隊列的步驟：
安裝插件
首先，需要安裝 rabbitmq_delayed_message_exchange 插件?？梢酝ㄟ^以下命令安裝：
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
創(chuàng)建延時交換機
延時隊列需要使用延時交換機?？梢允褂?x-delayed-message 類型創(chuàng)建一個延時交換機。以下是創(chuàng)建延時交換機的示例代碼：
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)建延時隊列時，需要將隊列綁定到延時交換機上，并設置隊列的 TTL（Time To Live）參數。以下是創(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");
在上述代碼中，將隊列綁定到延時交換機上，并設置了隊列的 TTL 參數為 5000 毫秒，即消息在發(fā)送到隊列后，如果在 5000 毫秒內沒有被消費者消費，則會被轉發(fā)到 delayed-exchange 交換機上，并發(fā)送到 delayed-queue 隊列中。
發(fā)送延時消息
發(fā)送延時消息時，需要設置消息的 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());
在上述代碼中，設置了消息的 expiration 屬性為 5000 毫秒，并將消息發(fā)送到 delayed-exchange 交換機上，路由鍵為 delayed-queue，消息內容為 “Hello, delayed queue!”。
消費延時消息
消費延時消息時，需要設置消費者的 QOS（Quality of Service）參數，以控制消費者的并發(fā)處理能力。以下是消費延時消息的示例代碼：
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);
});
在上述代碼中，設置了 QOS 參數為 1，即每次只處理一個消息。然后使用 basicConsume 方法消費 delayed-queue 隊列中的消息，并在消費完成后，使用 basicAck 方法確認消息已被消費。
通過上述步驟，就可以實現 RabbitMQ 延時隊列，用于實現定時任務等功能。
RabbitMQ延時隊列是一種常見的消息隊列應用場景，它可以在消息發(fā)送后指定一定的時間后才能被消費者消費，通常用于實現一些延時任務，例如訂單超時未支付自動取消等。
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)建一個延時隊列，設置x-dead-letter-exchange和x-dead-letter-routing-key參數
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ā)送消息到延時隊列中，設置expiration參數
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)建了一個延時隊列，設置了x-dead-letter-exchange和x-dead-letter-routing-key參數。然后，我們發(fā)送了一條消息到延時隊列中，設置了expiration參數，表示這條消息延時10秒后才能被消費。
注意，如果我們想要消費延時隊列中的消息，需要創(chuàng)建一個消費者，并監(jiān)聽這個隊列。當消息被消費時，需要發(fā)送ack確認消息已經被消費，否則消息會一直留在隊列中。
到此這篇關于RabbitMQ延時隊列詳解與Java代碼實現的文章就介紹到這了,更多相關Java實現RabbitMQ延時隊列內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！