平常項目中很多場景需要使用延時消息處理，例如訂單超過多久沒有支付需要取消等。如何在消息中間件RabbitMQ中實現該功能？下面描述下使用
Dead Letter Exchange實現延時消息場景，當然會有別的其他實現方式。
1. 什么是Dead Letter Exchange？
RabbitMQ中通常消息被直接發送到隊列中或者從Exchange中Route到隊列上后，消息如果被消費者消費完畢並確認后消息就會從Broker中被刪除。
如果存在以下三種情況，同時隊列上設置了Dead Letter Exchange，消息會被轉送到Dead Letter Exchange中。
- 消息被拒絕(
basicReject或者basicNack) requeue=false - 消息存活時間超過了TTL預設值(
x-message-ttl) - 隊列滿了
Dead Letter Exchange像平常的Exchange一樣，可以設置它的BuiltinExchangeType，也可以為它綁定隊列。
這里我們可以通過設定Dead Letter Exchange，並為它綁定一個隊列，然后定義Consumer消費這個隊列，就可以達到處理延時消息的功能。
2. 代碼實例
流程先：
I. 定義消息生產者
/***
* 消息發送者
*/
static class NormalEXSend {
private Connection conn;
private Channel chnl;
public NormalEXSend(String tag) throws IOException, TimeoutException {
ConnectionFactory connFact = initConnFac();
conn = connFact.newConnection();
chnl = conn.createChannel();
// 定義正常工作Exchange
chnl.exchangeDeclare(WORKER_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
// 定義 dead letter exchange
chnl.exchangeDeclare(DELAY_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 60000); // timeout 1min
args.put("x-dead-letter-exchange", DELAY_EXCHANGE_NAME);
args.put("x-dead-letter-routing-key", DEAD_ROUTING_KEY);
// 定義正常工作Queue同時設置dead letter exchange
chnl.queueDeclare(WORKER_QUEUE_NAME, false, false, false, args);
// 綁定到正常工作Exchange
chnl.queueBind(WORKER_QUEUE_NAME, WORKER_EXCHANGE_NAME, tag);
}
/**
* 發送消息
* @param key
* @param msg
* @throws IOException
*/
public void send(String key, String msg) throws IOException {
AMQP.BasicProperties props = MessageProperties.PERSISTENT_TEXT_PLAIN;
// send a message to a exchange
chnl.basicPublish(WORKER_EXCHANGE_NAME, key, props, msg.getBytes());
System.out.println(String.format("[%s|%s|Sender] send 【%s】 to exchange:%s", Thread.currentThread().getName(), System.currentTimeMillis(), msg, WORKER_EXCHANGE_NAME));
}
}
II. 定義延時消息處理者
其中receive方法中
consumer的handleDelivery方法參數properties可以獲取到消息的death原因properties.getHeaders().get("x-first-death-reason")，可能值rejected | expired | maxlen。此處可以根據判斷此值去處理由於超時而引起death的消息（就是我們想要處理的延時消息）。
/**
* 延時消息處理者
*/
static class DelayEXRecv {
private Connection conn;
private Channel chnl;
public DelayEXRecv() throws IOException, TimeoutException {
ConnectionFactory connFact = initConnFac();
conn = connFact.newConnection();
chnl = conn.createChannel();
// 定義延時消息隊列
chnl.queueDeclare(DELAY_QUEUE_NAME, false, false, false, null);
// 綁定到延時消息Exchange
chnl.queueBind(DELAY_QUEUE_NAME, DELAY_EXCHANGE_NAME, DEAD_ROUTING_KEY);
}
/**
* 接收消息
* @throws IOException
*/
public void receive() throws IOException {
chnl.basicQos(1);
// no auto ack
boolean autoAck = false;
chnl.basicConsume(DELAY_QUEUE_NAME, autoAck, new DefaultConsumer(chnl) {
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
// 打印出消息的death原因 rejected | expired | maxlen
// 項目中可以根據原因處理目標消息
System.out.println(String.format("[%s|%s|Delay_Receiver] received the delay msg 【%s】 from EXCHANGE: %s, the delay reason is: %s", Thread.currentThread().getName(), System.currentTimeMillis(), message, envelope.getExchange(), properties.getHeaders().get("x-first-death-reason")));
// 確認消息
chnl.basicAck(envelope.getDeliveryTag(), false);
}
});
}
}
III. 試驗一把
private static final String WORKER_EXCHANGE_NAME = "exchange.worker";
private static final String DELAY_EXCHANGE_NAME = "exchange.delay";
private static final String WORKER_QUEUE_NAME = "queue.worker";
private static final String DELAY_QUEUE_NAME = "queue.delay";
private static final String DEAD_ROUTING_KEY = "dead.key.message";
public static void main(String[] args) {
ExecutorService exec = Executors.newFixedThreadPool(2);
exec.execute(new Runnable() {
@Override
public void run() {
try {
String key = "worker";
NormalEXSend sender = new NormalEXSend(key);
for (int i =0; i < 5; i++) {
sender.send(key, String.format("YaYYY, one message, No.:%s!", i));
Thread.sleep(3000);
}
} catch (IOException | TimeoutException | InterruptedException e) {
e.printStackTrace();
}
}
});
exec.execute(new Runnable() {
@Override
public void run() {
try {
DelayEXRecv receiver = new DelayEXRecv();
receiver.receive();
System.out.println(String.format("[%s|%s|Delay_Receiver] Starting the Delay Msg Receiver process...", Thread.currentThread().getName(), System.currentTimeMillis()));
} catch (IOException | TimeoutException e) {
e.printStackTrace();
}
}
});
exec.shutdown();
}
IV. 打印結果
[pool-1-thread-2|1515750089010|Delay_Receiver] Starting the Delay Msg Receiver process...
[pool-1-thread-1|1515750089020|Sender] send 【YaYYY, one message, No.:0!】 to exchange:exchange.worker
[pool-1-thread-1|1515750092020|Sender] send 【YaYYY, one message, No.:1!】 to exchange:exchange.worker
[pool-1-thread-1|1515750095020|Sender] send 【YaYYY, one message, No.:2!】 to exchange:exchange.worker
[pool-1-thread-1|1515750098021|Sender] send 【YaYYY, one message, No.:3!】 to exchange:exchange.worker
[pool-1-thread-1|1515750101022|Sender] send 【YaYYY, one message, No.:4!】 to exchange:exchange.worker
[pool-2-thread-4|1515750149038|Delay_Receiver] received the delay msg 【YaYYY, one message, No.:0!】 from EXCHANGE: exchange.delay, the delay reason is: expired
[pool-2-thread-5|1515750152035|Delay_Receiver] received the delay msg 【YaYYY, one message, No.:1!】 from EXCHANGE: exchange.delay, the delay reason is: expired
[pool-2-thread-6|1515750155035|Delay_Receiver] received the delay msg 【YaYYY, one message, No.:2!】 from EXCHANGE: exchange.delay, the delay reason is: expired
[pool-2-thread-7|1515750158036|Delay_Receiver] received the delay msg 【YaYYY, one message, No.:3!】 from EXCHANGE: exchange.delay, the delay reason is: expired
[pool-2-thread-8|1515750161036|Delay_Receiver] received the delay msg 【YaYYY, one message, No.:4!】 from EXCHANGE: exchange.delay, the delay reason is: expired
可以看出消息是在指定延時的1min后才被獲取消費。
Yayy, 至此結束。