1. zk中存储的kafka信息
/kafka/brokers/ids存储了在线的broker id。
/kafka/brokers/topics/xxx/partitions/n/state存储了Leader是谁以及isr队列
/kafka/controller辅助Leader选举，每个broker都有一个controller，谁先在zk中注册上，谁就辅助Leader选举。
2. broker总体工作流程
1）每台broker启动后在zk中注册，即/kafka/borkers/ids
2）每台broker去抢占式注册controller，用于后面Leader选举
3）由注册的controller监听/kafka/borkers/ids节点变化
4）开始Leader选举，选举标准是以isr中存活为前提，以AR中排在前面的优先（AR是所有副本的集合，启动时会有一个固定的AR顺序，比如ar[1, 0, 2]）
5）controller将选举出来的信息（Leader和isr信息）传到zk中，即/kafka/brokers/topics/xxx/partitions/n/state
6）其他broker的controller会从zk中同步相关信息
Kafka生产者发送数据到broker，数据在底层以Log方式（逻辑概念）存储，实际上是Segment（物理概念），一般1个Segment是1G，包含.log文件和.index文件，.index文件是索引，用于快速查询数据
7）如果Leader挂了，controller监听到节点变化，选举新的Leader，选举标准依然是以isr中存活为前提，以AR中排在前面的优先，最后更新Leader和isr队列信息
3. 新节点服役
新节点服役后，以前的topic所在的分区不会出现在新节点，即新节点不会分摊旧节点的存储压力。如果需要新节点参与进来，就需要进行一种类似于负载均衡的配置。先创建一个topic-to-move.json配置文件：
{"topics": [{"topic": "first"}],"version": 1
}生成一个负载均衡的计划：
bin/kafka-reassign-repartitions.sh --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate
上面一行是当前的分区分配，下面一行是建议的分区分配计划，创建副本存储计划increase-replication-factor.json，里面内容是上面得分建议计划。最后执行存储计划：
bin/kafka-reassign-repartitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute
还可以验证计划：
bin/kafka-reassign-repartitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify
查询这个topic的分区详情
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --topic first --describe
4. 退役旧节点
退役旧节点与服役新节点有一些类似，先创建一个topic-to-move.json配置文件，与服役新节点时一样，然后生成一个计划，只不过--broker-list 改为"0,1,2"，接着执行计划，验证计划，都与服役新节点一样。
最后在退役节点关闭kafka服务
bin/kafka-server-stop.sh5. Leader选举验证
创建四个分区四个副本的topic并查看：
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic atguigu2 --partitions 4 --replications-factor 4bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu2
把3号broker停掉，那么isr队列中没有3，并且4号分区的Leader变为2
再把2号干掉
再恢复3号，发现Leader未变，仅isr队列信息中新增了3号
再恢复2号
再干掉1号
这样就验证了第二节讲的选举标准： 以isr中存活为前提，以AR中排在前面的优先
6. Leader和Follower故障处理细节
LEO：Log End Offset，每个副本的最后一个offset+1
HW：high watermark，高水位线，所有副本中最小的LEO，消费者能够看到的最大的offset就是HW - 1
1）如果Follower挂了，该Follower会立即被踢出isr，isr中其他Leader和Follower正常接受/同步数据，待该Follower恢复后，会读取上次的HW，将自己高于HW的数据丢弃，从HW开始与Leader同步，等到该Follower的LEO大于等于该Partition的HW，则重新加入isr队列。
2）如果Leader挂了， Leader会立即被踢出isr，并且会选出一个新的Leader，其余的Follower会将高于HW的数据丢弃，然后与新的Leader进行同步。此时只能保证数据的一致性，不能保证数据不丢失。
7. 手动调整分区副本
如果服务器的存储能力不同，希望将数据更多的存储在空间大的服务器上，那么就不应该按照Kafka分区副本的默认均匀分配，而是需要手动调整。创建4个分区，两个副本，都存在0号和1号broker上面。
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic three --partitions 4 --replications-factor 2bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic three
创建increase-replication-factor.json：
{"partitions": [{"topic": "three", "partitions": 0, "replicas": [0, 1]},{"topic": "three", "partitions": 1, "replicas": [0, 1]},{"topic": "three", "partitions": 2, "replicas": [1, 0]},{"topic": "three", "partitions": 3, "replicas": [1, 0]}],"version": 1
}执行存储计划：
bin/kafka-reassign-repartitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute
最后查看
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic three
以上是减少副本，增加副本也是类似，先创建一个3个分区，1个副本的topic：
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic four --partitions 3 --replications-factor 1创建increase-replication-factor.json：
{"partitions": [{"topic": "four", "partitions": 0, "replicas": [0, 1, 2]},{"topic": "four", "partitions": 1, "replicas": [0, 1, 2]},{"topic": "four", "partitions": 2, "replicas": [0, 1, 2]}],"version": 1
}执行计划：
bin/kafka-reassign-repartitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute8. Leader Partition自动平衡
在Leader选举验证小节中，如果2号和3号节点都挂了，然后又恢复，则Leader过于集中在0号和1号节点，而Kafka生产者和消费者都是只对Leader操作，所以0号和1号的压力会很大，造成负载不均衡。 未解决该问题，Kafka会自动再平衡，auto.leader.rebalance.enable默认设为true。
什么时机会触发再平衡呢？一个参考指标是broker的不平衡率，leader.imbalance.per.broker.percentage，默认是10%，另一个指标是负载检查的间隔时间，leader.imbalance.check.interval.seconds，默认是300秒。
不平衡率的计算：
实际生产环境中，不一定需要开启再平衡，因为上述例子中其实已经相对平衡了，但是根据规则，需要触发再平衡，因此会需要消耗大量资源。
9. 文件存储机制
Topic是逻辑上的概念，而partition是物理上的概念，每个partition对应一个log文件，该log文件存储的就是Kafka生产者的数据。生产的数据不断地追加到log文件中，为防止log文件过大导致检索数据慢，Kafka采取了分片和索引的机制：每个partition分为多个segment，每个segment包括.index文件（偏移量索引文件）、.log文件（日志文件）、.timeindex文件（时间戳索引文件）。这些文件位于一个文件夹中，文件夹命名规则：topic名称+分区号。index和log文件的命名是以当前segment的第一条数据的offset来命名。
log文件和index文件详解：
10. 文件清除策略
Kafka数据默认保存7天，7天后数据自动删除或者压缩。可通过如下参数修改保存时间（从上到下优先级依次增高）：
log.retention.hours
log.retention.minutes
log.retention.ms
默认检查数据是否超期的间隔时间是5分钟，可通过参数log.retention.check.interval.ms进行修改。
如果是删除数据，log.cleanup.policy=delete，基于时间删除是默认打开的，以segment中最大的时间戳作为该文件的时间戳。而基于空间大小进行删除是默认关闭的（log.retention.bytes=-1），即数据超过阈值，删除最早的数据。
如果是压缩数据，log.cleanup.policy=compact，此时对于相同key的不同value值，只保留最新的。（与之前的snappy压缩概念不同）
注意，压缩后的offset可能不是连续的，比如上图没有 offset 6，如果从offset 6开始消费，则会从7开始消费。
11. 高效读写
1）Kafka本身是分布式集群，采用分区，并行度高
2）读数据采用稀疏索引，可以快读定位数据
3）顺序写磁盘，数据以追加的方式写到log文件，这比随机写的速度要快很多，因为省去了大量的磁头寻址时间
4）采用页缓存和零拷贝技术
零拷贝：Kafka的数据加工处理操作交由Kafka生产者和消费者处理。Broker应用层不关心存储的数据，因此就不用走应用层，传输效率高。（传统数据复制方式：从磁盘中读取文件到内核缓冲区，内核读取缓冲区数据复制到用户缓冲区，用户缓冲区的数据复制到socket缓冲区，socket缓冲区数据发送到网卡，再到消费者）
页缓存：Kafka重度依赖Linux提供的页缓存功能。当上层有写操作时，操作系统只是将数据写入页缓存。当读操作发生时，从页缓存中读，如果找不到，再从磁盘中读。页缓存是把尽可能多的空闲内存当做磁盘内存来用。
