MQ如何保证消息不丢失

如何保证消息不丢失

哪些环节会造成消息丢失

其实主要就是跨网络的环境中需要考虑消息的丢失主要是有以下几个方面

• 生产者往MQ发送消息
• MQ的Broker是集群有主从的主节点把消息同步到从节点时也需要考虑消息丢失问题
• 消息从内存持久化到硬盘时MQ的消息是工作在内存中的但是内存是断电就丢失数据所以需要持久化到磁盘这一步也需要考虑消息丢失问题
• 消费者消费MQ的消息

如下图所示的四个步骤都有可能造成消息丢失

如何去防止消息丢失

其实也就是针对上面四个环节来分析保证每个环节的消息不丢失

• 生产者发送消息不丢失

  • kafka消息发送+回调。生产者向MQ发送一个消息之后MQ会向生产者发送一个请求执行相应的回调函数如果一直没有执行回调函数Produce就知道消息发送失败了就可以重新发送消息

  • RocketMQ它是Kafka之后出来的也支持消息发送+回调的机制。同时它还支持事务消息来保证生产者发送消息不丢失

    RocketMQ的事务消息是保证生产者 本地事务和发送消息两步是原子性的

    1. Producer向MQ发送一个half消息对于Consumer是不可见的首先确定MQ是正常运行的

    2. 执行本地事务

    3. 向MQ发送真正的消息并携带本地事务的执行结果

    4. 如果本地事务执行结果是成功那么消费者就可以消费此消息如果本地事务执行结果是失败那么MQ就会丢弃此消息如果本地事务执行结果是未知那么就会经过以下的步骤

    5. 过一段时间MQ向Producer发送一个请求回查本地事务的执行结果

    6. Producer会执行相应的操作去检查当前 本地事务的执行结果并将结果发送给MQ

    7. 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