互联网面试必杀：如何保证消息中间件全链路数据100%不丢失

Posted 2021-02-02 jajian

背景引入

这篇文章，我们来聊聊在线上生产环境使用消息中间件技术的时候，从前到后的全链路到底如何保证数据不能丢失。

这个问题，在互联网公司面试的时候高频出现，而且也是非常现实的生产环境问题。

如果你的简历中写了自己熟悉MQ技术（RabbitMQ、RocketMQ、Kafka），而且在项目里有使用的经验，那么非常实际的一个生产环境问题就是：投递消息到MQ，然后从MQ消费消息来处理的这个过程，数据到底会不会丢失。

面试官此时会问：如果数据会丢失的话，你们项目生产部署的时候，是通过什么手段保证基于MQ传输的数据100%不会丢失的？麻烦结合你们线上使用的消息中间件来具体说说你们的技术方案。

这个其实就是非常区分面试候选人技术水平的一个问题。

实际上相当大比例的普通工程师，哪怕是在一些中小型互联网公司里工作过的，也就是基于公司部署的MQ集群简单的使用一下罢了，可能代码层面就是基本的发送消息和消费消息，基本没考虑太多的技术方案。

但是实际上，对于MQ、缓存、分库分表、NoSQL等各式各类的技术以及中间件在使用的时候，都会有对应技术相关的一堆生产环境问题。

那么针对这些问题，就必须要有相对应的一整套技术方案来保证系统的健壮性、稳定性以及高可用性。

所以其实中大型互联网公司的面试官在面试候选人的时候，如果考察对MQ相关技术的经验和掌握程度，十有八九都会抛出这个使用MQ时一定会涉及的数据丢失问题。因为这个问题，能够非常好的区分候选人的技术水平。

所以这篇文章，我们就来具体聊聊基于RabbitMQ这种消息中间件的背景下，从投递消息到MQ，到从MQ消费消息出来，这个过程中有哪些数据丢失的风险和可能。

然后我们再一起来看看，应该如何结合MQ自身提供的一些技术特性来保证数据不丢失？

目前已有的技术方案

经过之前几篇文章的讨论，目前我们已经初步知道，第一个会导致数据丢失的地方，就是消费者获取到消息之后，没有来得及处理完毕，自己直接宕机了。

此时RabbitMQ的自动ack机制会通知MQ集群这条消息已经处理好了，MQ集群就会删除这条消息。

那么这条消息不就丢失了么？不会有任何一个消费者处理到这条消息了。

所以之前我们详细讨论过，通过在消费者服务中调整为手动ack机制，来确保消息一定是已经成功处理完了，才会发送ack通知给MQ集群。

否则没发送ack之前消费者服务宕机，此时MQ集群会自动感知到，然后重发消息给其他的消费者服务实例。

《扎心！线上服务宕机时，如何保证数据100%不丢失？》这篇文章，详细讨论了这个问题，手动ack机制之下的架构图如下所示：

当时除了这个数据丢失问题之外，还有另外一个问题，就是MQ集群自身如果突然宕机，是不是会导致数据丢失？

默认情况下是肯定会的，因为queue和message都没采用持久化的方式来投递，所以MQ集群重启会导致部分数据丢失。

所以《消息中间件集群崩溃，如何保证百万生产数据不丢失？》这篇文章，我们分析了如何采用持久化的方式来创建queue，同时采用持久化的方式来投递消息到MQ集群，这样MQ集群会将消息持久化到磁盘上去。

此时如果消息还没来得及投递给消费者服务，然后MQ集群突然宕机了，数据是不会丢失的，因为MQ集群重启之后会自动从磁盘文件里加载出来没投递出去的消息，然后继续投递给消费者服务。

同样，该方案沉淀下来的系统架构图，如下所示：

数据100%不丢失了吗？

大家想一想，到目前为止，咱们的架构一定可以保证数据不丢失了吗？

其实，现在的架构，还是有一个数据可能会丢失的问题。

那就是上面作为生产者的订单服务把消息投递到MQ集群之后，暂时还驻留在MQ的内存里，还没来得及持久化到磁盘上，同时也还没来得及投递到作为消费者的仓储服务。

此时要是MQ集群自身突然宕机，咋办呢？

尴尬了吧，驻留在内存里的数据是一定会丢失的，我们来看看下面的图示。

按需制定技术方案

现在，我们需要考虑的技术方案是：订单服务如何保证消息一定已经持久化到磁盘？

实际上，作为生产者的订单服务把消息投递到MQ集群的过程是很容易丢数据的。

比如说网络出了点什么故障，数据压根儿没传输过去，或者就是上面说的消息刚刚被MQ接收但是还驻留在内存里，没落地到磁盘上，此时MQ集群宕机就会丢数据。

所以首先，我们得考虑一下作为生产者的订单服务要如何利用RabbitMQ提供的相关功能来实现一个技术方案。

这个技术方案需要保证：只要订单服务发送出去的消息确认成功了，此时MQ集群就一定已经将消息持久化到磁盘了。

我们必须实现这样的一个效果，才能保证投递到MQ集群的数据是不会丢失的。

需要研究的技术细节

这里我们需要研究的技术细节是：仓储服务手动ack保证数据不丢失的实现原理。

之前，笔者就收到很多同学提问：

仓储服务那块到底是如何基于手动ack就可以实现数据不丢失的？

RabbitMQ底层实现的细节和原理到底是什么？

为什么仓储服务没发送ack就宕机了，RabbitMQ可以自动感知到他宕机了，然后自动重发消息给其他的仓储服务实例呢？

这些东西背后的实现原理和底层细节，到底是什么？

还有一个最大的问题，就是生产者投递出去的消息，可能会丢失。

丢失的原因有很多，比如消息在网络传输到一半的时候因为网络故障就丢了，或者是消息投递到MQ的内存时，MQ突发故障宕机导致消息就丢失了。

针对这种生产者投递数据丢失的问题，RabbitMQ实际上是提供了一些机制的。

比如，有一种重量级的机制，就是事务消息机制。采用类事务的机制把消息投递到MQ，可以保证消息不丢失，但是性能极差，经过测试性能会呈现几百倍的下降。

所以说现在一般是不会用这种过于重量级的机制，而是会用轻量级的confirm机制。

但是我们这篇文章还不能直接讲解生产者保证消息不丢失的confirm机制，因为这种confirm机制实际上是采用了类似消费者的ack机制来实现的。

所以，要深入理解confirm机制，我们得先从这篇文章开始，深入的分析一下消费者手动ack机制保证消息不丢失的底层原理。

ack机制回顾

其实手动ack机制非常的简单，必须要消费者确保自己处理完毕了一个消息，才能手动发送ack给MQ，MQ收到ack之后才会删除这个消息。

如果消费者还没发送ack，自己就宕机了，此时MQ感知到他的宕机，就会重新投递这条消息给其他的消费者实例。

通过这种机制保证消费者实例宕机的时候，数据是不会丢失的。

再次提醒一下大家，如果还对手动ack机制不太熟悉的同学，可以回头看一下之前的一篇文章：《扎心！线上服务宕机时，如何保证数据100%不丢失？》。然后这篇文章，我们将继续深入探讨一下ack机制的实现原理。

ack机制实现原理：delivery tag

如果你写好了一个消费者服务的代码，让他开始从RabbitMQ消费数据，这时这个消费者服务实例就会自己注册到RabbitMQ。

所以，RabbitMQ其实是知道有哪些消费者服务实例存在的。

大家看看下面的图，直观的感受一下：

接着，RabbitMQ就会通过自己内部的一个“basic.delivery”方法来投递消息到仓储服务里去，让他消费消息。

投递的时候，会给这次消息的投递带上一个重要的东西，就是“delivery tag”，你可以认为是本次消息投递的一个唯一标识。

这个所谓的唯一标识，有点类似于一个ID，比如说消息本次投递到一个仓储服务实例的唯一ID。通过这个唯一ID，我们就可以定位一次消息投递。

所以这个delivery tag机制不要看很简单，实际上他是后面要说的很多机制的核心基础。

而且这里要给大家强调另外一个概念，就是每个消费者从RabbitMQ获取消息的时候，都是通过一个channel的概念来进行的。

大家回看一下下面的消费者代码片段，我们必须是先对指定机器上部署的RabbitMQ建立连接，然后通过这个连接获取一个channel。

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

而且如果大家还有点印象的话，我们在仓储服务里对消息的消费、ack等操作，全部都是基于这个channel来进行的，channel又有点类似于是我们跟RabbitMQ进行通信的这么一个句柄，比如看看下面的代码：

另外这里提一句：之前写那篇文章讲解手动ack保证数据不丢失的时候，有很多人提出疑问：为什么上面代码里直接是try finally，如果代码有异常，那还是会直接执行finally里的手动ack？其实很简单，自己加上catch就可以了。

好的，咱们继续。你大概可以认为这个channel就是进行数据传输的一个管道吧。对于每个channel而言，一个“delivery tag”就可以唯一的标识一次消息投递，这个delivery tag大致而言就是一个不断增长的数字。

大家来看看下面的图，相信会很好理解的：

如果采用手动ack机制，实际上仓储服务每次消费了一条消息，处理完毕完成调度发货之后，就会发送一个ack消息给RabbitMQ服务器，这个ack消息是会带上自己本次消息的delivery tag的。

咱们看看下面的ack代码，是不是带上了一个delivery tag？

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

然后，RabbitMQ根据哪个channel的哪个delivery tag，不就可以唯一定位一次消息投递了？

接下来就可以对那条消息删除，标识为已经处理完毕。

这里大家必须注意的一点，就是delivery tag仅仅在一个channel内部是唯一标识消息投递的。

所以说，你ack一条消息的时候，必须是通过接受这条消息的同一个channel来进行。

大家看看下面的图，直观的感受一下。

其实这里还有一个很重要的点，就是我们可以设置一个参数，然后就批量的发送ack消息给RabbitMQ，这样可以提升整体的性能和吞吐量。

比如下面那行代码，把第二个参数设置为true就可以了。

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), true);

看到这里，大家应该对这个ack机制的底层原理有了稍微进一步的认识了。起码是知道delivery tag是啥东西了，他是实现ack的一个底层机制。

然后，我们再来简单回顾一下自动ack、手动ack的区别。

实际上默认用自动ack，是非常简单的。RabbitMQ只要投递一个消息出去给仓储服务，那么他立马就把这个消息给标记为删除，因为他是不管仓储服务到底接收到没有，处理完没有的。

所以这种情况下，性能很好，但是数据容易丢失。

如果手动ack，那么就是必须等仓储服务完成商品调度发货以后，才会手动发送ack给RabbitMQ，此时RabbitMQ才会认为消息处理完毕，然后才会标记消息为删除。

这样在发送ack之前，仓储服务宕机，RabbitMQ会重发消息给另外一个仓储服务实例，保证数据不丢。

RabbitMQ如何感知到仓储服务实例宕机

之前就有同学提出过这个问题，但是其实要搞清楚这个问题，其实不需要深入的探索底层，只要自己大致的思考和推测一下就可以了。

如果你的仓储服务实例接收到了消息，但是没有来得及调度发货，没有发送ack，此时他宕机了。

我们想一想就知道，RabbitMQ之前既然收到了仓储服务实例的注册，因此他们之间必然是建立有某种联系的。

一旦某个仓储服务实例宕机，那么RabbitMQ就必然会感知到他的宕机，而且对发送给他的还没ack的消息，都发送给其他仓储服务实例。

所以这个问题以后有机会我们可以深入聊一聊，在这里，大家其实先建立起来这种认识即可。

我们再回头看看下面的架构图：

仓储服务处理失败时的消息重发

首先，我们来看看下面一段代码：

假如说某个仓储服务实例处理某个消息失败了，此时会进入catch代码块，那么此时我们怎么办呢？难道还是直接ack消息吗？

当然不是了，你要是还是ack，那会导致消息被删除，但是实际没有完成调度发货。

这样的话，数据不是还是丢失了吗？因此，合理的方式是使用nack操作。

就是通知RabbitMQ自己没处理成功消息，然后让RabbitMQ将这个消息再次投递给其他的仓储服务实例尝试去完成调度发货的任务。

我们只要在catch代码块里加入下面的代码即可：

channel.basicNack(delivery.getEnvelope().getDeliveryTag(),  true);

注意上面第二个参数是true，意思就是让RabbitMQ把这条消息重新投递给其他的仓储服务实例，因为自己没处理成功。

你要是设置为false的话，就会导致RabbitMQ知道你处理失败，但是还是删除这条消息，这是不对的。

同样，我们还是来一张图，大家一起来感受一下：

阶段总结

这篇文章对之前的ack机制做了进一步的分析，包括底层的delivery tag机制，以及消息处理失败时的消息重发。

通过ack机制、消息重发等这套机制的落地实现，就可以保证一个消费者服务自身突然宕机、消息处理失败等场景下，都不会丢失数据。

来源：【微信公众号 - 石杉的架构笔记】