利用rabbitMq来实现延迟队列的功能
那么如何利用rabbitMq来实现延迟队列的功能呢？
请先注意一点，RabbitMQ本身没有直接支持延迟队列功能，但是可以通过以下特性模拟出延迟队列的功能。那么这是通过哪些特性呢,那就让我们来认识一下这两个特性吧.
TTL（过期时间）
RabbitMQ可以对消息和队列设置TTL（过期时间）。
RabbitMQ针对队列中的消息过期时间(Time To Live, TTL)有两种方法可以设置。第一种方法是通过队列属xìng设置，队列中所有消息都有相同的过期时间。第二种方法是对消息进行单独设置，每条消息TTL可以不同。如果上述两种方法同时使用，则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值，就成为dead message,消费者将无法再收到该消息。
1. 设置队列属xìng。
通过队列属xìng设置消息TTL的方法是在queue.declare方法中加入x-message-ttl参数，单位为ms。
JAVA中SDK设置方法如下：
- Map<String, Object> args = new HashMap<String, Object>();
- args.put("x-message-ttl", 60000);
- channel.queueDeclare("myqueue", false, false, false, args);
- rabbitmqctl set_policy TTL ".*" '{"message-ttl":60000}' --apply-to queues
- $ curl -i -u guest:guest -H "content-type:application/json" -XPUT
- -d'{"auto_delete":false,"durable":true,"arguments":{"x-message-ttl": 60000}}'
- http://localhost:15672/api/queues/{vhost}/{queuename}
如果不设置TTL，则表示此消息不会过期。如果将TTL设置为0，则表示除非此时可以直接将消息投递到消费者，否则该消息会被立即丢弃，这个特xìng可以部分代替RabbitMQ 3.0.0以前支持的immediate参数，之所以说部分代替，是因为immediate参数在投递失败会有basic.return方法将消息体返回（这个功能可以利用下面的DLX功能来实现），详见RabbitMQ（二）AMQP协议mandatory和immediate标志位区别。
2. 设置消息属xìng。
针对每条消息设置TTL的方法是在basic.publish方法中加入expiration的属xìng参数，单位为ms。
JAVASDK设置如下：
- byte[] messageBodyBytes = "Hello, world!".getBytes();
- AMQP.BasicProperties properties = new AMQP.BasicProperties();
- properties.setExpiration("60000");
- channel.basicPublish("myexchange", "routingkey", properties, messageBodyBytes);
还可以通过HTTP接口调用如下： - $ curl -i -u guest:guest -H "content-type:application/json" -XPOST
- -d'{"properties":{"expiration":"60000"},"routing_key":"routingkey","payload":"my
- body","payload_encoding":"string"}'
- http://localhost:15672/api/exchanges/{vhost}/{exchangename}/publish
对于第一种设置队列TTL属xìng的方法，一旦消息过期，就会从队列中抹去，而第二种方法里，即使消息过期，也不会马上从队列中抹去，因为每条消息是否过期时在即将投递到消费者之前判定的，为什么两者得chǔ理方法不一致？因为第一种方法里，队列中已过期的消息肯定在队列头部，RabbitMQ只要定期从队头开始扫描是否有过期消息即可，而第二种方法里，每条消息的过期时间不同，如果要删除所有过期消息，势必要扫描整个队列，所以不如等到此消息即将被消费时再判定是否过期，如果过期，再进行删除。
DLX（死信邮箱） 利用DLX，当消息在一个队列中变成死信后，它能被重新publish到另一个Exchange，这个Exchange就是DLX。消息变成死信一向有以下几种情况： - 消息被拒绝（basic.reject or basic.nack）并且requeue=false
- 消息TTL过期
- 队列达到最大长度
DLX也是一下正常的Exchange同一般的Exchange没有区别，它能在任何的队列上被指定，实际上就是设置某个队列的属xìng，当这个队列中有死信时，RabbitMQ就会自动的将这个消息重新发布到设置的Exchange中去，进而被路由到另一个队列，publish可以监听这个队列中消息做相应的chǔ理，这个特xìng可以弥补RabbitMQ 3.0.0以前支持的immediate参数中的向publish确认的功能。
JAVA的SDK设置 - channel.exchangeDeclare("some.exchange.name", "direct");
-
- Map<String, Object> args = new HashMap<String, Object>();
- args.put("x-dead-letter-exchange", "some.exchange.name");
- channel.queueDeclare("myqueue", false, false, false, args);
你也可以为这个DLX指定routing key，如果没有特殊指定，则使用原队列的routing key - args.put("x-dead-letter-routing-key", "some-routing-key");
还可以使用policy来配置： - rabbitmqctl set_policy DLX ".*" '{"dead-letter-exchange":"my-dlx"}' --apply-to queues
“死信”被republish时，会在消息的header中增加一个jiào做“x-death"的数组内容，包含了以下字段内容： - queue - the name of the queue the message was in before it was dead-lettered,
- reason - see below,
- time - the date and time the message was dead lettered as a 64-bit AMQP format timestamp,
- exchange - the exchange the message was published to (note that this will be a dead letter exchange if the message is dead lettered multiple times), and
- routing-keys - the routing keys (including CC keys but excluding BCC ones) the message was published with.
- original-expiration (if the message was dead-letterered due to per-message TTL) - the originalexpiration property of the message. The expiration property is removed from the message on dead-lettering in order to prevent it from expiring again in any queues it is routed to.
The reason is a name describing why the message was dead-lettered and is one of the following:- rejected - the message was rejected with requeue=false,
- expired - the TTL of the message expired; or
- maxlen - the maximum allowed queue length was exceeded.
在使用immediate参数时，如果消费者拒收消息这个应答是没有办法通知到publish的，但是利用DLX，拒收的消息会变成“死信”，这个死信会被republish到DLX路由的队列中去，分析这个新的信息的header中的x-death内容，可以判断出消息变成死信的原因，这样publish端可以做更灵活的chǔ理。
本站仅提供存储服务，所有内容均由用户发布，如发现有害或侵权内容，请
点击举报。
