RabbitMQ-中死信交换机

在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();// 设置消息的TTLint 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 -> {});// ...其他业务代码}
}