如何保证消息不丢失
哪些环节会造成消息丢失
其实主要就是跨网络的环境中需要考虑消息的丢失,主要是有以下几个方面
- 生产者往MQ发送消息
- MQ的Broker是集群有主从的,主节点把消息同步到从节点时也需要考虑消息丢失问题
- 消息从内存持久化到硬盘时,MQ的消息是工作在内存中的,但是内存是断电就丢失数据,所以需要持久化到磁盘,这一步也需要考虑消息丢失问题
- 消费者消费MQ的消息
如下图所示的四个步骤都有可能造成消息丢失
如何去防止消息丢失
其实也就是针对上面四个环节来分析,保证每个环节的消息不丢失
-
生产者发送消息不丢失
-
kafka:消息发送+回调。生产者向MQ发送一个消息之后,MQ会向生产者发送一个请求执行相应的回调函数,如果一直没有执行回调函数Produce就知道消息发送失败了就可以重新发送消息
-
RocketMQ:它是Kafka之后出来的,也支持消息发送+回调的机制。同时它还支持事务消息来保证生产者发送消息不丢失
RocketMQ的事务消息是保证生产者 本地事务和发送消息两步是原子性的:
-
Producer向MQ发送一个half消息(对于Consumer是不可见的),首先确定MQ是正常运行的
-
执行本地事务
-
向MQ发送真正的消息,并携带本地事务的执行结果
-
如果本地事务执行结果是成功,那么消费者就可以消费此消息;如果本地事务执行结果是失败,那么MQ就会丢弃此消息;如果本地事务执行结果是未知,那么就会经过以下的步骤
-
过一段时间MQ向Producer发送一个请求回查本地事务的执行结果
-
Producer会执行相应的操作去检查当前 本地事务的执行结果,并将结果发送给MQ
-
MQ再判断本地事务的执行结果,如果还是未知就继续重复执行5~7步,默认会重复执行15次。
-
-
RabbitMQ:
也支持消息发送+回调的机制。
它还支持手动事务机制。
RabbitMQ的提供了这些API方法,让我们程序自己去实现事务的逻辑
channel.txSelect()开启事务 channel.txCommit()提交事务 channel.txRollback()回滚事务
我们首先开启事务,然后执行本地事务,再执行后面两个Api方法。这种手动事务机制有一个问题就是对channel是会产生阻塞的,会造成吞吐量下降
在RabbitMQ3.*版本开始,它还支持Publisher Confirm机制,相当于是生产者确认机制,整个处理流程和RocketMQ事务消息的处理流程基本是一样的
-
-
MQ主从同步消息不丢失
-
RocketMQ
在RocketMQ中对于普通集群,主从数据复制有两种方式:同步复制和异步复制。同步复制就能保证消息不丢失,异步复制效率高但是可能丢消息
第二种方式就是Dledger集群,它在主从数据复制时采用两阶段提交来保证消息不丢失
普通集群就是我们指定哪个节点是Master,哪个节点是Slave;而Dledger集群会频繁的隔一段时间从至少三个节点中选举一个节点成为Master,其余的为Slave,当Producer发送消息到Master后,Master会直接返回给Producer,然后当前消息会标记为UnCommited,再给Slave进行消息同步,当大部分Slave都同步成功后才会把消息的状态改为Commited。这就是这里的两阶段。
-
RabbitMQ
普通集群:消息是分散存储在各个节点,节点之间不会主动进行消息同步,只有在消费时才会进行消息同步。就比如Producer生产一个消费发送到了A节点,这个时间集群中各个节点是不同步消息的,Consumer却在B节点上消费这条消息,这个时候才会把A节点的消息同步到B节点来 再进行消费。这种方式是可以丢失消息的
镜像集群:当Producer生产一个消息后,各个节点之间会主动的进行消息的同步,这样数据安全性会更高
-
Kafka
它通常是在允许少量消息丢失的场景,它可以通过配置acks,配置为0 1 all 。这就相当于RocketMQ的同步复制/异步复制
-
-
MQ消息持久化存盘时消息不丢失
- RocketMQ:提供了一种配置的方式,可以选择同步刷盘,也可以选择异步刷盘
- RabbitMQ:将队列配置成持久化队列,这样就可以保证消息不丢失。在RabbitMQ3.*版本中还有一个Quorum类型的队列,会采用Raft协议来进行消息同步
-
消费者消费消息时消息不丢失
一般情况下,MQ的队列中会有一个offset偏移量指向当前消费消息的位置,Consumer消费消息之后会往MQ返回一个消息,然后MQ就会把offset偏移量往前移动。如果consumer消费失败了,那么MQ的offset也不会移动,下次Consume再重新消费就行了。
会造成消费时消息丢失才场景是:Consumer消费消息变为异步的方式,刚开始收到需要消费的消息就往MQ发送一个消息,然后再去执行本地事务,这个时候如果本地事务执行失败,可是发送给MQ的确认消息却已经发送成功了,这就造成了消费端消息丢失。
解决消费者消息消息时不丢失的方式:
- RocketMQ:使用默认的消费方式就行,不要采用异步方式
- RabbitMQ:它在消费消息时有一个autocommit自动提交的机制,我们将autocommit关闭,不要让它自动提交,改为手动提交offset
- Kafka:也是一样的改为手动提交offset