消费者系统拿到消息就一定会消费吗？
如果消费者系统已经拿到了这条消息，但是消息目前还在他的内存里，还没执行消费的逻辑，此时他就直接提交了这条消息的offset到broker去说自己已经处理过了。接着消费者系统在这个时候直接崩溃了，内存里的消息就没了，消息也没消费成功，结果Broker已经收到他提交的消息offset了，还以为他已经处理完这条消息了。等消费者系统重启的时候，就不会再次消费这条消息了。RocketMQ消费者的与众不同的地方
RocketMQ的消费者中会注册一个监听器，就是MessageListenerConcurrently这个东西，当你的消费者获取到一批消息之后，就会回调你的这个监听器函数，让你来处理这一批消息。
然后当你处理完毕之后，你才会返ConsumeConcurrentlyStatus.CONSUME_SUCCESS作为消费成功的示意，告诉RocketMQ，这批消息我已经处理完毕了。所以对于RocketMQ而言，其实只要你的消费者系统是在这个监听器的函数中先处理一批消息，基于这批消息都处理完成，然后返回了那个消费成功的状态，接着才会去提交这批消息的offset到broker去。所以在这个情况下，如果你对一批消息都处理完毕了，然后再提交消息的offset给broker，接着消费者系统崩溃了，此时是不会丢失消息的。
那么如果是消费者系统获取到一批消息之后，还没处理完，也就没返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS这个状态呢，自然没提交这批消息的offset给broker呢，此时消费者系统突然挂了，会怎么样？其实在这种情况下，你对一批消息都没提交他的offset给broker的话，broker不会认为你已经处理完了这批消息，此时你突然消费者系统的一台机器宕机了，他其实会感知到你的消费者系统的一台机器作为一个Consumer挂了。接着他会把你没处理完的那批消息交给消费者系统的其他机器去进行处理，所以在这种情况下，消息也绝对是不会丢失的。需要警惕的地方：不能异步消费消息
在默认的Consumer的消费模式之下，必须是你处理完一批消息了，才会返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS这个状态标识消息都处理结束了，去提交offset到broker去。在这种情况下，正常来说是不会丢失消息的，即使你一个Consumer宕机了，他会把你没处理完的消息交给其他Consumer去处理。
但是这里我们要警惕一点，就是我们不能在代码中对消息进行异步的处理，如果我们开启了一个子线程去处理这批消息，然后启动线程之后，就直接返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS状态了，如果要是用这种方式来处理消息的话，那可能就会出现你开启的子线程还没处理完消息呢，你已经返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS状态了，就可能提交这批消息的offset给broker了，认为已经处理结束了。然后此时你消费者系统突然宕机，必然会导致你的消息丢失了！
因此在RocketMQ的场景下，我们如果要保证消费数据的时候别丢消息，你就老老实实的在回调函数里处理消息，处理完了你再返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS状态表明你处理完毕了！消息处理失败场景下的方案
消费者底层的一些依赖可能有故障了，比如数据库宕机，缓存宕机之类的，此时你就没办法完成消息的处理了，那么可以通过一些返回状态去让消息进入RocketMQ自带的重试队列，同时如果反复重试还是不行，可以让消息进入RocketMQ自带的死信队列，后续针对死信队列中的消息进行单独的处理就可以了。
若有收获，就点个赞吧
0 人点赞
