RabbitMQ实现延迟消息

前提条件

确保RabbitMQ已安装并启用了RabbitMQ Delayed Message插件。如果尚未启用，可以按照以下步骤操作：

下载插件：
- 从RabbitMQ社区插件页面下载rabbitmq_delayed_message_exchange插件。
安装插件：
- 将插件文件（.ez文件）放置在RabbitMQ插件目录中，通常为/usr/lib/rabbitmq/lib/rabbitmq_server-<version>/plugins。

启用插件：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

Maven依赖

在你的Maven项目的pom.xml中添加RabbitMQ客户端库的依赖：

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.13.0</version>
</dependency>

生产者（Producer）代码

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.util.HashMap;
import java.util.Map;public class Producer {private final static String EXCHANGE_NAME = "delayed_exchange";private final static String QUEUE_NAME = "delayed_queue";private final static String ROUTING_KEY = "delayed_key";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明延迟交换机Map<String, Object> args = new HashMap<>();args.put("x-delayed-type", BuiltinExchangeType.DIRECT.getType());channel.exchangeDeclare(EXCHANGE_NAME, "x-delayed-message", true, false, args);// 声明队列channel.queueDeclare(QUEUE_NAME, true, false, false, null);// 绑定队列到延迟交换机channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);String message = "Hello World with delay!";int delay = 5000; // 延迟时间，以毫秒为单位// 设置消息属性，包括延迟时间Map<String, Object> headers = new HashMap<>();headers.put("x-delay", delay);AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder().headers(headers).deliveryMode(2); // 使消息持久化// 发布消息channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, props.build(), message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "' with delay " + delay + " ms");}}
}

消费者（Consumer）代码

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;public class Consumer {private final static String QUEUE_NAME = "delayed_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明队列channel.queueDeclare(QUEUE_NAME, true, false, false, null);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");};// 监听队列并处理消息channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });}}
}

说明

生产者（Producer）代码：
- 声明了一个延迟交换机，类型为x-delayed-message，并设置了x-delayed-type参数为direct。
- 声明了一个队列并将其绑定到延迟交换机。
- 设置消息的x-delay头部，指定延迟时间。
- 将消息发布到延迟交换机。
消费者（Consumer）代码：
- 声明了一个队列。
- 使用DeliverCallback监听队列并处理收到的消息。