死信队列
死信队列: DLX, Dead-Letter-Exchange
- 利用DLX, 当消息在一个队列中变成死信(dead message)之后, 它能被重新publish到另一个Exchange, 这个Exchange就是DLX
- DLX也是一个正常的Exchange, 和一般的Exchange没有区别, 他能在任何的队列上被指定, 实际上就是设置某个队列的属性
- 当这个队列中有死信时,RabbitMQ就会自动的将这个消息重新发布到设置的Exchange上去, 进而被路由到另一个队列
- 可以监听这个队列的消息做相应的处理, 这个特征可以弥补RabbitMQ3.0以前支持的immediate参数功能
消息变成死信的情况
- 消息被拒绝(basic.reject/basic.nack), 并且requeue=false(关闭重回队列)
- 消息TTL过期
- 队列达到最大长度
死信队列的设置
- 首先需要设置死信队列的Exchange和Queue, 然后进行绑定
- Exchange: dlx.exchange
- Queue: dlx.queue
- RoutingKey: #
- 然后进行正常的声明交换机, 队列, 绑定, 只不过需要在队列上添加一个参数即可
- arguments.put("x-dead-letter-exchange","dlx.exchange");
- 这样消息在过期, requeue=false, 队列在达到最大长度时, 消息就直接路由到死信队列了!
死信队列代码实现
消费者
代码语言：javascript复制package com.dance.redis.mq.rabbit.dlx;
import com.dance.redis.mq.rabbit.RabbitMQHelper;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
public class Receiver4DLXtExchange {
public static void main(String[] args) throws Exception {
Channel channel = RabbitMQHelper.getChannel();
// 声明正常的 exchange queue 路由规则
String queueName = "test_dlx_queue";
String exchangeName = "test_dlx_exchange";
String routingKey = "group.*";
RabbitMQHelper.exchangeDeclare(channel,exchangeName,RabbitMQHelper.EXCHANGE_TYPE_TOPIC);
// 注意在这里要加一个特殊的属性arguments: x-dead-letter-exchange
Map<String, Object> arguments = new HashMap<>();
// 指定死信队列
arguments.put("x-dead-letter-exchange", "dlx.exchange");
// 指定死信队列的路由规则
arguments.put("x-dead-letter-routing-key", "dlx.*");
RabbitMQHelper.queueDeclare(channel,queueName,true,arguments);
channel.queueBind(queueName, exchangeName, routingKey);
// 声明死信队列
// dlx declare:
RabbitMQHelper.exchangeDeclare(channel,"dlx.exchange",RabbitMQHelper.EXCHANGE_TYPE_TOPIC);
channel.queueDeclare("dlx.queue", false, false, false, null);
channel.queueBind("dlx.queue", "dlx.exchange", "#");
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException {
// 拒绝并且不重回队列, 这样就会进入死信队列
channel.basicReject(envelope.getDeliveryTag(),false);
}
};
// 参数：队列名称、是否自动ACK、Consumer
channel.basicConsume(queueName, false, consumer);
//等待回调函数执行完毕之后，关闭资源。
TimeUnit.SECONDS.sleep(50);
channel.close();
RabbitMQHelper.closeConnection();
}
}生产者
代码语言：javascript复制package com.dance.redis.mq.rabbit.dlx;
import com.dance.redis.mq.rabbit.RabbitMQHelper;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;
public class Sender4DLXExchange {
public static void main(String[] args) throws Exception {
Channel channel = RabbitMQHelper.getChannel();
String exchangeName = "test_dlx_exchange";
String routingKey = "group.dlx";
Map<String, Object> headers = new HashMap<>();
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.deliveryMode(2)
.contentEncoding("UTF-8")
.headers(headers).build();
String msg = "Hello World RabbitMQ 4 DLX Exchange Message ... ";
channel.basicPublish(exchangeName, routingKey, props, msg.getBytes());
}
}死信队列测试
启动消费者
启动生产者
查看消费者
应为直接拒绝了, 所以没有消费, 查看控制台
已经被加入到死信队列中了, 为啥是3呢, 应为我之前测试了两次, 这个时候, 如果是写业务的话, 就可以通过消费死信队列的消息, 完成消费失败的, 或者过期的补偿了~
我这里只是用了拒绝策略, TTL过期和队列满, 都会进入, 可以自己试一下
