第 4 章 Kafka Broker
4.1 Kafka Broker 工作流程
4.1.1 Zookeeper 存储的 Kafka 信息
（1）启动 Zookeeper 客户端。
bin/zkCli.sh
因为你在配置kafka的时候指定了它的名字。
（2）通过 ls 命令可以查看 kafka 相关信息。
[zk: localhost:2181(CONNECTED) 2] ls /kafka
zk中有一个节点 consumers 这个里面，老版本0.9版本之前，存放的是消费者的偏移量（offset，这次消费者消费到哪个地方了，下次从这个地方继续消费），新版本的根本没放在zk中，直接放在集群中了。
可以借助一个工具：漂亮的zoo，通过图形化界面查看zk中的消息。
4.1.2 Kafka Broker 总体工作流程
1）每一个broker上线时，会在zk中进行注册
2）每个broker中都有一个controller，controller会争先抢占zk中 controller节点的注册权，谁先抢到，谁选举时说了算。假如broker0中的controller中抢到了，那它就是说了算的人。该controller一直监听ids节点是否有挂掉的节点。
3）选举规则是：在ISR中存活为前提，按照AR中排在前面的优先，例如 ar[1,0,2] ,isr[1,0,2],那么Leader会按照1，0，2 进行顺序的轮询。
4）选举出来的新节点，注册到zk中，将信息记录在zk中。
5）其他contorller将zk中的信息同步下来。
6）假定broker中的leader挂掉了，会进行重新的选举。
7）客户端发送消息给Leader，Leader记录数据，落盘，形成Log，Log底层使用的是Segment，Segment底层每一个G，是一个单独的文件，1G内的数据要想查找迅速又分成了两个文件 log和index.
模拟 Kafka 上下线，Zookeeper 中数据变化
（1）查看/kafka/brokers/ids 路径上的节点。
[zk: localhost:2181(CONNECTED) 2] ls /kafka/brokers/ids
[0, 1, 2]
（2）停止 hadoop104 上的 kafka。
bin/kafka-server-stop.sh
（3）再次查看/kafka/brokers/ids 路径上的节点。
[zk: localhost:2181(CONNECTED) 3] ls /kafka/brokers/ids
[0, 1]
实操：将hadoop13停止掉，会重新选举，两次查看的对比图如下：
4.2 生产经验——节点服役和退役
4.2.1 服役新节点
1）新节点准备
（1）关闭 bigdata03，进行一个快照，并右键执行克隆操作。
（2）开启 bigdata04，并修改 IP 地址。
vi /etc/sysconfig/network-scripts/ifcfg-ens33修改完记得重启网卡：
systemctl restart network（3）在 bigdata04 上，修改主机名称为 bigdata04。
hostname bigdata04 # 临时修改[root@bigdata04 ~]# vim /etc/hostname
bigdata04
还要记得修改 /etc/hosts文件，并进行同步
修改bigdata01的hosts 文件，修改完之后，记得同步一下192.168.52.11 bigdata01
192.168.52.12 bigdata03
192.168.52.13 bigdata02
192.168.52.14 bigdata04xsync.sh /etc/hosts
scp -r /etc/hosts root@bigdata04:/etc/（4）重新启动 bigdata03、bigdata04。
（5）修改 bigdata04 中 kafka 的 broker.id 为 3。
进入bigdata04的kafka中，修改里面的配置文件 config/server.properties（6）删除 bigdata04 中 kafka 下的 datas 和 logs。
rm -rf datas/* logs/*
（7）启动 bigdata01、bigdata02、bigdata03 上的 kafka 集群。
先启动zk集群
xcall.sh zkServer.sh stop
xcall.sh zkServer.sh start启动kafka集群（只能启动三台）
kf.sh start （8）单独启动 bigdata04 中的 kafka。
bin/kafka-server-start.sh -daemon ./config/server.properties
查看kafka集群first主题的详情：
bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --topic first --describe
发现副本数并没有增加。由于我之前创建first这个主题的时候只有一个副本，不是三个副本，所以呢，演示效果不佳。
kafka-topics.sh --bootstrap-server bigdata01:9092 --topic third --create --partitions 3 --replication-factor 3
2）执行负载均衡操作
（1）创建一个要均衡的主题
创建一个文件：vi topics-to-move.json
写上如下代码，如果多个topic 可以使用,分隔
{"topics": [{"topic": "third"}],"version": 1
}（2）生成一个负载均衡的计划
在创建的时候，记得启动bigdata04节点，否则计划中还是没有bigdata04
bin/kafka-reassign-partitions.sh --bootstrap-server bigdata01:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate
未来的分区策略拷贝一份：
{"version":1,"partitions":[{"topic":"third","partition":0,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"topic":"third","partition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"third","partition":2,"replicas":[1,2,3],"log_dirs":["any","any","any"]}]}（3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2、broker3 中）。
vi increase-replication-factor.json{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[3,2,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[0,3,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,0,2],"log_dirs":["any","any","any"]}]}
以上这个内容来自于第二步的执行计划。（4）执行副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server bigdata01:9092 --reassignment-json-file increase-replication-factor.json --execute
（5）验证副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server bigdata01:9092 --reassignment-json-file increase-replication-factor.json --verify
如果不相信添加成功，可以查看first节点的详情：
4.2.2 退役旧节点
1）执行负载均衡操作
先按照退役一台节点，生成执行计划，然后按照服役时操作流程执行负载均衡。
（1）创建一个要均衡的主题
kafka下添加文件：vim topics-to-move.json
添加如下内容：
{"topics": [{"topic": "third"}],"version": 1
}（2）创建执行计划。
bin/kafka-reassign-partitions.sh --bootstrap-server bigdata01:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate
（3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2 中）。
添加文件： vi increase-replication-factor.json
添加如下代码：
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[0,2,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[1,0,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[2,1,0],"log_dirs":["any","any","any"]}]}（4）执行副本存储计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop11:9092 --reassignment-json-file increase-replication-factor.json --execute（5）验证副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop11:9092 --reassignment-json-file increase-replication-factor.json --verify
2）执行停止命令
在 bigdata04上执行停止命令即可。
bin/kafka-server-stop.sh
4.3 Kafka 副本
4.3.1 副本基本信息 --副本又叫副本因子
（1）Kafka 副本作用：提高数据可靠性。
（2）Kafka 默认副本 1 个，生产环境一般配置为 2 个，保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率。
（3）Kafka 中副本分为：Leader 和 Follower。Kafka 生产者只会把数据发往 Leader， 然后 Follower 找 Leader 进行同步数据。
（4）Kafka 分区中的所有副本（包含Leader）统称为 AR（Assigned Repllicas）。
AR = ISR + OSR
ISR，表示和 Leader 保持同步的 Follower 集合。如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms 参数设定，默认 30s。Leader 发生故障之后，就会从 ISR 中选举新的 Leader。
OSR，表示 Follower 与 Leader 副本同步时，延迟过多的副本.
4.3.2 Leader 选举流程
Kafka 集群中有一个 broker 的 Controller 会被选举为 Controller Leader，负责管理集群broker 的上下线，所有 topic 的分区副本分配和 Leader 选举等工作。
Controller 的信息同步工作是依赖于 Zookeeper 的。
（1）创建一个新的 topic，4 个分区，4 个副本
bin/kafka-topics.sh --bootstrap-server hadoop11:9092 --create --topic bigdata2308 --partitions 4 --replication-factor 4
（2）查看 Leader 分布情况
bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --describe --topic bigdata2305
Topic: bigdata2301 Topic: bigdata2301 Partition: 0 Leader: 0 Replicas: 0,2,3,1 Isr: 0,2,3,1Topic: bigdata2301 Partition: 1 Leader: 2 Replicas: 2,3,1,0 Isr: 2,3,1,0Topic: bigdata2301 Partition: 2 Leader: 3 Replicas: 3,1,0,2 Isr: 3,1,0,2Topic: bigdata2301 Partition: 3 Leader: 1 Replicas: 1,0,2,3 Isr: 1,0,2,3（3）停止掉 hadoop13 的 kafka 进程，并查看 Leader 分区情况
bin/kafka-server-stop.shbin/kafka-topics.sh --bootstrap-server bigdata01:9092 --describe
--topic bigdata2305
Topic: bigdata2301 Partition: 0 Leader: 0 Replicas: 0,2,3,1 Isr: 0,2,1
Topic: bigdata2301 Partition: 1 Leader: 2 Replicas: 2,3,1,0 Isr: 2,1,0
Topic: bigdata2301 Partition: 2 Leader: 1 Replicas: 3,1,0,2 Isr: 1,0,2
Topic: bigdata2301 Partition: 3 Leader: 1 Replicas: 1,0,2,3 Isr: 1,0,2（4）停止掉 hadoop14 的 kafka 进程，并查看 Leader 分区情况
通过以上演示，大家可以发现，选举是按照AR(跟Replicas一样)进行的，而不是ISR
4.3.3 Leader 和 Follower 故障处理细节
LEO演示-- 每一个副本最后的偏移量offset + 1
HW(高水位线 High Water) 演示：所有副本中，最小的LEO
由于数据同步的时候先进入Leader,随后同步给Follower，假如Follower挂掉了，Leader和其他的Follower 继续往前存储数据，挂掉的节点从ISR集合中剔除，此时挂掉的Follower又重启了，它会先从上一次挂掉的节点的HW开始同步数据，直到追上最后一个Follower为止,此时会重新回归ISR。
4.3.4 分区副本分配
如果 kafka 服务器只有 4 个节点，那么设置 kafka 的分区数大于服务器台数，在 kafka底层如何分配存储副本呢？ 答案是否定的。
1）创建 16 分区，3 个副本
（1）创建一个新的 topic，名称为 second。
bin/kafka-topics.sh --bootstrap-server hadoop11:9092 --create --partitions 16 --replication-factor 3 --topic second
bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --create --topic bigdata230801 --partitions 3 --replication-factor 4
假如你有3个broker ,却创建4个副本，报错！！Error while executing topic command : Replication factor: 4 larger than available brokers: 3.
[2023-09-13 18:43:47,458] ERROR org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 4 larger than available brokers: 3bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --create --topic bigdata23 --partitions 4 --replication-factor 2
假如你有3个broker ,却创建4个分区，是可以的。以上错误的意思是，目前只有2台服务器，却要创建3个副本，创建不了。（2）查看分区和副本情况
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic second
Topic: second4 Partition: 0 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2
Topic: second4 Partition: 1 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic: second4 Partition: 2 Leader: 2 Replicas: 2,3,0 Isr: 2,3,0
Topic: second4 Partition: 3 Leader: 3 Replicas: 3,0,1 Isr: 3,0,1
Topic: second4 Partition: 4 Leader: 0 Replicas: 0,2,3 Isr: 0,2,3
Topic: second4 Partition: 5 Leader: 1 Replicas: 1,3,0 Isr: 1,3,0
Topic: second4 Partition: 6 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
Topic: second4 Partition: 7 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2
Topic: second4 Partition: 8 Leader: 0 Replicas: 0,3,1 Isr: 0,3,1
Topic: second4 Partition: 9 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2
Topic: second4 Partition: 10 Leader: 2 Replicas: 2,1,3 Isr: 2,1,3
Topic: second4 Partition: 11 Leader: 3 Replicas: 3,2,0 Isr: 3,2,0
Topic: second4 Partition: 12 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2
Topic: second4 Partition: 13 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic: second4 Partition: 14 Leader: 2 Replicas: 2,3,0 Isr: 2,3,0
Topic: second4 Partition: 15 Leader: 3 Replicas: 3,0,1 Isr: 3,0,1
kafka在进行初始化的时候，选举谁当第一Leader，是有一定的算法的。算法保障了Leader不在一个broker里面。4.3.5 生产经验——手动调整分区副本的存储
在生产环境中，每台服务器的配置和性能不一致，但是Kafka只会根据自己的代码规则创建对应的分区副本，就会导致个别服务器存储压力较大。所有需要手动调整分区副本的存储。
需求：创建一个新的topic，4个分区，两个副本，名称为three。将 该topic的所有副本都存储到broker0和broker1两台服务器上。
手动调整分区副本存储的步骤如下：
（1）创建一个新的 topic，名称为 three。
bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --create --partitions 4 --replication-factor 2 --topic three
（2）查看分区副本存储情况。
bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --describe --topic three
（3）创建副本存储计划（所有副本都指定存储在 broker0、broker1 中）。
vi increase-replication-factor.json
输入如下内容：
{ "version":1, "partitions":[{"topic":"three","partition":0,"replicas":[0,1]}, {"topic":"three","partition":1,"replicas":[0,1]}, {"topic":"three","partition":2,"replicas":[1,0]}, {"topic":"three","partition":3,"replicas":[1,0]}]
}4）执行副本存储计划。
屁股决定脑袋
bin/kafka-reassign-partitions.sh --bootstrap-server bigdata01:9092 --reassignment-json-file increase-replication-factor.json --execute
（5）验证副本存储计划。
bin/kafka-reassign-partitions.sh -- bootstrap-server bigdata01:9092 --reassignment-json-file increase-replication-factor.json --verify
（6）查看分区副本存储情况
bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --describe --topic three
4.3.6 生产经验——Leader Partition 负载平衡
正常情况下，Kafka本身会自动把Leader Partition均匀分散在各个机器上，来保证每台机器的读写吞吐量都是均匀的。但是如果某些broker宕机，会导致Leader Partition过于集中在其他少部分几台broker上，这会导致少数几台broker的读写请求压力过高，其他宕机的broker重启之后都是follower partition，读写请求很低，造成集群负载不均衡。
auto.leader.rebalance.enable，默认是true。 自动Leader Partition 平衡
• leader.imbalance.per.broker.percentage， 默认是10%。每个broker允许的不平衡 的leader的比率。如果每个broker超过 了这个值，控制器会触发leader的平衡。
• leader.imbalance.check.interval.seconds， 默认值300秒。检查leader负载是否平衡的间隔时间。
从以上可以看出，Leader 0，明明是3需要变为Leader,就说明这个中有Leader挂了再重启的情况，4个节点，一个节点不平衡，1/4 >10%,所以会触发再平衡，其他节点也是一样的。
生产环境下：该值默认为true，一般修改为false，因为不影响正常的使用，再平衡会造成资源的浪费。
4.3.7 生产经验——增加副本因子
在生产环境当中，由于某个主题的重要等级需要提升，我们考虑增加副本。副本数的
增加需要先制定计划，然后根据计划执行。
1）创建 topic
bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --create --partitions 3 --replication-factor 1 --topic four
通过命令行修改副本是否成功？
分区是可以通过语句修改的，只能改多，不能改少，副本创建以后就不能直接修改了。bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --alter --partitions 3 --replication-factor 3 --topic four
没办法使用命令修改的。2）手动增加副本存储
通过命令查看副本情况
（1）创建副本存储计划（所有副本都指定存储在 broker0、broker1、broker2 中）。
vi increase-replication-factor.json
添加如下内容：
{"version":1,"partitions":[{"topic":"four","partition":0,"replicas":[0,1,2]},{"topic":"four","partition":1,"replicas":[0,1,2]},{"topic":"four","partition":2,"replicas":[0,1,2]}]}（2）执行副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop11:9092 --reassignment-json-file increase-replication-factor.json --execute
查看副本情况：
4.4 文件存储
4.4.1 文件存储机制 （重要）
1）Topic 数据的存储机制
Topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是Producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制，将每个partition分为多个segment。每个segment包括：“.index”文件、“.log”文件和.timeindex等文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号，例如：first-0。
2）思考：Topic 数据到底存储在什么位置？
（1）启动生产者，并发送消息。
bin/kafka-console-producer.sh -- bootstrap-server bigdata01:9092 --topic first
>hello world
（2）查看 hadoop11（或者 hadoop12、hadoop13）的/opt/installs/kafka3/datas/first-1 （first-0、first-2）路径上的文件。
进入查看：ls
00000000000000000092.index
00000000000000000092.log
00000000000000000092.snapshot
00000000000000000092.timeindex
leader-epoch-checkpoint
partition.metadata（3）直接查看 log 日志，发现是乱码
（4）通过工具查看 index 和 log 信息。
kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.indexkafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.log
kafka存储数据的时候使用的是稀疏索引，所以运行速度快。
4.4.2 文件清理策略
Kafka 中默认的日志（这个地方是数据的意思，就是Segment）保存时间为 7 天，可以通过调整如下参数修改保存时间。
log.retention.hours，最低优先级小时，默认 7 天。
log.retention.minutes，分钟。 --如果设置了该值，小时的设置不起作用。
log.retention.ms，最高优先级毫秒。 --如果设置了该值，分钟的设置不起作用。
log.retention.check.interval.ms，负责设置检查周期，默认 5 分钟。
那么日志一旦超过了设置的时间，怎么处理呢？
Kafka 中提供的日志清理策略有 delete 和 compact 两种。
1）delete 日志删除：将过期数据删除
log.cleanup.policy = delete 所有数据启用删除策略
（1）基于时间：默认打开。以 segment 中所有记录中的最大时间戳作为该文件时间戳。
（2）基于大小：默认关闭。超过设置的所有日志总大小，删除最早的 segment。
log.retention.bytes，默认等于-1，表示无穷大。
思考：如果一个 segment 中有一部分数据过期，一部分没有过期，怎么处理？
2）compact 日志压缩(合并的意思，不是真的压缩)
compact日志压缩：对于相同key的不同value值，只保留最后一个版本。 log.cleanup.policy = compact 所有数据启用压缩策略
压缩后的offset可能是不连续的，比如上图中没有6，当从这些offset消费消息时，将会拿到比这个offset大的offset对应的消息，实际上会拿到offset为7的消息，并从这个位置开始消费。
这种策略只适合特殊场景，比如消息的key是用户ID，value是用户的资料，通过这种压缩策略，整个消息集里就保存了所有用户最新的资料。
比如：张三 去年18岁，今年19岁，这种场景下可以进行压缩。
4.5 高效读写数据 （面试题）
1）Kafka 本身是分布式集群，可以采用分区技术，并行度高
2）读数据采用稀疏索引，可以快速定位要消费的数据。（mysql中索引多了之后，写入速度就慢了）
3）顺序写磁盘
Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程是一直追加到文件末端， 为顺序写。官网有数据表明，同样的磁盘，顺序写能到 600M/s，而随机写只有 100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。
4）页缓存 + 零拷贝技术
零拷贝：Kafka的数据加工处理操作交由Kafka生产者和Kafka消费者处理。Kafka Broker应用层不关心存储的数据，所以就不用 走应用层，传输效率高PageCache页缓存：Kafka重度依赖底层操作系统提供的PageCache功 能。当上层有写操作时，操作系统只是将数据写入 PageCache。当读操作发生时，先从PageCache中查找，如果找不到，再去磁盘中读取。实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用
生产者将数据发送给kafka,kafka将数据交给Linux内核，Linux内核将数据放入自身操作系统的页缓存中，然后到一定值写入磁盘，假如消费者过来消费，直接从页缓存中，通过网卡发送给消费者，根本就没有去kafka的业务系统中获取数据，所以速度比较快。
跟这个问题非常像：mysql读取数据的速度为什么这么快！--buffer pool
