简介
什么是死信队列
?其本质也是一个队列,只是配置上了一个 Dead-Letter-Exchange
属性。
什么是死信
?死信其实也可以理解为普通的的消息,只是在以下几种情况会称为死信。
- 消息 TTL 过期时。
- 被
Nack/Reject
并且requeue = false
时。 - 队列达到最大长度(超过
Max length
时)。
死信队列工作原理
死信队列
中成为死信的消息会转发到普通的交换机
中(起名带有 dlx
关键字,是为了方便知道这个是用来接收死信用的),再路由到一个普通的队列
(起名同交换机)中,最后再由消费这个 DLX
队列的消费者处理,是记录异常或者通知运维人员。
这里根据图明确一下:
红色的为死信队列
,在没有死信消息出现的时候,其就是普通的队列,当出现死信时,会转发到指定的交换机中。
配置
- 建立
死信队列
// 配置声明队列时使用的参数
Map<String, Object> args = new HashMap<>(1);
// 设置死信队列指向的交换机
args.put("x-dead-letter-exchange", EXCHANGE_DLX);
// 声明队列
channel.queueDeclare(
// 队列名称,名称的规则:什么类型.干什么的
QUEUE_NAME,
// 持久,如果为 false,重启后队列就没了
true,
// 是否为当前 connection 独占,如果为 true,那其他 connection 就无法连接
false,
// 自动删除,如果为 true,队列中没有消息之后,队列就没了
false,
// 参数
args
);
也可以通过 Web 管控台建立
死信队列会多一个 DLX
的特征
完整的意思就是:建立了一个死信队列叫做 queue.cat
,如果出现死信消息就会把这个死信消息转发到交换机 exchange.dlx
上,exchange.dlx
type 为 topic
,通过路由 Key #
绑定到队列 queue.dlx
上,死信消息最终由 queue.dlx
的消费者进行消费。
完整代码:
@Slf4j
public class DlxTest {
private static final String EXCHANGE_NAME = "exchange.cat.dog";
private static final String EXCHANGE_DLX = "exchange.dlx";
private static final String QUEUE_NAME = "queue.cat";
private static final String QUEUE_DLX = "queue.dlx";
private static final String KEY_NAME = "key.yingduan";
private static final String KEY_DLX = "#";
private static Connection connection;
private static Channel channel;
/**
* 监听队列回调函数
*/
DeliverCallback deliverCallback = (consumerTag, message) -> {
log.info("【监听队列回调函数】- 入参,message: ${}$", new String(message.getBody(), StandardCharsets.UTF_8));
// 手动 Ack
// multiple: 为 true 是确认多条。为 false 是确认单条。建议 false,true 容易搞混。
channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
};
@BeforeEach
void init() throws Exception {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setUsername("admin");
connectionFactory.setPassword("kzh_mxg4vfb2QRP*xkv");
connection = connectionFactory.newConnection();
channel = connection.createChannel();
// 声明死信交换机
channel.exchangeDeclare(
// 交换机名称
EXCHANGE_DLX,
// 交换机类型,这里选择直接交换
BuiltinExchangeType.TOPIC,
// 持久,如果为 false,重启后交换机就没了
true,
// 自动删除,如果为 true,没有队列依赖此交换机,交换机就没了
false,
// 参数
null
);
// 声明死信队列
channel.queueDeclare(
// 队列名称
QUEUE_DLX,
// 持久,如果为 false,重启后队列就没了
true,
// 是否为当前 connection 独占,如果为 true,那其他 connection 就无法连接
false,
// 自动删除,如果为 true,队列中没有消息之后,队列就没了
false,
// 参数
null
);
// 绑定死信 Exchange 和死信 Queue
channel.queueBind(
QUEUE_DLX,
EXCHANGE_DLX,
KEY_DLX
);
// 声明交换机 Exchange
channel.exchangeDeclare(
// 交换机名称,名称的规则:什么类型.交换者A.交换者B
EXCHANGE_NAME,
// 交换机类型,这里选择直接交换
BuiltinExchangeType.DIRECT,
// 持久,如果为 false,重启后交换机就没了
true,
// 自动删除,如果为 true,没有队列依赖此交换机,交换机就没了
false,
// 参数
null
);
// 配置声明队列时使用的参数
Map<String, Object> args = new HashMap<>(1);
// 设置死信队列指向的交换机
args.put("x-dead-letter-exchange", EXCHANGE_DLX);
// 声明队列
channel.queueDeclare(
// 队列名称,名称的规则:什么类型.干什么的
QUEUE_NAME,
// 持久,如果为 false,重启后队列就没了
true,
// 是否为当前 connection 独占,如果为 true,那其他 connection 就无法连接
false,
// 自动删除,如果为 true,队列中没有消息之后,队列就没了
false,
// 参数
args
);
// 绑定 Exchange 和 Queue
channel.queueBind(
QUEUE_NAME,
EXCHANGE_NAME,
KEY_NAME
);
// 消费者监听,关闭自动 ACK
// channel.basicConsume(QUEUE_NAME, false, deliverCallback, (consumerTag) -> {});
}
@AfterEach
void destroy() throws Exception {
channel.close();
connection.close();
}
/**
* 消息发送
*/
@Test
void publisher() throws Exception {
// 过期时间 15 秒
AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().expiration("15000").build();
// AMQP.BasicProperties basicProperties = null;
channel.basicPublish(EXCHANGE_NAME, KEY_NAME, basicProperties, "一只英短猫来了".getBytes(StandardCharsets.UTF_8));
TimeUnit.SECONDS.sleep(3);
// TimeUnit.MINUTES.sleep(3);
}
}
文章评论