RabbitMQ实现延迟消息

前提条件

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

1. 下载插件：

   • 从RabbitMQ社区插件页面下载rabbitmq_delayed_message_exchange插件。

2. 安装插件：

   • 将插件文件（.ez文件）放置在RabbitMQ插件目录中，通常为/usr/lib/rabbitmq/lib/rabbitmq_server-/plugins。

3. 启用插件：

   rabbitmq-plugins enable rabbitmq_delayed_message_exchange
   

Maven依赖

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

<dependency>
    <groupId>com.rabbitmqgroupId>
    <artifactId>amqp-clientartifactId>
    <version>5.13.0version>
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 -> { });
        }
    }
}

说明

1. 生产者（Producer）代码：

   • 声明了一个延迟交换机，类型为x-delayed-message，并设置了x-delayed-type参数为direct。
   • 声明了一个队列并将其绑定到延迟交换机。
   • 设置消息的x-delay头部，指定延迟时间。
   • 将消息发布到延迟交换机。

2. 消费者（Consumer）代码：

   • 声明了一个队列。
   • 使用DeliverCallback监听队列并处理收到的消息。

通过上述步骤和代码示例，你可以在Java中实现RabbitMQ的延迟消息功能。