RabbitMQ入门_11_DLX

参考资料: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 的死信事件,则抛弃该消息。

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

上一篇:KnockoutJs学习笔记(十二)


下一篇:HDU 2089 - 不要62 - [数位DP][入门题]