RabbitMQ中的延迟消息

Question

是否可以通过RabbitMQ延迟发送消息？例如，我想在30分钟后终止客户端会话，并发送一条消息，该消息将在30分钟后处理。

Answer 1

您可以尝试以下两种方法：

旧的方法：在每个消息/队列(策略)中设置TTL(生存时间)头，然后引入来处理它。一旦ttl过期，您的消息将从DLQ移动到主队列，以便侦听器可以处理它。

最新的方法：最近RabbitMQ提出了RabbitMQ Delayed插件，使用它你可以实现相同的插件支持，这是从RabbitMQ-3.5.8开始提供的。

您可以使用类型x-delay - message声明一个交换，然后使用自定义头部x- delay发布消息，以毫秒为单位表示消息的延迟时间。消息将在延迟x-毫秒后传送到相应的队列

byte[] messageBodyBytes = "delayed payload".getBytes("UTF-8");
Map<String, Object> headers = new HashMap<String, Object>();
headers.put("x-delay", 5000);
AMQP.BasicProperties.Builder props = new 
AMQP.BasicProperties.Builder().headers(headers);
channel.basicPublish("my-exchange", "", props.build(), messageBodyBytes);

更多信息请点击这里：git

Answer 2

随着RabbitMQ v2.8的发布，计划交付现在是可用的，但作为一个间接功能：http://www.javacodegeeks.com/2012/04/rabbitmq-scheduled-message-delivery.html

Answer 3

多亏了Norman's answer，我可以在Node.js中实现它。

从代码中看，一切都很清楚。

var ch = channel;
ch.assertExchange("my_intermediate_exchange", 'fanout', {durable: false});
ch.assertExchange("my_final_delayed_exchange", 'fanout', {durable: false});

// setup intermediate queue which will never be listened.
// all messages are TTLed so when they are "dead", they come to another exchange
ch.assertQueue("my_intermediate_queue", {
      deadLetterExchange: "my_final_delayed_exchange",
      messageTtl: 5000, // 5sec
}, function (err, q) {
      ch.bindQueue(q.queue, "my_intermediate_exchange", '');
});

ch.assertQueue("my_final_delayed_queue", {}, function (err, q) {
      ch.bindQueue(q.queue, "my_final_delayed_exchange", '');

      ch.consume(q.queue, function (msg) {
          console.log("delayed - [x] %s", msg.content.toString());
      }, {noAck: true});
});