回答面试官：如何保证消息不丢失

比眉伴天荒 2022-08-28 02:58 254阅读 0赞

> rocketmq是阿里开源的一个性能很强大的消息队列，很多公司都在用，而且经历了多次双十一的洗礼，支持多种特性

对于这个技术点不知道大家掌握的如何了，消息队列现在应该是公司必备的技能之一了，无论是RabbitMQ还是rocketmq，或者支持大数量的kafka

**今天我们要说的一个问题，是rocketmq如何保证消息的不丢失？？**

**![44541ef04f4d260c48b350ae8b5f93a9.png][]**

不知道大家对于这个问题遇到过没有，或者大家听到这个问题的第一反应是什么，应该如何做，如何避免消息丢失，一起来看看

首先我们知道rocketmq的一个消息从生产到最终的消费过程需要经历总共三个阶段，或者说会经过三个地方，分别是**producer的发送端、broker的持久化机制、以及consumer的消费端**

![8f1b6a115fd87c206f4b95b4f877e0e5.png][]

**从生产者producer的角度：消息生产之后传递到broker，如果消息未能正确的存储到broker中，算作消息丢失**

**从broker的角度：消息默认保存到broker的内存中，异步保存到磁盘上，如果发生宕机、磁盘崩溃会造成消息丢失**

**从消费者consumer的角度：消息完成了持久化之后，consumer拉取之后未能成功消费且未反馈给broker，这样算作消息丢失，可能消费过程异常或者网络抖动造成消息丢失**

> 生产者角度：消费生产之后传递到broker，如果消息未能正确的保存到broker中，算作消息丢失

从生产者的角度，生产了消息就是要通过网络发送到broker，其实只需要保证一点，就是确认这个消息已经成功发送到broker上了

生产者只需要接收发送消息返回的确认响应即可，就可以代表消息发送成功

**代码示例：**

DefaultMQProducer mqProducer=new DefaultMQProducer("test");
    // 设置 nameSpace地址
    mqProducer.setNamesrvAddr("namesrvAddr");
    mqProducer.start();
    Message msg = new Message("topic" /* Topic */,
            "Captain".getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
    );
    // 发送消息到Broker
    try {
        SendResult sendResult = mqProducer.send(msg);
    } catch (Exception e) {
        e.printStackTrace();
    }

当然，发送消息也分为同步和异步两种，消息发送成功之后会返回下面这四种不同的响应状态

**SEND\_OK**

消息发送成功，但是也并不意味这完全代表不会丢失消息，这还要取决于broker的刷盘方式

这个下面在broker方面会说，需要启动SYNC\_MASTER或SYNC\_FLUSH。（也就是同步）

**FLUSH\_DISK\_TIMEOUT**

如果Broker设置MessageStoreConfig的FlushDiskType = SYNC\_FLUSH（默认为ASYNC\_FLUSH），并且Broker没有在MessageStoreConfig的syncFlushTimeout（默认为5秒）内完成刷新磁盘，得到这个状态

也就是此时刷盘超时，未能在规定时间内落到硬盘上，检查设置是否合理、硬盘大小等情况

**FLUSH\_SLAVE\_TIMEOUT**

如果Broker的角色是SYNC\_MASTER（默认为ASYNC\_MASTER），并且从属Broker未在MessageStoreConfig的syncFlushTimeout（默认为5秒）内完成与主服务器的同步，返回这个结果

主从同步时间默认也是5秒，需要完成主从同步，这个下面在说broker的时候也会说到，你想啊，要是master挂了或者磁盘崩溃了，这是不是也不能百分百保证消息不丢失

**SLAVE\_NOT\_AVAILABLE**

如果Broker的角色是SYNC\_MASTER（默认为ASYNC\_MASTER），但没有配置slave Broker，返回这个状态

这个在保证消息绝对不丢失的情况下是不允许存在的，这个状态也是属于需要处理的，没有可靠的slave，也就意味着没有可靠的备份数据，所以这种情况也需要考虑

![9c3276e2dec40d5e7dce1fd372b8d78c.png][]

另外呢，上面还说了一种异步的消息发送方式，这种一般是用于链路较长，对于时间比较敏感的业务

这种情况下需要特别注意的就是我们需要设置消息发送完成的回调，这样才能更好的保证消息不丢失

采取事务消息的投递方式，并不能保证消息100%投递成功到了Broker，但是如果消息发送Ack失败的话，此消息会存储在CommitLog当中，但是对ConsumerQueue是不可见的

可以在日志中查看到这条异常的消息，严格意义上来讲，也并没有完全丢失

> broker：消息默认保存到broker的内存中，异步保存到磁盘上，如果发生宕机、磁盘崩溃会造成消息丢失

顺序消费这个场景其实不是特别的常见，但是也是必不可少的，因为在某些业务场景下顺序是很关键的，保证消息的消费顺序也是很关键

消息到了broker之后，默认是优先保存到broker的内存中，然后立刻返回响应给生产者producer，然后broker自己定期将消息批量的异步的保存到硬盘上

有的小伙伴一小子就发现了问题不是那么简单，消息来了之后还没保存到硬盘，就直接返回了，broker直接宕机崩溃了，那这消息岂不无迹可寻了

**这样的优点是提高交互的效率，同时减少IO的次数，问题就是会造成消息丢失**

如果我们想要保证消息不丢失，那就需要保证消息成功保存到broker之后才可以返回，只需要将消息的保存机制修改为同步刷盘的方式，也就是只有消息保存到broker的磁盘成功之后，才会返回响应

## 默认情况为 ASYNC_FLUSH 
    flushDiskType = SYNC_FLUSH

如果broker未能在规定的同步时间（默认5秒）完成刷盘，将返回**FLUSH\_DISK\_TIMEOUT**给生产者

上面也介绍了这个了**FLUSH\_DISK\_TIMEOUT**了

一般在系统中为了保证可用性，broker通常采用的都是一主master多从slave的部署方式，属于**集群部署**

为了保证消息不丢失，消息需要复制到slave节点，其实默认的情况下，消息写入到broker之后就会返回成功

**但是！如果master突然宕机或者磁盘崩溃了，那么这个消息就彻底丢失了，没有备份，所以呢，这里还需要把master和slave的异步复制改成同步复制**

## master 节点配置
    flushDiskType = SYNC_FLUSH
    brokerRole=SYNC_MASTER
    
    
    ## slave 节点配置
    brokerRole=slave
    flushDiskType = SYNC_FLUSH

**也就是只有slave也刷盘到磁盘成功之后，才会给producer返回成功**

当然你要这里说，master和slave也可能同时宕机，同时磁盘崩溃，那最终还是无法满足百分百保证消息的不丢失

![cab4adcbcc88781e2769eef9f752f2eb.png][]

这种问题啊，其实就像是TCP的三次交互一样，三次交互之后一定保证客户端和服务端通信成功了吗，答案是不一定

**我们只能在有限的资源下尽量的去满足系统的稳定性**

> consumer：消息完成了持久化之后，consumer拉取之后未能成功消费且未反馈给broker，这样算作消息丢失，可能消费过程异常或者网络抖动造成消息丢失

消费者从broker拉取消息，然后进行相应的业务的消费，消费成功会返回一个消费成功的状态给broker，broker如果没收到确认信息，消费者下次拉取重新拉取该消息

// 注册回调实现类来处理从broker拉取回来的消息
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            // 标记该消息已经被成功消费
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });

**consumer自身可以维护一个持久化的offset**，对应MessageQueue里面的min offset，标记已经成功消费或者已经成功发回到broker的消息下标

如果consumer消费失败，会把这个消息发回给broker，发回成功后，更新自己的offset

如果发回给broker时，broker挂掉了，那么consumer也会定时重试这个操作

即使consumer和broker一起挂掉了，消息也不会丢失，因为consumer里面的offset会定时持久化，重启之后，继续拉取offset之前的消息到本地，重新消费

有道无术，术可成；有术无道，止于术

欢迎大家关注**Java之道**公众号

![93cfee175373ddfbacda8084618b5b93.png][]

好文章，我在看❤️

[44541ef04f4d260c48b350ae8b5f93a9.png]: /images/20220828/9da677207af748589158c26381b54582.png
[8f1b6a115fd87c206f4b95b4f877e0e5.png]: /images/20220828/df79eefb5715402eb7f386dc85786abf.png
[9c3276e2dec40d5e7dce1fd372b8d78c.png]: /images/20220828/cbe3fb1266d147d685dc3d1b37fc75ee.png
[cab4adcbcc88781e2769eef9f752f2eb.png]: /images/20220828/fd2a4564df714bbea6bad76f8d1e432f.png
[93cfee175373ddfbacda8084618b5b93.png]: /images/20220828/0c1815f29cbe4e06a8600dea78ec6d79.png