RabbitMQ实现延迟消息

前提条件

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

  1. 下载插件:

    • 从RabbitMQ社区插件页面下载rabbitmq_delayed_message_exchange插件。
  2. 安装插件:

    • 将插件文件(.ez文件)放置在RabbitMQ插件目录中,通常为/usr/lib/rabbitmq/lib/rabbitmq_server-<version>/plugins
  3. 启用插件:

    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 -> { });
        }
    }
}

说明

  1. 生产者(Producer)代码:

    • 声明了一个延迟交换机,类型为x-delayed-message,并设置了x-delayed-type参数为direct
    • 声明了一个队列并将其绑定到延迟交换机。
    • 设置消息的x-delay头部,指定延迟时间。
    • 将消息发布到延迟交换机。
  2. 消费者(Consumer)代码:

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

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

相关推荐

  1. RabbitMQ实现延迟消息

    2024-07-09 17:04:07       10 阅读
  2. RabbitMQ如何实现延迟消息

    2024-07-09 17:04:07       22 阅读
  3. RabbitMQ延迟消息(通过死信交换机实现

    2024-07-09 17:04:07       10 阅读

最近更新

  1. MATLAB6:M文件和控制流

    2024-07-09 17:04:07       0 阅读
  2. early-stopping pytorch refs

    2024-07-09 17:04:07       0 阅读
  3. C++案例三:猜数字游戏

    2024-07-09 17:04:07       0 阅读
  4. 构建高可用应用的设计模式与实践

    2024-07-09 17:04:07       0 阅读
  5. MySQL简介

    2024-07-09 17:04:07       0 阅读

热门阅读

  1. 【数据挖掘】银行信用卡风险大数据分析与挖掘

    2024-07-09 17:04:07       7 阅读
  2. emacs 重新加载磁盘上的文件

    2024-07-09 17:04:07       7 阅读
  3. Linux学习笔记(一)

    2024-07-09 17:04:07       6 阅读
  4. 用WPF实现的窗体是怎么运行的?

    2024-07-09 17:04:07       8 阅读
  5. Mac OS M3 安装 Docker 并解决芯片不支持问题

    2024-07-09 17:04:07       4 阅读
  6. Fedora 41 移除 Python 2支持

    2024-07-09 17:04:07       7 阅读