使用Java实现RabbitMQ延时队列

更新时间：2023年06月20日 14:27:21 作者：土豆鱼_

RabbitMQ 延时队列是指消息在发送到队列后，并不立即被消费者消费，而是等待一段时间后再被消费者消费，本文为大家介绍了实现RabbitMQ延时队列的Java代码，希望对大家有所帮助

RabbitMQ 延时队列介绍

RabbitMQ 延时队列是指消息在发送到队列后，并不立即被消费者消费，而是等待一段时间后再被消费者消费。这种队列通常用于实现定时任务，例如，订单超时未支付系统取消订单释放所占库存等。

RabbitMQ实现延时队列的方法有多种，其中比较常见的是使用插件或者通过DLX（Dead Letter Exchange）机制实现。

1.使用插件实现延时队列

RabbitMQ提供了rabbitmq_delayed_message_exchange插件，可以通过该插件实现延时队列。该插件的原理是在消息发送时，将消息发送到一个特定的Exchange中，然后该Exchange会根据消息中的延时时间将消息转发到指定的队列中，从而实现延时队列的功能。

使用该插件需要先安装插件，然后创建一个Exchange，并将该Exchange的类型设置为x-delayed-message，然后将该Exchange与队列绑定即可。

2.使用DLX机制实现延时队列

消息的TTL就是消息的存活时间。RabbitMQ可以对队列和消息分别设置TTL。而对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。如果队列设置了，消息也设置了，那么会取小的。所以一个消息如果被路由到不同的队列中，这个消息死亡的时间有可能不一样（不同的队列设置）。这里单讲单个消息的TTL，因为它才是实现延迟任务的关键。可以通过设置消息的expiration字段或者x- message-ttl属性来设置时间，两者是一样的效果。

DLX机制是RabbitMQ提供的一种消息转发机制，它可以将无法被处理的消息转发到指定的Exchange中，从而实现消息的延时处理。具体实现步骤如下：

创建一个普通的Exchange和Queue，并将它们绑定在一起。
创建一个DLX Exchange，并将普通Exchange绑定到该DLX Exchange上。
将Queue设置为具有TTL（Time To Live）属性，并设置消息过期时间。
将Queue绑定到DLX Exchange上。

当消息过期后，会被发送到DLX Exchange中，然后再由DLX Exchange将消息转发到指定的Exchange中，从而实现延时队列的功能。

使用DLX机制实现延时队列的优点是不需要安装额外的插件，但是需要对消息的过期时间进行精确控制，否则可能会出现消息过期时间不准确的情况。

Java语言设置延时队列

下面是使用 Java 语言通过 RabbitMQ 设置延时队列的步骤：

1.安装插件

首先，需要安装 rabbitmq_delayed_message_exchange 插件。可以通过以下命令安装：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

2. 创建延时交换机

延时队列需要使用延时交换机。可以使用 x-delayed-message 类型创建一个延时交换机。以下是创建延时交换机的示例代码：

Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");
channel.exchangeDeclare("delayed-exchange", "x-delayed-message", true, false, args);

3.创建延时队列

创建延时队列时，需要将队列绑定到延时交换机上，并设置队列的 TTL（Time To Live）参数。以下是创建延时队列的示例代码：

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 毫秒，即消息在发送到队列后，如果在 5000 毫秒内没有被消费者消费，则会被转发到 delayed-exchange 交换机上，并发送到 delayed-queue 队列中。

4.发送延时消息

发送延时消息时，需要设置消息的 expiration 属性，该属性表示消息的过期时间。以下是发送延时消息的示例代码：

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 毫秒，并将消息发送到 delayed-exchange 交换机上，路由键为 delayed-queue，消息内容为 "Hello, delayed queue!"。

5.消费延时消息

消费延时消息时，需要设置消费者的 QOS（Quality of Service）参数，以控制消费者的并发处理能力。以下是消费延时消息的示例代码：

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延时队列是一种常见的消息队列应用场景，它可以在消息发送后指定一定的时间后才能被消费者消费，通常用于实现一些延时任务，例如订单超时未支付自动取消等。

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;
                                              */
        // 创建一个支持延时队列的Exchange
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-delayed-type", "direct");
        channel.exchangeDeclare(EXCHANGE_NAME, "x-delayed-message", true, false, arguments);
        // 创建一个延时队列，设置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);
        // 发送消息到延时队列中，设置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();
    }
}

在上面的代码中，我们创建了一个支持延时队列的Exchange，并创建了一个延时队列，设置了x-dead-letter-exchange和x-dead-letter-routing-key参数。然后，我们发送了一条消息到延时队列中，设置了expiration参数，表示这条消息延时10秒后才能被消费。

注意，如果我们想要消费延时队列中的消息，需要创建一个消费者，并监听这个队列。当消息被消费时，需要发送ack确认消息已经被消费，否则消息会一直留在队列中。

到此这篇关于使用Java实现RabbitMQ延时队列的文章就介绍到这了,更多相关Java RabbitMQ延时队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Java语言中4种内部类的超详细讲解
这篇文章主要给大家介绍了关于Java语言中4种内部类的超详细讲解,内部类可以分为:实例内部类、静态内部类和成员内部类,每种内部类都有它特定的一些特点,文中介绍的非常详细,需要的朋友可以参考下
2023-04-04
详解Java中的ThreadLocal
ThreadLocal是JDK包提供的，它提供线程本地变量，如果创建一个ThreadLocal变量，那么访问这个变量的每个线程都会有这个变量的一个副本，在实际多线程操作的时候，操作的是自己本地内存中的变量，从而规避了线程安全问题
2021-06-06
Java多线程并发编程(互斥锁Reentrant Lock)
这篇文章主要介绍了ReentrantLock 互斥锁，在同一时间只能被一个线程所占有，在被持有后并未释放之前，其他线程若想获得该锁只能等待或放弃,需要的朋友可以参考下
2017-05-05
使用idea和gradle编译spring5源码的方法步骤
这篇文章主要介绍了详解使用idea和gradle编译spring5源码，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-12-12
Java响应式编程之Flux与SseEmitter深度解析(附详细代码)
这篇文章主要介绍了Java响应式编程之Flux与SseEmitter深度解析的相关资料,从需求分析、技术背景、使用方法、底层原理、性能对比到生产环境实战,全面解析了这两种技术的特点、应用场景及优缺点,需要的朋友可以参考下
2026-01-01
Java合并区间的实现
本文主要介绍了Java合并区间的实现,通过合理使用集合类和排序算法,可以有效地解决合并区间问题,具有一定的参考价值,感兴趣的可以了解一下
2023-08-08
Java长度不足左位补0的3种实现方法
这篇文章主要介绍了Java长度不足左位补0的3种实现方法小结，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2022-12-12
java生成在线验证码
这篇文章主要介绍了java生成在线验证码,需要的朋友可以参考下
2023-10-10
JDK1.8使用的垃圾回收器和执行GC的时长以及GC的频率方式
这篇文章主要介绍了JDK1.8使用的垃圾回收器和执行GC的时长以及GC的频率方式，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2023-05-05
SpringBoot 图书管理系统(删除、强制登录、更新图书)详细代码
在企业开发中,通常不采用delete语句进行物理删除,而是使用逻辑删除,逻辑删除通过修改标识字段来表示数据已被删除,方便数据恢复,本文给大家介绍SpringBoot 图书管理系统实例代码,感兴趣的朋友跟随小编一起看看吧
2024-09-09