面试经常会被问到“怎么避免RabbitMQ消息不丢失”。首先需要明确的是,RabbitMQ在默认情况并不会保证消息的不丢失。不过RabbitMQ提供了一些机制,可以有效的避免消息在传输和处理过程中不丢失。
先通过RabbitMQ的架构模型看一下消息的处理过程:
在这过程中以下几个环节可能会丢失消息:
如下图
很显然,在这样的处理过程中,要保证消息不丢失需要做到以下几点:
如何做到以上几点?
RabbitMQ为了适应各个场景的使用,以上的功能需要开发者按照定义自行设置。
下面以Java为例,看一下在代码中如何进行相关设置。
在构建channel时添加确认机制,通过确认机制可以得知Exchange是否接收到消息,当消息未发送至Exchange时可以进行补偿措施。
java 代码解读复制代码 channel.addConfirmListener((sequenceNumber, multiple) -> {
System.out.println("消息成功发送到交换机");
}, (sequenceNumber, multiple) -> {
System.err.println("消息未发送到交换机,补偿操作。");
});
java 代码解读复制代码channel.addReturnListener((replyCode, replyText, exchange, routingKey, basicProperties, body) -> {
System.err.format("消息 %s 未路由到指定队列: %s, replyText: %s,replyCode: %d%n", body, routingKey, replyText, replyCode);
});
java 代码解读复制代码 channel.basicPublish("", "",true, "", "");
channel.queueDeclare
声明的队列是非持久的,这意味着如果RabbitMQ服务器重启,该队列所有未处理的消息都会丢失。java 代码解读复制代码 //构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除","其他参数")
channel.queueDeclare("", true, false, false, null);
deliveryMode
值为2意味着消息持久化。java 代码解读复制代码//设置消息持久化
AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder().deliveryMode(2).build();
channel.basicPublish("", "",true, basicProperties, "");
当消息持久化至队列时,RabbitMQ已经保证了消息的可靠投递。
但是,为确保消费者的成功消费,消费端的确认机制通常被设置为手动确认模式,当消费者成功消费后向RabbitMQ发送确认信号,RabbitMQ才会从队列中删除该消息。
java 代码解读复制代码DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
//消息处理后手动ACK
channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
};
// ack为false
channel.basicConsume("", false, deliverCallback, consumerTag -> {
});
消息不丢失对于消息中间件来说是至关重要的,RabbitMQ通过消息的持久化、生产者确认、消费者确认等机制,有效地确保了消息的可靠性,避免了消息的丢失。
本文系转载,前往查看
如有侵权,请联系 cloudcommunity@tencent.com 删除。
本文系转载,前往查看
如有侵权,请联系 cloudcommunity@tencent.com 删除。