当前位置：首页 > article >正文

RabbitMQ如何实现延时队列？

article 2024/12/23 7:39:01

1.什么是RabbitMQ?

RabbitMQ 是一个开源的消息代理软件，它实现了高级消息队列协议（AMQP）。它允许不同的应用程序通过消息进行通信，并且能够可靠地传递、路由和存储消息。

2.利用RabbitMQ实现延时队列的方法

2.1利用死信交换机（Dead - Letter - Exchange，DLX）来实现

原理：

当消息在队列中过期后，会被发送到死信交换机，消费者从连接死信交换机的队列中获取过期后的消息进行处理。

步骤示例：

声明交换机和队列：

首先需要声明一个普通交换机（比如direct类型的交换机）和一个死信交换机（同样可以是direct类型），以及两个队列，一个是普通队列用于接收消息，另一个是连接死信交换机的队列用于接收过期消息。

假设使用 Java 和 RabbitMQ 的 Java 客户端amqp - client来实现，代码如下：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;
public class DelayedQueueExample {
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    private static final String DEAD_LETTER_EXCHANGE = "dead_letter_exchange";
    private static final String NORMAL_QUEUE = "normal_queue";
    private static final String DEAD_LETTER_QUEUE = "dead_letter_queue";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 声明死信交换机
        channel.exchangeDeclare(DEAD_LETTER_EXCHANGE, "direct");
        // 声明普通交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, "direct");
        // 声明死信队列
        channel.queueDeclare(DEAD_LETTER_QUEUE, false, false, false, null);
        // 声明普通队列，并设置死信交换机和路由键
        Map<String, Object> args = new HashMap<>();
        args.put("x - dead - letter - exchange", DEAD_LETTER_EXCHANGE);
        args.put("x - dead - letter - routing - key", "dead_letter_key");
        channel.queueDeclare(NORMAL_QUEUE, false, false, false, args);
        // 将普通队列绑定到普通交换机
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "normal_key");
        // 将死信队列绑定到死信交换机
        channel.queueBind(DEAD_LETTER_QUEUE, DEAD_LETTER_EXCHANGE, "dead_letter_key");
    }
}

发送消息并设置过期时间

当队列和交换机声明完成后，可以通过普通交换机发送消息到普通队列，并设置消息的过期时间（expiration属性）。例如，发送一条文本消息并设置过期时间为 5000 毫秒（5 秒）：

String message = "Delayed Message";
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
       .expiration("5000")
       .build();
channel.basicPublish(NORMAL_EXCHANGE, "normal_key", properties, message.getBytes());

消费者从死信队列中接收消息进行处理。可以使用basicConsume方法来设置消费者从死信队列接收消息。例如：

Consumer consumer = new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        String receivedMessage = new String(body, "UTF - 8");
        System.out.println("Received delayed message: " + receivedMessage);
    }
};
channel.basicConsume(DEAD_LETTER_QUEUE, true, consumer);

2.2 使用rabbitmq_delayed_message_exchange 插件

原理：RabbitMQ 提供了一个名为rabbitmq_delayed_message_exchange的插件来实现延时队列。这个插件允许创建一种特殊的交换机，称为延迟交换机（Delayed Message Exchange）。当消息发送到这种交换机时，可以指定一个延迟时间，在延迟时间过后，消息才会被路由到相应的队列，然后消费者就可以从队列中获取并处理消息。

步骤示例：

安装插件：

首先需要安装rabbitmq_delayed_message_exchange插件。对于不同的操作系统和 RabbitMQ 安装方式，安装步骤会有所不同。在基于 Debian 或 Ubuntu 的系统中，如果是通过包管理方式安装的 RabbitMQ，可以使用以下命令安装插件：

sudo rabbitmq - plugins enable rabbitmq_delayed_message_exchange

安装完成后，需要重启 RabbitMQ 服务使插件生效。

声明延迟交换机和队列：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class DelayedQueuePluginExample {
    private static final String DELAYED_EXCHANGE = "delayed_exchange";
    private static final String QUEUE = "delayed_queue";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 声明延迟交换机，设置类型为x - delayed - message
        channel.exchangeDeclare(DELAYED_EXCHANGE, "x - delayed - message");
        channel.queueDeclare(QUEUE, false, false, false, null);
        // 将队列绑定到延迟交换机
        channel.queueBind(QUEUE, DELAYED_EXCHANGE, "");
    }
}

发送延迟消息：

String message = "Delayed Message via Plugin";
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
       .headers(new HashMap<String, Object>() {
            {
                put("x - delayed - type", "direct");
                put("x - delay", 10000);
            }
        })
       .build();
channel.basicPublish(DELAYED_EXCHANGE, "", properties, message.getBytes());

后续消费者接收消息就和普通交换机一样。

查看全文

http://www.kler.cn/a/447609.html