MQ如何保证消息不丢失

时间：2023-02-03 15:55:06

如何保证消息不丢失

哪些环节会造成消息丢失

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

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

        MQ如何保证消息不丢失

    • 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
标签：

相关文章