RabbitMQ-中死信交换机

本文介绍: 在RabbitMQ中，死信交换机（DLX，Dead Letter Exchange）是一种用于处理无法正常消费的消息的机制。当消息在一个队列中变成死信（dead letter）之后，它可以被重新发布到另一个交换机，这个交换机就被称为死信交换机。RabbitMQ延迟队列可以通过使用死信交换机和消息的TTL设置来实现。

在RabbitMQ中，死信交换机（DLX，Dead Letter Exchange）是一种用于处理无法正常消费的消息的机制。当消息在一个队列中变成死信（dead letter）之后，它可以被重新发布到另一个交换机，这个交换机就被称为死信交换机。消息变成死信的原因通常有以下几种情况：

消息被拒绝（Basic.Reject/Basic.Nack）并且设置为不重新入队列。
消息在队列中的存活时间（TTL）到期。
队列达到最大长度。

RabbitMQ延迟队列可以通过使用死信交换机和消息的TTL设置来实现。

设置死信交换机和延迟队列

以下是如何使用Java代码来设置死信交换机和延迟队列：

import com.rabbitmq.client.*;

public class DLXExample {

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        String exchangeName = "dlx_exchange";
        String routingKey = "dlx_routing_key";
        String queueName = "dlx_queue";

        // 声明一个死信交换机
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC, true);

        // 设置队列的死信交换机属性
        AMQP.Queue.DeclareOk queue = channel.queueDeclare(queueName, true, false, false, Map.of(
                "x-dead-letter-exchange", exchangeName,
                "x-dead-letter-routing-key", routingKey
        ));

        // 绑定队列到死信交换机
        channel.queueBind(queue.getQueue(), exchangeName, routingKey);

        // ...其他业务代码
    }
}

在上述代码中，通过x-dead-letter-exchange和x-dead-letter-routing-key参数，我们将队列绑定到了一个死信交换机。当该队列中的消息变为死信时，它们会被发送到这个死信交换机，并根据路由键路由到相应的队列。

为了实现延迟队列的效果，可以设置消息的TTL（Time-To-Live），或者设置队列的TTL。当消息过期后，如果队列设置了死信交换机，消息就会被发送到死信交换机。

发布消息到延迟队列

发布带有TTL的消息到队列：

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;

import java.util.HashMap;
import java.util.Map;

public class DelayedMessagePublisher {

    public static void main(String[] argv) throws Exception {
        // ...建立连接和通道的代码

        String queueName = "delayed_queue";
        String dlxExchangeName = "dlx_exchange";
        String dlxRoutingKey = "dlx_routing_key";

        // 声明延迟队列，将消息过期后路由到死信交换机
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", dlxExchangeName);
        args.put("x-dead-letter-routing-key", dlxRoutingKey);
        channel.queueDeclare(queueName, true, false, false, args);

        String message = "This is a delayed message";
        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();

        // 设置消息的TTL
        int ttl = 10000; // 消息的存活时间，单位为毫秒
        builder.expiration(String.valueOf(ttl));
        AMQP.BasicProperties properties = builder.build();

        // 发布消息到延迟队列
        channel.basicPublish("", queueName, properties, message.getBytes());
        System.out.println("Sent message: " + message + " with TTL: " + ttl);

        // ...关闭连接等清理代码
    }
}

在这个例子中，我们设置了消息的TTL属性，并将其发布到了延迟队列。消息过期后，它会自动转发到绑定的死信交换机，然后进入相应的队列。

消费死信队列中的消息

消费来自死信交换机的消息和常规消息消费类似，但是通常这些消息需要特殊处理，因为它们可能代表了失败或者需要延迟处理的情况。

import com.rabbitmq.client.*;

public class DLXConsumer {

    public static void main(String[] argv) throws Exception {
        // ...建立连接和通道的代码

        String dlxQueueName = "dlx_queue";

        // 启动消费者监听死信队列
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Received from DLX: " + message);
            // 确认消息
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        channel.basicConsume(dlxQueueName, false, deliverCallback, consumerTag -> {});

        // ...其他业务代码
    }
}