Kafka之Broker原理

1. 日志数据的存储

1.1 Partition

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

2. 每个Partition中的消息是有序的，顺序写入，但是全局不一定有序

3. 在服务器上，每个Partition都有一个物理目录（TopicN）后面的数字代表分区

1.2 Replica副本

1. 为了提高分区的可靠性，Kafka设计了副本机制

2. 副本数必须小于等于节点数，而不能大于Broker的数量

3. Leader对外提供读写服务，Follower唯一的任务就是从Leader异步拉取数据

1.3 Segment段

1. 为了防止Log不断追加导致文件过大，导致检索消息效率变低，一个Partition又 被划分成多个Segment来组织数据.

在这里会有3个配置，也就是log的阈值配置。什么时候下进行分段

• log.segment.bytes：根据日志文件大小
• log.roll.hours、log.roll.ms：根据时间戳差值log.index.size.max.bytes：根据索引文件大小

每一个segment都是由一个log文件和2个index文件组成的,其中时间戳索引的创建方式可以自定义的执行createTime或LogAppendTime.默认是creareTime

1.4 Sparse Index(稀疏索引)

索引文件的查看可以通过以下命令进行查看

kfaka索引文件中记录的Offset不是连续的，而是采用了稀疏索引。根据配置的大小，稀疏索引记录的是从Log中的哪个位置开始检索，比如配置的是4kb，则当log文件中向下存储的数据达到4kb的话，就会记录一个索引值

1.5 分区副本在Broker上的分布

创建一个topic

./kafka-topics.sh--bootstrap-server192.168.61.101:9092--create--topic3p3r--partitions3--replication-factor3

假设配置的是3p3r，则我们看下服务器上的存储

查看Topic信息

./kafka-topics.sh--bootstrap-server192.168.61.101:9092--describe--topic3p3r

其中 Partition是分区，Leader后面代表的是在哪台服务器上，Replicas就是副本信息，ISR是个副本队列

假设配置的是4p2r，则物品们查看topic信息如图所示

创建、查看topic

./kafka-topics.sh --bootstrap-server 192.168.61.101:9092 --create --topic 4p2r --partitions 4 --replication-factor 2

./kafka-topics.sh --bootstrap-server 192.168.61.101:9092 --describe --topic 4p2r

假设我们配置的是6p2r

由以上我们可以看出，副本分配的两个基本原则和规律

1、副本会被平均分布在所有的Broker之上

2、partition的多个副本应该分配在不同的Broker上

基于上面的规则，分区副本最终落入哪个节点，还会收到两个随机数的影响

1、第一个随机数：startIndex，决定了第一个分区的第一个副本的放置位置

2、第二个随机数：nextReplicaShift，决定了分区中，副本跟副本的间距nextReplicaShift%(BrokerSize-1)

这样设计的目的在于提高Broker服务器的容灾能力

2. 消息保留与清理机制

对于一些太久的日志，我们需要一定的清理策略。

当开启清理策略后，有两种方式提供开发者选择

log.cleanup.policy=delete（默认项） // 删除策略

log.cleanup.policy=compact // 压缩策略

2.1 删除策略(delete)

kafka可以通过定时任务实现日志数据的删除，默认5分钟执行一次

log.retention.check.interval.ms=300000

那么要删除什么样的数据呢?kafka提供了两个纬度以及对应不同的配置

时间纬度

log.retention.hours（默认值是168个小时，时间戳超过的数据会被删除）

log.retention.minutes（默认值是空，优先级比小时高）

log.retention.ms（默认值是空，优先级比分钟高）

若产生消息的速度不均匀，有时多、有时少，就可以根据日志大小删除

log.retention.bytes（表示所有日志文件的总大小，默认值是-1，代表不限制大小）

log.segment.bytes（对单个Segment文件大小进行限制，默认值1G）

2.2 压缩策略(compact)

若设置为压缩策略compact,则表示不清楚日志，只对日志数据进行压缩处理

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

kafka中是存储多次的，如: _ _consumer_offsets

那么压缩策略是怎么做的呢?（将相同的key进行去重压缩）

3. Broker高可用架构

高可用，无非就是选举机制、数据的一致性也就是主从同步，以及对于故障的处理，由于kafka是直接数据存储在磁盘中的，因此无需考虑持久化，Broker的高可用 涉及到一系列的动作

• 选举出一个Controller
• 从分区中选举出Leader角色
• 主从同步
• Replica故障处理

3.1 选举机制

3.1.1 Controller选举

Controller其实就是一个Broker,由它来负责选举出新的Leader,那么Controller是怎么选举出来的呢

3.1.2 分区副本Leader的选举

在讲解Leader选举前，我们先复习以下博客Kafka之Producer原理-CSDN博客中提到的ISR机制的几个概念

AR（Assigned-Replicas），一个分区所有的副本

ISR（In-Sync Replicas），在AR中，跟Leader保持积极同步数据的副本

OSR（Out-Sync-Replicas），在AR中，跟Leader同步滞后的副本

AR = ISR + OSR

• 当Leader副本发生故障时，只有在ISR中的副本才能参与新Leader的选举 - 问题:如果ISR为空呢? unclean.leader.election.enable配置为false OSR也可以进行选举
• Kafka采用了类似于继位传嫡的选举协议，选择ISR中位置靠前的节点成为新的Leader.

3.2 主从同步

从节点和主节点的同步过程如下:

1、首先，Follower节点向Leader发送一个fetch请求

2、然后，Leader向Follower发送数据

3、接着，Follower接收到数据响应后，依次写入消息、并更新LEO值

4、最后，Leader更新HW（ISR最小的LEO）

5、循环上述过程，直至所有Follower完成数据同步

整体流程图如下所示:

Kafka设计的ISR复制，既可以在保障数据一致性，又可以提供高吞吐量(ISR队列中清除响应不积极的Follower节点)

3.3 Replica故障处理

• Follower发生故障，会被先提出ISR，Follower恢复之后，从HW开始同步数据
• Leader发生故障，会先选举出一个新的Leader,其它的Follower将高于HW的消息截取掉，然后从新的Leader同步数据

4. 总结

    本文介绍Broker服务器，主要讲了Broker中日志的存储，从大到小依次为Partition、Segment，副本机制的具体存储形式，是怎么进行负载均衡和容灾保障的，在Segment中我们直到了Segment是由一个Log文件和两个索引文件组成的，索引文件主要起的是一个提升查询效率的作用。随后当kafka中log文件过大的时候，kagka中提供了两种维度上的删除策略以及相同key去重压缩的compact策略。最后，kafka高可用中的选举机制是先到先得选举Controller,再根据ISR副本队列嫡长子继位的算法进行Leader的选举；以及Kafka中的主从同步是以高水位HW为界限，不断的同步数据，直到LEO值相等完成数据的同步。最后讲到了副本故障的处理，针对follwe节点故障，则直接踢出ISR队列，Leader故障，就会触发选举机制，选举出一个新的Leader，最后数据从LEO处以上的开始同步，高于HW的消息全部截断。