RabbitMQ 之七 TTL与Dead lettering

RabbitMq允许我们为message和queue设置过期时间TTL (time to live) ,可以通过代码或者控制台设置TTL，可以为一个或者多个queue设置TTL，或者为每个独立的message设置TTL。

设置x消息TTL的两种方式

一、为队列中的每个消息设置TTL

在声明队列的时候通过如下代码设置：

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-message-ttl", 60000);
channel.queueDeclare("myqueue", false, false, false, args);

这种情况下，这个queue中的消息存活时间超过配置的TTL就是死亡的消息，各个queue中的消息死亡互相不影响。

TTL的值是一个大于0的整数，单位是毫秒。

如果queue存在的情况下，可以通过命令行来设置TTL。

如果消息被重新分配给queue，消息的原始过期时间将会被保留。

当TTL被设置为0的情况下，意味着消息一旦到达queue就会过期，除非马上被分配给一个consumer，因此设置TTL为0可以当作是

basic.publish方法中immediate参数标志的替代。

二、在发布消息的时候设置TTL

可以为单个消息设置TTL，在发布消息的时候通过如下代码设置

byte[] messageBodyBytes = "Hello, world!".getBytes();
AMQP.BasicProperties properties = new AMQP.BasicProperties();
properties.setExpiration("60000");
channel.basicPublish("my-exchange", "routing-key", properties, messageBodyBytes);

为queue设置TTL
为queue设置TTL意味着queue如果在TTL时间限制内没被使用过将会被自动删除，未被使用过意味着这个queue没有consumers，这个queue没有重新声明过，basic.get方法
没有被调用过。
通过如下代码为queue设置TTL

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

x-expires的值是一个大于0的整数，单位是毫秒。

Dead Letter Exchanges

以下几种情况发生时，使用Dead Letter可以让一个消息重新发布到另一个交换机，
消息被拒绝(basic.reject或者basic,nack被调用)并且requeue=false;
消息的TTL到期
queue的长度已满
Dead letter exchanges(DLXS)也是一个普通的exchange。
以下代码设置queue的是dlxs

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);

以上代码声明了一个新的exchange名字为some.exchange.name,然后将这个exchange设置为queue name为myqueue的DLXS
当队列声明的时候，这个exchange可以还没被声明，但是当message被dead-lettered时这个exchange必须存在。也可以为被dead-lettering的消息设置routing-key。使用参数

args.put("x-dead-letter-routing-key", "some-routing-key");

Dead-Lettered Messages

死信被re-publish后，会在消息的header中增加一个叫做“x-death"的数组内容，包含了以下字段内容：

queue 消息被重新路由前所在的queue的名称

reason 消息成为死信的原因，有如下几种

rejected 消息被拒绝并且requeue=false,

expired message的TTL

maxLen queue的最大消息长度

time 消息成为死信的日期时间

exchange 消息被重新路由到exchange的名称。

routing-keys 消息被发布时的routing-keys

original-expiration 消息的原始expiration属性。

使用TTL和D饿啊的lettering可以实现延迟队列

模型如下：



代码如下：Recv.java

package org.delayQueue;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;

public class Recv {

public static void main(String[] args) throws IOException, TimeoutException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare("exchangeA", "direct");
channel.exchangeDeclare("exchangeB", "direct");

Map<String, Object> queueArgs = new HashMap<String, Object>();
queueArgs.put("x-dead-letter-exchange", "exchangeB");
channel.queueDeclare("queueA", false, false, false, queueArgs);
channel.queueDeclare("queueB", false, false, false, null);

// 绑定路由
channel.queueBind("queueA", "exchangeA", "");
channel.queueBind("queueB", "exchangeB", "");
System.out.println("ready to receive message");
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException
{
String message = new String(body, "UTF-8");
properties.getHeaders();
System.out.println("received message:" + message + ",date:" + System.currentTimeMillis());
}
};
channel.basicConsume("queueB", true, consumer);
}
}

Producer.java

package org.delayQueue;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Producer {

public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare("exchangeA", "direct");
channel.exchangeDeclare("exchangeB", "direct");

String message = "hello world!" + System.currentTimeMillis();
// 设置延时属性
AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
// 持久性 non-persistent (1) or persistent (2)
AMQP.BasicProperties properties = builder.expiration("3000").deliveryMode(2).build();
// routingKey =delay_queue 进行转发
channel.basicPublish("exchangeA", "", properties, message.getBytes());
System.out.println("sent message: " + message + ",date:" + System.currentTimeMillis());
// 关闭频道和连接
channel.close();
connection.close();
}
}

执行结果如下





参考：http://www.rabbitmq.com/dlx.html