RabbitMQ入门_11_DLX

2023-08-20 07:59:16

参考资料：https://www.rabbitmq.com/dlx.html

队列中的消息可能会成为死信消息（dead lettered）。让消息成为死信消息的事件有：

消息被取消确认（nack 或 reject），且设置为不重入队列（requeue = false）
消息TTL过期
队列达到长度限制

死信消息会被死信交换机（Dead Letter Exchange, DLX）重新发布。

gordon.study.rabbitmq.dlx.TestDlx.java

public class TestDlx {

    private static final String DLX_EXCHANGE_NAME = "exchangeDLX";

    private static final String DLX_QUEUE_NAME = "queueDLX";

    private static final String QUEUE_NAME = "queue";

    public static void main(String[] argv) throws Exception {

        ConnectionFactory factory = new ConnectionFactory();

        factory.setHost("localhost");

        Connection connection = factory.newConnection();

        Channel senderChannel = connection.createChannel();

        Channel consumerChannel = connection.createChannel();

        Map<String, Object> args = new HashMap<String, Object>();

        args.put("x-message-ttl", 3000); // 设置队列中消息存活时间为3秒

        args.put("x-max-length", 5); // 设置队列最大消息数量为5

        args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME); // 设置DLX

        senderChannel.queueDeclare(QUEUE_NAME, false, false, true, args);

        senderChannel.queueDeclare(DLX_QUEUE_NAME, false, false, true, null);

        senderChannel.exchangeDeclare(DLX_EXCHANGE_NAME, "direct", false, true, null);

        // 将死信队列绑定到死信交换机上，绑定键为 QUEUE_NAME。消息发送时使用的绑定键也会是 QUEUE_NAME

        senderChannel.queueBind(DLX_QUEUE_NAME, DLX_EXCHANGE_NAME, QUEUE_NAME);

        // 发布6个消息

        for (int i = 0; i < 6;) {

            String message = "NO. " + ++i;

            senderChannel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));

        }

        // 监视死信队列

        Consumer dlxConsumer = new DefaultConsumer(consumerChannel) {

            @Override

            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)

                    throws IOException {

                String message = new String(body, "UTF-8");

                System.out.printf("consume: %s, envelop: %s, properties: %s\n", message, envelope, properties);

            }

        };

        consumerChannel.basicConsume(DLX_QUEUE_NAME, true, dlxConsumer);

        Thread.sleep(100);

        GetResponse resp = consumerChannel.basicGet(QUEUE_NAME, false);

        consumerChannel.basicReject(resp.getEnvelope().getDeliveryTag(), false);

    }

}

代码第19行通过 x-dead-letter-exchange 参数定义了当前队列指定的死信交换机是 DLX_EXCHANGE_NAME，当前队列中所有的死信消息都将交由 DLX_EXCHANGE_NAME 再一次分发到新的队列。

代码第23行创建了死信交换机 DLX_EXCHANGE_NAME，可见，死信交换机就是普通的交换机。

从以上两部分代码顺序可知，死信交换机可以在被队列引用后才创建。RabbitMQ 不会去验证死信交换机设置是否有效，当死信消息找不到指定的交换机时，死信消息会被RabbitMQ安静的丢弃，而不是抛出异常。

既然死信交换机就是普通的交换机，那么它就需要根据消息的绑定键来分发消息。死信消息的绑定键遵守以下规则：当队列指定了死信路由键（x-dead-letter-routing-key）参数时，死信消息使用该参数指定的路由键作为自己的路由键；否则使用消息原来的路由键。所以，示例代码中，死信消息的路由键是代码第30行发布消息时指定的路由键 QUEUE_NAME。如果想修改死信消息路由键，可以在第19行下面增加

        args.put("x-dead-letter-routing-key", "some-routing-key");

观察日志输出：

consume: NO. 1, envelop: Envelope(deliveryTag=1, redeliver=false, exchange=exchangeDLX, routingKey=queue), properties: #contentHeader<basic>(content-type=null, content-encoding=null, headers={x-death=[{queue=queue, time=Sat Jun 10 11:51:48 CST 2017, count=1, reason=maxlen, routing-keys=[queue], exchange=}]}, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)

consume: NO. 2, envelop: Envelope(deliveryTag=3, redeliver=false, exchange=exchangeDLX, routingKey=queue), properties: #contentHeader<basic>(content-type=null, content-encoding=null, headers={x-death=[{queue=queue, time=Sat Jun 10 11:51:48 CST 2017, count=1, reason=rejected, routing-keys=[queue], exchange=}]}, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)

consume: NO. 3, envelop: Envelope(deliveryTag=4, redeliver=false, exchange=exchangeDLX, routingKey=queue), properties: #contentHeader<basic>(content-type=null, content-encoding=null, headers={x-death=[{queue=queue, time=Sat Jun 10 11:51:51 CST 2017, count=1, reason=expired, routing-keys=[queue], exchange=}]}, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)

......

死信消息相比原来的消息发生了一些改变。除了上面提到的 exchange 变为死信交换机名称，routingKey 可能变为新的路由键（由 x-dead-letter-routing-key 参数决定），死信处理过程还在死信消息头中增加了 x-death 数组信息。每一次死信事件对应一个数组项，包含以下字段，

queue：本次死信事件发生前，消息所属队列
reason：死信原因，分为 rejected、expired 与 maxlen
time：死信事件发生时间
exchange：死信事件前，消息发布时指定的交换机
routing-keys：死信事件前，消息发布时指定的路由键
count：当前 queue 与当前 reason 表述的死信事件发生的次数
original-expiration：对于消息 TTL，因为超时导致死信时，会移除 TTL（否则永远触发超时），该字段记录原来设定的消息 TTL 值

当死信消息再次触发死信事件时，一般会产生一个新的数组项，插到数组的最前头。但是，如果 x-death 数组已经包含一个相同 queue 与 reason 的数组项，则直接将该数组项移到数组最前头，并将其 count 值加一。

未确认问题：

1. 当消息被 reject 回队列头，同时又超过队列长度限制时，怎么处理？

试验结果好像是直接变为 maxlen reason 的死信消息

2. DLX很可能形成环（最简单的场景就是DLX与原交换机相同），这时消息有可能无限触发死信事件吗（例如超过队列长度限制）？

官方说法是处在DLX环中的消息，如果经历了整个环都没有触发过 rejected reason 的死信事件，则抛弃该消息。

这些问题有点偏，目前就不花时间研究了。

码农公寓

相关文章