浅谈MQ-kafka存储原理

墨蓝 2023-01-21 10:26 122阅读 0赞

# Kafka Broker 存储原理 #

## 1.文件的存储结构 ##

![image-20210505180153257][]

### 1.1 partition分区 ###

为了实现横向扩展，把不同数据存放在不同的Broker上，同时降低单台服务器的访问压力，我们把一个topic中的数据分隔成多个partition。

一个partition中的消息是有序的，顺序写入，但是全局不一定有序。

在服务器上，每个partition都有一个物理目录，topic名字后面的数字标号即代表分区。

### 1.2 replica副本 ###

为了提高分区的可靠性，kafka又设计了副本机制。

创建topic的时候，通过指定replication-factor确定topic的副本数。

注意：副本数必须小于等于节点数，而不能大于Broker的数量，否则会报错。

./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 4 --partitions 1 --topic overrep

这样就可以保证，绝对不会有一个分区的两个副本分布在同一个节点上，不然副本机制也失去了备份的意义了。

![image-20210505181207727][]

这些所有的副本分为两种角色，leader对外提供读写服务。follower唯一的任务就是从leader异步拉取数据。

**思考：为什么不能像MySQL一样实现读写分离?写操作都在leader上，读操作都在follower上。**

这个是设计思想的不同。读写都发生在leader节点，就不存在读写分离带来的一致性问题了，这个叫做单调读一致性。

### 1.3 如何区分leader ###

问题来了，如果分区有多个副本，哪一个节点上的副本是leader呢？

怎么查看所有副本中谁是leader？

sh ./kafka-topics.sh --topic businessMessage --describe --zookeeper localhost:2181

Topic: businessMessage PartitionCount: 3 ReplicationFactor: 3 Configs:  
Topic: businessMessage Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,0,2  
Topic: businessMessage Partition: 1 Leader: 2 Replicas: 2,0,1 Isr: 0,1,2  
Topic: businessMessage Partition: 2 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2

解释：这个topic有3个分区3个副本。  
第一个分区的3个副本编号0，1，2（代表Broker的序号），同步中的是0，1，2。第二个副本是leader。

### 1.4 副本在Broker的分布 ###

实际上，分配策略是由Admin Utils.scala的assign Replicas To Brokers函数决定的。

规则如下：

1.  fir to fall，副本因子不能大于Broker的个数；
2.  第一个分区(编号为0的分区) 的第一个副本放置位置是随机从broker List选择的；
3.  其他分区的第一个副本放置位置相对于第0个分区依次往后移。
    
    也就是说：如果我们有5个Broker，5个分区，假设第1个分区的第1个副本放在第四个Broker上，那么第2个分区的第1个副本将会放在第五个Broker上； 第三个分区的第1个副本将会放在第一个Broker上； 第四个分区的第1个副本将会放在第二  
    个Broker上，依次类推；
4.  每个分区剩余的副本相对于第1个副本放置位置其实是由next Replica Shift决定的，而这个数也是随机产生的。  
    ![image-20210505184006329][]

这样设计可以提高容灾能力。怎么讲?

在每个分区的第一个副本错开之后，一般第一个分区的第一个副本(按Broker编号排序) 都是leader。leader是错开的，不至于一挂影响太大。

bin目录下的kafka-reassign-partitions.sh可以根据Broker数量变化情况重新分配分区。

### 1.5 Segment（分段） ###

为了防止log不断追加导致文件过大，导致检索消息效率变低，一个partition又被划分成多个segment来组织数据(MySQL也有segment的逻辑概念，叶子节点就是数据段，非叶子节点就是索引段)。

在磁盘上，每个segment由一个log文件和2个index文件组成。

00000000000000849751.index
    00000000000000849751.log
    00000000000000849751.time index

这三个文件是成套出现的。

leader-epoch-checkpoint 文件中保存了每一任leader开始写入消息时的offset。

1.  .log日志文件(日志就是数据)  
    在一个segment文件里面，日志是追加写入的。如果满足一定条件，就会切分日志文件，产生一个新的segment。什么时候会触发segment的切分呢？
    
    第一种是根据日志文件大小。当一个segment写满以后，会创建一个新的segment，用最新的offset作为名称。这个例子可以通过往一个Topic发送大量消息产生。
    
    segment的默认大小是1073741824 bytes(1G) ，由这个参数控制：
    
        log.segment.bytes
    
    第二种是根据消息的最大时间戳，和当前系统时间戳的差值。
    
    有一个默认的参数，168个小时(一周)：
    
        log.roll.hours=168
    
    意味着：如果服务器上次写入消息是一周之前，旧的segment就不写了，现在要创建一个新的segment。
    
    还可以从更加精细的时间单位进行控制，如果配置了毫秒级别的日志切分间隔，会优先使用这个单位。否则就用小时的。
    
        log.roll.ms
    
    第三种情况，offset索引文件或者timestamp索引文件达到了一定的大小，默认是10485760字节(10M)。如果要减少日志文件的切分，可以把这个值调大一点。
    
        log.index.size.max.bytes
    
    亦即：索引文件写满了，数据文件也要跟着拆分，不然这一套东西对不上。
2.  .index 偏移量（offset）索引文件
3.  .timeindex时间戳（timestamp）索引文件

### 1.6 索引（index） ###

由于一个segment的文件里面可能存放很多消息，如果要根据offset获取消息，必须要有一种快速检索消息的机制。这个就是索引。在kafka中设计了两种索引。

偏移量索引文件记录的是offset和消息物理地址(在log文件中的位置) 的映射关系。时间戳索引文件记录的是时间戳和offset的关系。

当然，内容是二进制的文件，不能以纯文本形式查看。bin目录下有dump log工具。

查看最后10条offset索引：

sh kafka-dump-log.sh --files /tmp/kafka-logs/mytopic-0/00000000000000000000.index | head -n 10

![image-20210505191552712][]

注意kafka的索引并不是每一条消息都会建立索引，而是一种稀疏索引sparse index(DB 2和Mon gdb中都有稀疏索引) 。

![image-20210505190004548][]

所以问题就来了，这个稀疏索引到底有多稀疏?也就是说，隔几条消息才产生一个索引记录?或者隔多久?或者隔多少大小的消息?

实际上是用消息的大小来控制的，默认是4KB：

log.index.interval.bytes=4096

只要写入的消息超过了4KB，偏移量索引文件.index和时间戳索引文件.time index就会增加一条索引记录(索引项)。

这个值设置越小，索引越密集。值设置越大，索引越稀疏。

相对来说，越稠密的索引检索数据更快，但是会消耗更多的存储空间。

越的稀疏索引占用存储空间小，但是插入和删除时所需的维护开销也小。

Kafka索引的时间复杂度为O(log2n) +O(m) ，n是索引文件里索引的个数，m为稀疏程度。

第二种索引类型是时间戳索引。

为什么会有时间戳索引文件呢?光有offset索引还不够吗?会根据时间戳来查找消息吗?

首先消息是必须要记录时间戳的。客户端封装的Producer Record和ConsumerRecord都有一个long timestamp属性。

为什么要记录时间戳呢?

1.  如果要基于时间切分日志文件，必须要记录时间戳；
2.  如果要基于时间清理消息，必须要记录时间戳。

设计一个时间戳索引，可以根据时间戳查询。

注意时间戳有两种，一种是消息创建的时间戳，一种是消费在Broker追加写入的时间。到底用哪个时间呢?由一个参数来控制：

log.message.timestamp.type=CreateTime

默认是创建时间。如果要改成日志追加时间，则修改为LogAppendTime。

查看最早的10条时间戳索引：

sh kafka-dump-log.sh --files /tmp/kafka-logs/mytopic-0/00000000000000000000.timeindex | head -n 10

![image-20210505191430740][]

kafka如何基于索引快速检索消息？比如我要检索偏移量是10959的消息。

1.  消费的时候是能够确定分区的，所以第一步是找到在哪个segment中。Segment文件是用base offset命名的，所以可以用二分法很快确定(找到名字不小于10959的segment) 。
2.  这个segment有对应的索引文件，它们是成套出现的。所以现在要在索引文件中根据offset找position。
3.  得到position之后，到对应的log文件开始查找offset，和消息的offset进行比较，直到找到消息

思考一个面试问题：为什么kafka不用B+Tree?

Kafka是写多，查少。如果kafka用B+Tree，首先会出现大量的B+Tree，大量插入数据带来的B+Tree的调整会非常消耗性能。

## 2.消息清理机制 ##

# 消息清理开关
    log.cleaner.enable=true
    # 清理方式 1.直接删除 delete 2.对日志进行压缩 compact。默认是直接删除
    log.cleanup.policy=delete

### 2.1删除策略 ###

日志删除是通过定时任务实现的。默认5分钟执行一次，看看有没有需要删除的数据。

log.retention.check.interval.ms=300000

删除从哪里开始删呢?肯定是从最老的数据开始删。关键就是对于老数据的定义。

什么才是老数据的?

*  由时间控制

由一个参数来控制，默认：

log.retention.hours

默认值是168个小时(一周)，也就是时间戳超过一周的数据才会删除。

Kafka另外也提供了另外两个粒度更细的配置，分钟和毫秒。

log.retention.minutes

默认值是空。它的优先级比小时高，如果配置了则用这个。

log.retention.ms

默认值是空。它的优先级比分钟高，如果配置了则用这个。

这里还有一种情况，假设kafka产生消息的速度是不均匀的，有的时候一周几百万条，有的时候一周几千条，那这个时候按照时间来删除就不是那么合理了。

*  由文件大小控制

删除策略就是根据日志大小删除，先删旧的消息，删到不超过这个大小为止。

log.retention.bytes

默认值是-1，代表不限制大小，想写多少就写多少。log.retention.bytes指的是所有日志文件的总大小。也可以对单个segment文件大小进行限制。

log.segment.bytes

默认值1073741824字节(1G)。

### 2.2压缩策略 ###

问题：如果同一个key重复写入多次，会存储多次还是会更新?

比如用来存储位移的这个特殊的topic：\_\_consumer\_offsets，存储的是消费者id和partition的offset关系，消费者不断地消费消息commit的时候，是直接更新原来的offset，还是不断地写入新的offset呢?肯定是存储多次，不然怎么能实现顺序写。

当有了这些key相同的value不同的消息的时候，存储空间就被浪费了。压缩就是把相同的key合并为最后一个value。

这个压缩跟Compression的含义不一样。所以，这里称为压紧更加合适。

Log Compaction执行过后的偏移量不再是连续的，不过这并不影响日志的查询。

## 3.高可用架构 ##

### 3.1 Controller选举 ###

当创建添加一个的分区或者分区增加了副本的时候，都要从所有副本中选举一个新的Leader出来。

投票怎么玩?是不是所有的partition副本直接发起投票，开始竞选呢?比如用ZK实现。

利用ZK怎么实现选举? ZK的什么功能可以感知到节点的变化(增加或者减少)? 或者说，ZK为什么能实现加锁和释放锁?

3个特点：watch机制； 节点不允许重复写入； 临时节点。

这样实现是比较简单，但是也会存在一定的弊端。如果分区和副本数量过多，所有的副本都直接进行选举的话，一旦某个出现节点的增减，就会造成大量的watch事件被触发，ZK的负载就会过重。

Kafka早期的版本就是这样做的，后来换了一种实现方式。

不是所有的repalica都参与leader选举，而是由其中的一个Broker统一来指挥，这个Broker的角色就叫做Controller(控制器) 。

就像RedisSentinel的架构，执行故障转移的时候，必须要先从所有哨兵中选一个负责做故障转移的节点一样。Kafka也要先从所有Broker中选出唯一的一个Controller。

所有的Broker会尝试在zookeeper中创建临时节点/controller，只有一个能创建成功(先到先得)。

如果Controller挂掉了或者网络出现了问题，ZK上的临时节点会消失。其他的Broker通过watch监听到Controller下线的消息后，开始竞选新的Controller。方法跟之前还是一样的，谁先在ZK里面写入一个/controller节点，谁就成为新的Controller。

一个节点成为Controller之后，它肩上的责任也比别人重了几份，正所谓劳力越戴，责任越大：

*  监听Broker变化。
 *  监听Topic变化。
 *  监听Partition变化。
 *  获取和管理Broker、Topic、Partition的信息。
 *  管理Partion tion的主从信息。

### 3.2 分区副本leader选举 ###

https://kafka.apache.org/documentation/\#replication  
https://kafka.apache.org/documentation/\#design\_replicatedlog

Controller确定以后，就可以开始做分区选主的事情了(我叫它选举委员会主席) 。下面就是找候选人了。显然，每个replica都想推荐自己，但是所有的replica都有竞选资格吗?

并不是。这里要给大家说几个概念。

一个分区所有的副本，叫做Assigned-Replicas(AR) 。所有的皇太子。

这些所有的副本中，跟leader数据保持一定程度同步的，叫做In-Sync Replicas(ISR) 。

跟leader同步滞后过多的副本，叫做Out-Sync-Replicas(OSR) 。

AR=ISR+OSR。正常情况下OSR是空的，大家都正常同步，AR=ISR。

谁能够参加选举呢?肯定不是AR，也不是OSR，而是ISR。而且这个ISR不是固定不变的，还是一个动态的列表。

前面我们说过，如果同步延迟超过30秒，就踢出ISR，进入OSR； 如果赶上来了，就加入ISR。

默认情况下，当leader副本发生故障时，只有在IS R集合中的副本才有资格被选举为新的leader。

如果ISR为空呢? 在这种情况下，可以让ISR之外的副本参与选举。允许ISR之外的副本参与选举，叫做unclean leader election。

unclean.leader.election.enable=false

把这个参数改成true(一般情况不建议开启，会造成数据丢失) 。

好了，委员会主席有了，候选人也确定了，终于可以选举了吧?根据什么规则确定leader呢?

首先第一个问题：分布式系统中常见的选举协议有哪些(或者说共识算法)?

ZAB(ZK) 、Raft(Red is Sentinel) (他们都是Paxos算法的变种) ，它们的思  
想归纳起来都是：先到先得、少数服从多数。

但是kafka没有用这些方法，而是用了一种自己实现的算法。

为什么呢?比如ZAB这种协议，可能会出现脑裂(节点不能互通的时候，出现多个leader) 、惊群效应(大量watch事件被触发) 。

在这篇文章中：  
https://kafka.apache.org/documentation/\#design\_replicatedlog  
提到kafka的选举实现，最相近的是微软的PacificA算法。

There area rich variety of algorithms in this family including ZooKeeper’sZ ab，Raft，and  
View stamped Replication.The most similar academic publication we are aware of to Kafka’s  
actual implementation is PacificA from Microsoft.

在这种算法中，默认是让ISR中第一个replica变成leader。比如ISR是1、5、9，优先让1成为leader。这个跟中国古代皇帝传位是一样的，优先传给皇长子。

### 3.3 主从同步 ###

leader确定之后，客户端的读写只能操作leader节点。follower需要向leader同步数据。

不同的r aplica的offset是不一样的，同步到底怎么同步呢?

这里又要先讲解几个概念了。

![image-20210505194443428][]

LEO(Log End Offset) ：下一条等待写入的消息的offset(最新的offset+1)，图中分别是9，8，6。可以用命令看到：

sh kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group gp-test-group

PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG  
0 4 9 5

这个命令查看分区对应的offset：

sh kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 127.0.0.1:9092 --topic 'mytopic' --time -l

HW(Hign Watermark)：ISR中最小的LEO。Leader会管理所有ISR中最小的LEO作为HW，目前是6。

consumer最多只能消费到HW之前的位置(消费到offset 5的消息) 。也就是说：其他的副本没有同步过去的消息，是不能被消费的。

为什么要这样设计呢? 如果在同步成功之前就被消费了，consumer group的offset会偏大。如果leader崩溃，中间会缺失消息。

有了这两个offset之后，再来看看消息怎么同步。

Follower1同步了1条消息，follower2同步了2条消息。此时HW推进了2，变成8。

![image-20210505195558055][]

follower1同步了0条消息，follower2同步了1条消息。此时HW推进了1，变成9。LEO和HW重叠，所有的消息都可以消费了。

![image-20210505195932095][]

这里，我们关注一下，从节点怎么跟主节点保持同步?

1.  follower节点会向Leader发送一个fetch请求，leader向follower发送数据后，既需要更新follower的LEO。
2.  follower接收到数据响应后，依次写入消息并且更新LEO。
3.  leader更新HW(IS R最小的LEO) 。

kafka设计了独特的ISR复制，可以在保障数据一致性情况下又可提供高吞吐量。

### 3.4 replica 故障处理 ###

**follower故障**

首先follower发生故障，会被先踢出ISR。

follower恢复之后，从哪里开始同步数据呢?假设第1个replica宕机(中间这个) 。

![image-20210505194443428][]

恢复以后，首先根据之前记录的HW（6），把高于HW的消息截掉（6，7）。然后向leader同步消息。追上leader之后（30秒），重新加入ISR。

**leader故障**

假设图中leader发生故障。

首先选一个leader。因为replica 1(中间这个) 优先，它成为leader。

为了保证数据一致，其他的follower需要把高于HW的消息截取掉(这里没有消息需要截取)。

然后replica 2同步数据。

注意：这种机制只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。

[image-20210505180153257]: /images/20221020/5ee3474182ae451093e2723c5059dc76.png
[image-20210505181207727]: /images/20221020/3bce72bce1b74777b66b48df28086200.png
[image-20210505184006329]: /images/20221020/ccfa5768b2d24afd95a902ba17c679e0.png
[image-20210505191552712]: /images/20221020/000bf61396ad4c67b8cc5a685f1d6788.png
[image-20210505190004548]: /images/20221020/67a80d315c264af3b0a7efea586f6d2b.png
[image-20210505191430740]: /images/20221020/5ecb67f65d524907a2ebcd698fb647fa.png
[image-20210505194443428]: /images/20221020/8eb42ff4cff9477dbccb84cef6d8b076.png
[image-20210505195558055]: /images/20221020/fab27b3759b842d4bc318bfe88347449.png
[image-20210505195932095]: /images/20221020/afaa50a6bca248f39e1a5beba011c3d0.png