RabbitMQ如何保证消息的可靠?如RabbitMQ基础概念中的架构模型
可以看到一条消息的传递过程:
在这过程中以下几个环节可能会丢失消息:
如下图
所以要保证消息的可靠性需要做到以下几点:
如何做到以上几点?RabbitMQ为了适应各个场景的使用,以上的功能需要开发者按照定义自行设置实现。
以下是Java整合RabbitMQ的实现,参考Java整合RabbitMQ实现生产消费(7种通讯方式)
构建channel时添加确认监听机制,当消息未发送至交换机时做补偿措施。
channel.addConfirmListener((sequenceNumber, multiple) -> {
System.out.println("消息成功发送到交换机");
}, (sequenceNumber, multiple) -> {
System.err.println("消息未发送到交换机,补偿操作。");
});
channel.addReturnListener((replyCode, replyText, exchange, routingKey, basicProperties, body) -> {
System.err.format("消息 %s 未路由到指定队列: %s, replyText: %s,replyCode: %d%n", body, routingKey, replyText, replyCode);
});
channel.basicPublish("", "",true, "", "");
//构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除","其他参数")
channel.queueDeclare("", true, false, false, null);
//设置消息持久化
AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder().deliveryMode(2).build();
channel.basicPublish("", "",true, basicProperties, "");
当消息持久化至队列时已经保证了消息的可靠投递,为保证消息的正常消费,需要解决重复消费和消息丢失问题。
针对第一个场景的解决方案:设置手动ACK,并且业务处理和ack操作在一个事务中。
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 -> {
});
针对第二个场景的解决方案:发布消息时设置业务唯一标识,在消费后进行存储,如果有相同标识前来消费直接拒绝即可(具体业务具体分析)。设置业务唯一标识方式:
String correlationId = UUID.randomUUID().toString();
AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder().correlationId(correlationId).build();
channel.basicPublish("", "", basicProperties, "");
在以下场景中可能会发生消息丢失问题:
解决方案:设置手动ACK,并且业务处理和ack操作在一个事务中。
RabbitMQ 本身可以保证消息的可靠性,但是需要开发者去了解整体的流程,并且根据实际情况去自行保证。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。