作者 | 敖丙
来源 | 敖丙
什么叫可靠性？
大家都知道，系统架构有三高：「高性能、高并发和高可用」，三者的重要性不言而喻。
对于任意系统，想要同时满足三高都是一件非常困难的事情，大型业务系统或者传统中间件都会搭建复杂的架构来保证。
除以上三种模式之外，还有一个指标方向也很重要，那就是高可靠，甚至你可能会将它和「高可用」混淆起来。
事实上两者并不一样，高可用会更偏向于整体服务的可用性，防止系统宕机等等。而高可靠是指数据的可靠性保证嘛，你可以理解”高可靠“相比于系统三高会是一个更细一点的概念。
那么什么是数据的高可靠呢，总结一下就是系统要提供可靠的数据支撑，不能发生丢失、重复等错误现象。
所以每个开源中间件在发布版本时都会通过文档声明自己是超可靠的。
咱今天的主角kafka就是这么一个例子。
一些重要概念
因为有一段时间没讲消息队列了嘛，为了帮助你更好理解文章，我们来先复习一下kafka的基础概念：
record：消息，消息队列基础通信单位
topic：主题，目的就是将消息进行分类，不同业务类型的消息通常会被分发到不同的主题
partition：分区，每个主题可以创建多个分区，每个分区都由一系列有序和不可变的消息组成
replica：副本，每个分区都有一个至多个副本存在，它的主要作用是存储保存数据，以日志(Log)对象的形式体现。副本又分为leader副本和follower副本
offset：偏移量，每一个消息在日志文件中的位置都对应一个按序递增的偏移量，你可以理解为类似数组的存储形式
producer：生产者，生产消息的那一方
consumer：消费者，通常不同的业务都会有一到多个消费者组成消费者集群
broker：代理，一个Kafka集群由一个或多个Kafka实例构成，每一个Kafka实例就称为代理
如上图所示，一共存在主题1和主题2，主题1有两个分区，主题2只有一个分区，并且每个分区都存在一个leader副本和两个follower副本，它们分布在每个不同的代理节点上。
partition里只有leader副本负责与生产者、消费者之间数据的交互，follower副本会定期从leader副本拉取数据以保证整个集群数据可用性。
如何保证数据高可靠
Kafka是通过副本机制实现数据的存储的，所以就需要一些机制保证数据在跨集群的副本之间能够可靠地传输。
1.副本同步集合
业务数据封装成消息在系统中流转，由于各个组件都是分布在不同的服务器上的，所以主题和生产者、消费者之间的数据同步可能存在一定的时间延迟，Kafka通过延迟范围划分了几个不同的集合：
AR(Assigned Replicas)
指的是已经分配数据的分区副本，通常指的是leader副本 + follower副本。
ISR(In Sync Replicas)
指的是和leader副本数据保持同步的副本集合。当follower副本数据和leader副本数据保持同步，那么这些副本就处在ISR里面，ISR集合会根据数据的同步状态动态变化。
OSR(Out Sync Replicas)
一旦follower副本的数据同步进度跟不上leader了，那么它就会被放进叫做OSR的集合里。也就是这个集合包含的是不处于同步状态的分区副本。
OK，那有什么标准判断它是同步还是不同步呢？
通过replica.lag.time.max.ms这个参数来设置数据同步时间差，它的默认值是10s。
一旦从分区副本和主分区副本的消息相差10s以上，那么就认为消息处于OSR不同步的状态。若follower处于OSR集合里，那么在选取新的leader的时候就不会选举它作为新leader。
2.ACK应答机制
我们刚刚说了kafka是通过ack来发送数据同步信号的，那信号发送频率又有几种设定呢？
ack = 0
生产者发送一次消息就不再发送。不管是否发送成功，若发出去的消息处于通信的路上就丢失，或者还未做磁盘持久化操作，那么消息就可能丢失。
它的好处就是性能很高，你想呀你发送消息都不需要等待对方回复就持续发送下一批，那么消息等待的时间就节省出来了。同一时间范围内能比别人处理更多数据，缺点就是它的可靠性真的很低，数据真的是说丢就丢。
ack = 1
leader接收到消息并且写入到本地磁盘后就认为消息处理成功。这种方式可靠性会比上一种好一些，当leader接收到消息并且写入到本地磁盘后就认为消息处理成功，不论follower是否同步完这条消息就会返回给producer。
但是假如此刻partition leader所在的broker宕机了，如果那么数据也可能会丢失，所以follower副本的数据同步就很重要。
Kafka默认就采用这种方式。
ack = -1
producer只有收到分区内所有副本的响应ACK才会认为消息已经push成功。
这种方式虽然对于数据的可靠保障做得很好，但是就是性能很差，影响吞吐量，所以一般也不会采取。
那么它就绝对可靠吗？也不一定。最重要的还是取决于副本数据是否同步完成。若producer收到响应消息前leader副本挂掉，那么producer会因未收到消息重复发送消息，那就可能造成数据重复。怎么解决呢？只要保证业务幂等就行。
我们可以通过request.required.acks这个参数控制消息的发送频率。
3.消息语义
消息集群整体是一个复杂的系统，所以过程中可能会因为各种原因导致消息传递出错，Kafka对于这些可能遇到的场景定义了对应的的消息语义。
at most once
它代表消息可能被消费者消费0次或者1次。若场景如下：
消息从partition分发给消费者集群
消费者把自己收到的消息告诉集群，集群收到之后offset就会往后移动
消费者将数据入库做持久化
你一定想到了。在第三步消费者将消息入库时若因任何原因消费者A挂了，那么在将消费者切换到集群的消费者B后，数据还没入库呢。此时partition是浑然不知的呀，那么这就会造成一个问题：数据丢失。
at least once
它代表partition分发的消息至少被消费一次。其通信过程如下：
消息从partition分发给消费者集群
消费者将数据入库做持久化
消费者把自己收到的消息告诉集群，集群收到之后offset就会往后移动
假设consumer group在数据入库之后，在将数据返回给partition的过程中消费者A挂了，那么partition会因为接收不到响应ACK而重新发送数据，此时消费者B可能再次将原先的消息入库，这就造成了数据重复了。
在没有做任何幂等性保护的情况下，像重复转账，重付叠加积分这种业务，那么结果可能是致命的。
exactly once
代表消息正好能被消费一次，不丢失，不重复。
在at least once的情况基础上，假设consumerA在返回ack给partition的过程中宕机了。那么consumerB不会跟着partition的offset走，它会先去数据库里面查看最新消息对应的偏移位，再根据这个偏移位返回Kafka集群从对应的偏移位置出发，这就可以避免消息重复和消息丢失。
4.数据截断机制
我们开头说了真正处理数据的是leader副本，follower副本只负责数据的同步和保存，那如果因为leader宕机了二者数据不一致会怎么样呢？
在讲一致性保证过程之前还需了解两个Kafka用于表示副本数据同步的概念：
HW（High Watermark）：中文翻译为高水位，用来体现副本间数据同步的相对位置，consumer最多只能消费到HW所在的位置，通过HW我们可以判断数据对副本是否可见。
LEO（Log End Offset）：下一条待写入消息的记录位置。
leader副本从生产者获取消息，follower副本实时从leder同步数据，此时它们的同步数据是一致的都同步到2这个位置，并且下一个写入的消息都是偏移位4：
假设因为意外leader发生宕机，follower即被选为新leader，此后从生产者写入最新的偏移位4和5：
过了一段时间原leader通过修复恢复服务，它就会发现自己和新leader的数据是不一致的：
为了保证数据一致性就必须强行让一方妥协。因为数据是不断在刷新的，所以旧leader此时的优先级会小于新leader，因此它会将自己的数据截断到与新leader相同的HW和LEO位置，确保和新leader的数据一定相同，这就是Kafka数据截断机制。
5.数据清理机制
同其它中间件一样，Kafka的主要作用是通信，所以即使是将数据保存在磁盘上它还是会占用一定空间。为了节约存储空间它会通过一些机制对过期数据进行清理。
日志删除
日志删除会直接删除日志分段，kafka会维护一个定时任务来周期性检查和删除「过期数据」。
基于时间的日志删除
它在每一个日志段文件里面都维护一个最大时间戳来确认当前配置的删除时间，只要日志段写入新消息该字段都会被更新。一个日志段被写满了之后就不会再接收新的消息，它会去创建一个新的日志段文件往里面写数据。
每一个日志段文件被写满之后它的最大的时间戳都是保持不变的，Kafka只要通过当前时间与最大时间戳进行比较就可以判断该日志段文件是否过期。
Kafka默认配置log.retention.hours = 168，也就是7天的日志保留时间。
基于容量大小的日志删除
这和以上是异曲同工的方式， 只不过这次从时间换成了空间。
Kafka会通过每个日志段空间的大小计算一个总容量阈值，然后计算出当前的实际空间大小和总容量阈值的差值，如果这个差值大于单个日志段文件的大小那么就会删除掉最旧的那个日志段文件，反之则不做任何处理。
同理，这个阈值也可以通过log.retention.bytes参数来设置。
日志压缩
Kafka的消息是由键值组成的，如果日志段里存在多条相同key但是不同value的数据，那么它会选择性地清除旧数据，保留最近一条记录。
具体的压缩方式就是创建一个检查点文件，从日志起始位置开始遍历到最大结束位置，然后把每个消息的key和key对应的offset保存在一个固定容量的SkimpyOffsetMap中。
这样前面的值就会被后面的覆盖掉，如果日志文件里存在相同的key只有最新的那个会被保留。
总结
Kafka通过ACK应答机制保证了不同组件之间的通信效率，通过副本同步机制、数据截断和数据清理机制实现了对于数据的管理策略，保证整个系统运行效率。
作为一款高性能又同时兼顾高可靠性的消息中间件来说，Kafka能吹的点实在太多。如果本篇文章对你有所帮助，点击一下右下角的大拇指，下一次我们来详细讲解Kafka是如何实现副本间数据传递的。
往期推荐
Redis 6 中的多线程是如何实现的！？
快速入门 Docker 的五种网络模式
Redis 内存满了怎么办？这样置才正确！
Kubernetes 弃用 Docker ，我们该怎么办？
点分享
点收藏
点点赞
点在看
