RabbitMQ消息和队列的TTL以及死信队列和延迟队列

TTL：Time To Live的简称，即过时时间。RabbitMQ能够对消息和队列设置TTL。

设置消息的TTL

目前有两种方法设置消息的TTL，第一种方法是经过队列的属性设置，队列中的全部消息都有相同的过时时间。第二种方法是对消息自己进行单独设置，每条消息的TTL能够不一样。若是两种方法一块儿使用那么消息的TTL以二者之间较小的那个数值为准。消息在队列中的生存时间一旦超过设置的TTL值时，就会变成“死信”（Dead Message），消费者将没法再收到该消息（这点不是绝对的）。

经过队列的属性设置

经过队列属性设置消息的TTL方法是在channel.queueDeclare方法中加入x-message-ttl参数实现的，这个参数的单位是毫秒（ms）。示例部分代码以下：

        Map<String,Object> arg = new HashMap<String, Object>();
        arg.put("x-message-ttl",6000);
        channel.queueDeclare("normalQueue",true,false,false,arg);    

同时也能够经过Policy的方式来设置TTL，示例以下：

rabbitmqctl set_policy TTL ".*"  '{''message-ttl":6000}' --apply-to queues

若是不设置TTL，则表示此消息不会过时；若是设置为0，则表示除非此时能够直接将消息投递给消费者，不然该消息会被当即丢弃。

每条消息设置TTL

针对每条消息设置TTL的方法是在channel.basicPublish方法中加入expiration的属性参数，单位为毫秒。关键代码以下：

        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
        builder.deliveryMode(2);//持久化消息
        builder.expiration("6000");//设置过时时间
        AMQP.BasicProperties properties = builder.build();
        channel.basicPublish("exchange","routingkey",true,properties,"hello".getBytes());

对于经过队列设置的TTL属性的方法，一旦消息过时，就会从队列中抹去，而对于每条消息的设置TTL属性，即便消息过时，也不会立刻从队列中抹去，由于每条消息是否过时是在即将投递到消费者以前判断的。

为何这两种方式的处理不同呢？

　　由于在经过队列设置的TTL属性的方法里，对于已过时的消息确定是在队列的头部，RabbitMQ只是须要按期从队列头开始扫描是否有过时消息便可。而在消息设置TTL的方法里，每条消息的过时时间不一样，若是要删除全部的过时消息势必要扫描整个队列，因此不如等到此消息即将被消费时再判断是否过时，若是过时再进行删除便可。

设置队列的TTL

经过channel.queueDeclare方法中的x-expires参数能够控制队列被自动删除前处于未使用状态的时间。未使用的意思是队列上没有任何的消费者，队列也没有被从新声明，而且在过时时间段内也未调用过Basic.Get命令。

RabbitMQ会确保在过时时间到达后将队列删除，可是不保障删除的动做有多及时。在RabbitMQ重启后，持久化的队列的过时时间会被从新计算。

用于表示过时时间的x-expires参数以毫秒为单位，而且服从和x-message-ttl同样的约束条件，不过不能设置为0，好比该参数设置为1000，则表示该队列若是在1秒内未使用则会被删除。下面举例一个建立过时时间为30分钟的队列：

        Map<String,Object> ars = new HashMap<String, Object>();
        ars.put("x-expires",1800000);
        channel.queueDeclare("myqueue",false,false,false,ars);

死信队列

DLX，全称为Dead-Letter-Exchange，能够称之为死信交换器，也有人称为死信邮箱。当消息在一个队列中变为死信以后，它能被从新被发送到另外一个交换器中，这个交换器就是DLX，绑定DLX的队列就称之为死信队列。

消息变成死信通常是由如下几种状况形成的：

　　❤ 消息被拒绝，而且设置requeue参数为false

　　❤ 消息过时

　　❤ 队列达到最大长度

DLX也是一个正常的交换器，和通常的交换器没有区别，它能在任何的队列上被指定，实际上就是设置某个队列的属性。当这个队列存在死信时，RabbitMQ就会自动的将这个消息从新发布到设置的DLX上去，进而被路由到另外一个队列，即死信队列。能够监听这个队列中的消息以进行相应的处理。

方法：

　　经过在channel.queueDeclare方法中设置x-dead-letter-exchange参数来为这个队列添加DLX。示例代码以下：

    　　channel.exchangeDeclare("dlx_exchange","direct");
        Map<String,Object> ars = new HashMap<String, Object>();
        ars.put("x-dead-letter-exchange","dlx_exchange");
        //能够为这个DLX指定路由键，若是没有特殊指定，则使用原队列的路由键
        ars.put("x-dead-letter-routing-key","dlx-routing-key");
        //为队列添加DLX
        channel.queueDeclare("myqueue",false,false,false,ars);

经过Policy的方式设置：

rabbitmqctl set_policy DLX ".*" '{"dead-letter-change":"dlx_exchange"}' --apply-to queues

下面举个例子，为其设置TTL和DLX，以下：

　　　　 channel.exchangeDeclare("exchange.dlx","direct",true);
        channel.exchangeDeclare("exchange.normal","fanout",true);
        Map<String,Object> ars = new HashMap<String, Object>();
        ars.put("x-dead-letter-exchange","exchange.dlx");
        //能够为这个DLX指定路由键，若是没有特殊指定，则使用原队列的路由键
        ars.put("x-dead-letter-routing-key","dlx-routing-key");
        ars.put("x-message-ttl",10000);
        //为队列添加DLX
        channel.queueDeclare("queue.normal",true,false,false,ars);
        channel.queueBind("queue.normal","exchange.normal","");
        channel.queueDeclare("queue.dlx",true,false,false,null);
        channel.queueBind("queue.dlx","exchange.dlx","dlx-routing-key");
        channel.basicPublish("exchange.normal","rk",MessageProperties.PERSISTENT_TEXT_PLAIN,"dlx".getBytes());

上述例子分别建立了两个交换器exchange.normal和exchange.dlx，分别绑定了两个队列queue.normal和queue.dlx。

上述代码执行过程：

　　生产者首先发送一条携带路由键为“rk”的消息，而后通过交换器exchange.normal顺利的存储到队列queue.normal中，因为队列queue.normal设置了过时时间为10s，在这10s内没有消费者消费这条消息，那么断定这条消息过时。因为设置了DLX，过时之时，消息会被丢给交换器excange.dlx中，这时找到与exchange.dlx匹配的队列queue.dlx，最后消息被存储在queue.dlx这个死信队列中。

　　对于RabbitMQ来讲，DLX是一个很是有用的特性。它能够处理异常的状况下，消息不可以正确的被消费者消费（消费者调用了Basix.Nack或者Basic.Reject）而被置于死信队列中的状况，后续分析过程能够经过消费这个死信队列中的内容来分析当时所遇到的异常状况，进而改善和优化系统。

延迟队列

延迟队列所存储的对象是对应的延迟消息，所谓“延迟消息”是指当消息发送后，并不想让消费者当即拿到消息，而是等待特定时间后，消费者才能拿到这个消息进行消费。

在AMQP协议中，或者RabbitMQ自己没有直接支持延迟队列的功能，可是能够经过DLX和TTL模拟出延迟队列的功能。

其实在文章的上半段中的最后一个例子就是延迟队列的用法：对于queue.dlx这个死信队列来说。假设一个应用中须要将每条消息设置为10秒的延迟，生产者经过exchange.normal这个交换器将发送的消息存储在queue.normal这个队列中。消费者订阅的并不是是queue.normal这个队列，而是queue.dlx这个队列。当消息从queue.normal这个消息队列中过时以后被存入queue.dlx这个队列中，消费者就刚好延迟10秒接收到这个消息。

 　　在真实的应用中，对于延迟队列能够根据延迟时间的长短分为多个等级，通常为5秒，10秒，30秒，1分钟，5分钟，10分钟，30分钟，一个小时这几个维度。

参考：《RabbitMQ实战指南》 朱忠华 编著；