RabbitMQ 死信队列

文章目录

前言
- 1、死信交换机 DLX 与死信队列 DLQ
- 2、死信队列的实现
- - 2.1、声明原队列信息
  - 2.2、声明死信队列信息
  - 2.3、完整示例
- 3、死信消息流转原理

前言

消息过期以后，如果没有任何配置，是会直接丢弃的。我们可以通过配置让这样的消息变成死信（Dead Letter），在别的地方存储。

1、死信交换机 DLX 与死信队列 DLQ

队列在创建的时候可以指定一个死信交换机 DLX（Dead Letter Exchange）。死信交换机绑定的队列被称为死信队列 DLQ（Dead Letter Queue）。

DLX 实际上也是普通的交换机，DLQ 也是普通的队列，只是他有死信的特殊功能。（例如替补球员也是普通球员，只不过他有替补的功能）。

如果消息过期了，队列指定了DLX，就会发送到 DLX。如果 DLX 绑定了 DLQ，就会路由到 DLQ。消费者绑定当前DLQ，消息就会分发到消费者，进行消费。

2、死信队列的实现

2.1、声明原队列信息

声明原交换机（ori_exchange）、原队列（ori_queue），相互绑定，指定原队列的死信交换机为（dlx_exchange）

Map<String, Object> args = new HashMap<String, Object>();
// 声明 Direct 交换机
args.put("x-dead-letter-exchange", "dlx_exchange");
// args.put("x-dead-letter-routing-key", "dlx_routing_key"); // 可选，默认为消息的原路由键
args.put("x-message-ttl", 60000); // 可选，TTL (单位: ms)
args.put("x-max-length", 1000); // 可选，最大消息数
channel.exchangeDeclare("ori_exchange", "direct", true);
channel.queueDeclare("ori_queue", false, false, false, args);

x-dead-letter-exchange：定义了普通队列的 DLX，当消息变为死信时，会被发送到配置的 DLX。
x-dead-letter-routing-key：可选参数，定义消息在 DLX 中的路由键。如果未设置，默认使用消息原本的路由键。
x-message-ttl：可选参数，设置消息在队列中的 TTL，即消息在队列中存在的最大时间（毫秒）。
x-max-length：可选参数，设置队列的最大消息数,超过后会先清理头部消息到死信队列。

注意：如果同时指定了 Message TTL 和 Queue TTL，则小的那个时间生效

2.2、声明死信队列信息

声明死信交换机（DEAD_LETTER_EXCHANGE ）、死信队列（DEAD_LETTER_QUEUE），并且通过"#"绑定，代表无条件路由

channel.exchangeDeclare("dlx_exchange", "direct", true);
channel.queueDeclare("dlx_queue", false, false, false, null);
channel.queueBind("dlx_queue", "dlx_exchange", "dlx_routing_key");

dlx_exchange：定义了一个死信交换机。
dlx_queue：定义了一个专门用于处理死信消息的队列。
queueBind：将死信队列绑定到死信交换机，并指定路由键。

2.3、完整示例

生产者

public class Producer {private final static String ORI_EXCHANGE = "ori_exchange";private final static String ORI_QUEUE = "ori_queue";private final static String DLX_QUEUE = "dlx_queue";private final static String DLX_EXCHANGE = "dlx_exchange";private final static String DLX_ROUTING_KEY = "dlx_routing_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()) {// Create ori queue with DLX configurationMap<String, Object> args = new HashMap<>();args.put("x-dead-letter-exchange", DLX_EXCHANGE);args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);args.put("x-message-ttl", 60000); // 可选，TTL (单位: ms)args.put("x-max-length", 1000); // 可选，最大消息数channel.exchangeDeclare(ORI_EXCHANGE, "direct", true);channel.queueDeclare(ORI_QUEUE, true, false, false, args);channel.queueBind(ORI_QUEUE, ORI_EXCHANGE, "direct");// Create DLX and DLX queuechannel.exchangeDeclare(DLX_EXCHANGE, "direct", true);channel.queueDeclare(DLX_QUEUE, true, false, false, null);channel.queueBind(DLX_QUEUE, DLX_EXCHANGE, DLX_ROUTING_KEY);// Publish a test message to the primary queueString message = "Hello, World!";channel.basicPublish(ORI_EXCHANGE, ORI_QUEUE, null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");// Simulate message rejection// For example, consume and nack the message} catch (Exception e) {e.printStackTrace();}}}

消费者

public class Consumer {private final static String DLX_QUEUE = "dlx_queue";private final static String DLX_EXCHANGE = "dlx_exchange";private final static String DLX_ROUTING_KEY = "dlx_routing_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()) {channel.exchangeDeclare(DLX_EXCHANGE, "direct", true);channel.queueDeclare(DLX_QUEUE, true, false, false, null);channel.queueBind(DLX_QUEUE, DLX_EXCHANGE, DLX_ROUTING_KEY);DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");};channel.basicConsume(DLX_QUEUE, true, deliverCallback, consumerTag -> {});} catch (Exception e) {e.printStackTrace();}}
}