java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > RabbitMQ可靠性方案

RabbitMQ之消息的可靠性方案详解

作者:王廷云的博客

这篇文章主要介绍了RabbitMQ之消息的可靠性方案详解,MQ 消息数据完整的链路为:从 Producer 发送消息到 RabbitMQ 服务器中,再由 Broker 服务的 Exchange 根据 Routing_Key 路由到指定的 Queue 队列中,最后投送到消费者中完成消费,需要的朋友可以参考下

一、数据丢失场景

MQ 消息数据完整的链路为:从 Producer 发送消息到 RabbitMQ 服务器中,再由 Broker 服务的 Exchange 根据 Routing_Key 路由到指定的 Queue 队列中,最后投送到消费者中完成消费。

在这里插入图片描述

所以消息在上面三个节点都可能存在消息丢失的情况:

二、数据可靠性方案

上面已经了解到了消息数据可能丢失的环节,所以,我们需要针对每个环节进行处理,以防止数据的丢失。

1、生产者丢失消息解决方案

对于生产者消息丢失的问题,我们有常用的两种方案:

1-1、开启消息发送事务功能

我们可以选择使用 RabbitMQ 提供的事务功能:生产者在发送数据之前开启事物,然后再发送消息。

如果消息没有成功被 RabbitMQ 接收到,那么生产者会受到异常报错,这时就可以回滚事务,然后尝试重新发送;如果收到了消息,那么就可以提交事务。

伪代码如下:

channel.txSelect();// 开启事物
try{
	...
}catch(Exection e){
    channel.txRollback();// 回滚事物
    // 重新提交
}

这种方案有个比较大的缺点:RabbitMQ 事务一旦开启,就会变为同步阻塞操作,生产者会阻塞等待是否发送成功,由于比较耗性能而会造成吞吐量的下降。所以并不推荐这种方案。

1-2、开启 Confirm 消息确认机制

在生产者中开启了Confirm 模式,为每次写的消息分配一个唯一的 ID,然后再发送给 RabbitMQ 服务

同时也可以结合这个机制知道自己在内存里维护每个消息的 ID,如果超过一定时间还没接收到这个消息的回调,那么可以尝试进行重发。

伪代码如下:

//开启confirm
channel.confirm();
//发送成功回调
public void ack(String messageId){
}
// 发送失败回调
public void nack(String messageId){
    //重发该消息
}

由于事务机制是同步阻塞的,而 Confirm 机制是异步的,在发送消息之后可以接着发送下一个消息,最后通过 RabbitMQ 的回调告知成功与否,所以,生产者消息丢失方案一般都是采用 Confirm 确认机制。

2、MQ 队列丢失消息解决方案

对于 MQ 队列丢失消息的问题,我们可以开启消息的持久化,当然队列本身也要开启持久化,毕竟队列如果不存在了,哪怕消息持久化也没有用。

开启了消息队列的持久化后,可以将消息的持久化和生产者的 Confirm 机制配合起来,只有消息持久化到了磁盘,才会个生产者发送 ACK,这样就算是在持久化之前 RabbitMQ 挂了,数据丢了,生产者收不到 ACK 回调也会进行消息重发。

持久化有个关键的问题需要注意:

消息在正确存入 RabbitMQ 之后,还需要有一段时间(这个时间很短,但不可忽视)才能存入磁盘之中。因为 RabbitMQ 并不是为每条消息都做 fsync 的处理,可能仅仅保存到 cache 中而不是物理磁盘上,在这段时间内 RabbitMQ 的 broker 发生 crash,消息保存到 cache 但是还没来得及落盘,那么这些消息将会丢失。

解决这个问题的方案是 RabbitMQ 开启镜像队列,镜像队列相当于配置了副本,当 master 在此特殊时间内 crash 掉,可以自动切换到 slave,这样有效地保障了数据的丢失。

3、消费者丢失消息解决方案

针对消费者丢失消息问题,我们可以使用 RabbitMQ 提供的 ACK 应答机制,首先需要将 自动应答标志位 autoAck 设置为 false 来关闭 RabbitMQ 的自动ack,这是为了防止 Consumer 收到消息后,还没来得及处理完成就 crash 掉了。所以我们采用手动应答的方式:

String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

然后在消费者执行完毕之后手动应答: channel.basicAck

总结

RabbitMQ 消息的可靠性涉及 producer 端的确认机制、broker 服务的持久化与镜像队列的配置、consumer 端的确认机制。

要想确保消息的可靠性越高,那么性能也会随之而降,所以需要根据实际情况进行选择和取舍。

到此这篇关于RabbitMQ之消息的可靠性方案详解的文章就介绍到这了,更多相关RabbitMQ可靠性方案内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文