RabbitMQ入门_11_DLX

10477777

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

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

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

　　死信消息会被死信交换机（Dead Letter Exchange, DLX）重新发布。
　　gordon.study.rabbitmq.dlx.TestDlx.java
public>　　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(&quot;x-message-ttl&quot;, 3000); // 设置队列中消息存活时间为3秒
　　args.put(&quot;x-max-length&quot;, 5); // 设置队列最大消息数量为5
　　args.put(&quot;x-dead-letter-exchange&quot;, 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, &quot;direct&quot;, 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 = &quot;NO. &quot; + ++i;
　　senderChannel.basicPublish(&quot;&quot;, QUEUE_NAME, null, message.getBytes(&quot;UTF-8&quot;));
　　}
　　// 监视死信队列
　　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, &quot;UTF-8&quot;);
　　System.out.printf(&quot;consume: %s, envelop: %s, properties: %s\n&quot;, 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(&quot;x-dead-letter-routing-key&quot;, &quot;some-routing-key&quot;);　　

　　观察日志输出：
　　

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 的死信事件，则抛弃该消息。
　　这些问题有点偏，目前就不花时间研究了。