1. Kafka Broker 工作流程
1.1 Zookeeper 存储的 Kafka 信息
（1）启动 Zookeeper 客户端。
bin/zkCli.sh
（2）通过 ls 命令可以查看 kafka 相关信息。
ls /kafka
Zookeeper中存储的Kafka 信息
1.2 Kafka Broker 总体工作流程
1）模拟 Kafka上下线，Zookeeper中数据变化
（1）查看/kafka/brokers/ids 路径上的节点。
ls /kafka/brokers/ids [0, 1, 2]
（2）查看/kafka/controller 路径上的数据。
get /kafka/controller {"version":1,"brokerid":0,"timestamp":"1637292471777"}
（3）查看/kafka/brokers/topics/first/partitions/0/state 路径上的数据。
get /kafka/brokers/topics/first/partitions/0/state {"controller_epoch":24,"leader":0,"version":1,"leader_epoch":18,"isr":[0,1,2]}
（4）停止 hadoop104 上的 kafka。
bin/kafka-server-stop.sh
（5）再次查看/kafka/brokers/ids 路径上的节点
ls /kafka/brokers/ids [0, 1]
（6）再次查看/kafka/controller 路径上的数据。
get /kafka/controller {"version":1,"brokerid":0,"timestamp":"1637292471777"}
（7）再次查看/kafka/brokers/topics/first/partitions/0/state 路径上的数据。
get /kafka/brokers/topics/first/partitions/0/state {"controller_epoch":24,"leader":0,"version":1,"leader_epoch":18,"isr":[0,1]}
（8）启动 hadoop104 上的 kafka。
bin/kafka-server-start.sh -daemon ./config/server.properties
（9）再次观察（1）、（2）、（3）步骤中的内容。
1.3 Broker 重要参数
| -参数名称 | -描述 |
| replica.lag.time.max.ms | ISR 中，如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值，默认 30s。 |
| auto.leader.rebalance.enable | 默认是 true。 自动 Leader Partition 平衡。 |
| leader.imbalance.per.broker.percentage | 默认是 10%。每个 broker 允许的不平衡的 leader的比率。如果每个 broker 超过了这个值，控制器会触发 leader 的平衡。 |
| leader.imbalance.check.interval.seconds | 默认值 300 秒。检查 leader 负载是否平衡的间隔时间。 |
| log.segment.bytes | Kafka 中 log 日志是分成一块块存储的，此配置是指 log 日志划分 成块的大小，默认值 1G。 |
| log.index.interval.bytes | 默认 4kb，kafka 里面每当写入了 4kb 大小的日志（.log），然后就往 index 文件里面记录一个索引。 |
| log.retention.hours | Kafka 中数据保存的时间，默认 7 天。 |
| log.retention.minutes | Kafka 中数据保存的时间，分钟级别，默认关闭。 |
| log.retention.ms | Kafka 中数据保存的时间，毫秒级别，默认关闭。 |
| log.retention.check.interval.ms | 检查数据是否保存超时的间隔，默认是 5 分钟。 |
| log.retention.bytes | 默认等于-1，表示无穷大。超过设置的所有日志总大小，删除最早的 segment。 |
| log.cleanup.policy | 默认是 delete，表示所有数据启用删除策略；如果设置值为 compact，表示所有数据启用压缩策略。 |
| num.io.threads | 默认是 8。负责写磁盘的线程数。整个参数值要占总核数的 50%。 |
| num.replica.fetchers | 副本拉取线程数，这个参数占总核数的 50%的 1/3 |
| num.network.threads | 默认是 3。数据传输线程数，这个参数占总核数的50%的 2/3 。 |
| log.flush.interval.messages | 强制页缓存刷写到磁盘的条数，默认是 long 的最大值，9223372036854775807。一般不建议修改，交给系统自己管理。 |
| log.flush.interval.ms | 每隔多久，刷数据到磁盘，默认是 null。一般不建议修改，交给系统自己管理。 |
2. 节点服役和退役
2.1 服役新节点
1）新节点准备
(1)设置新增机器
vim /etc/hostname hadoop105
新增 haodoop105 中 kafka 的 broker.id 为 3。
（2）启动 hadoop102、hadoop103、hadoop104 上的 kafka 集群。
（3）单独启动 hadoop105 中的 kafka。
2）执行负载均衡操作
（1）创建一个要均衡的主题.
vim topics-to-move.json { "topics": [ {"topic": "first"} ], "version": 1 }
（2）生成一个负载均衡的计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate Current partition replica assignment {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[0,2,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[2,1,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,0,2],"log_dirs":["any","any","any"]}]} Proposed partition reassignment configuration {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,3,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"]}]}
（3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2、broker3 中）。
vim increase-replication-factor.json
输入如下内容：
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,3,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"]}]}
（4）执行副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute
（5）验证副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify Status of partition reassignment: Reassignment of partition first-0 is complete. Reassignment of partition first-1 is complete. Reassignment of partition first-2 is complete. Clearing broker-level throttles on brokers 0,1,2,3 Clearing topic-level throttles on topic first
2.2 退役旧节点
1）执行负载均衡操作
先按照退役一台节点，生成执行计划，然后按照服役时操作流程执行负载均衡。
（1）创建一个要均衡的主题。
vim topics-to-move.json { "topics": [ {"topic": "first"} ], "version": 1 }
（2）创建执行计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate Current partition replica assignment {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,2,3],"log_dirs":["any","any","any"]}]} Proposed partition reassignment configuration {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,2,0],"log_dirs":["any","any","any"]}]}
（3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2 中）。
vim increase-replication-factor.json {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,2,0],"log_dirs":["any","any","any"]}]}
（4）执行副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute
（5）验证副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify
2）执行停止命令
在 hadoop105 上执行停止命令即可。
bin/kafka-server-stop.sh
3. Kafka副本
3.1副本基本信息
（1）Kafka 副本作用：提高数据可靠性。
（2）Kafka 默认副本 1 个，生产环境一般配置为 2 个，保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率。
（3）Kafka 中副本分为：Leader 和 Follower。Kafka 生产者只会把数据发往 Leader，然后 Follower 找 Leader 进行同步数据。
（4）Kafka 分区中的所有副本统称为 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 副本同步时，延迟过多的副本。
3.2 Leader 选举流程
Kafka 集群中有一个 broker 的 Controller 会被选举为 Controller Leader，负责管理集群broker 的上下线，所有 topic 的分区副本分配和 Leader 选举等工作。
Controller 的信息同步工作是依赖于 Zookeeper 的。
（1）创建一个新的 topic，4 个分区，4 个副本
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic atguigu1 --partitions 4 --replication-factor 4 Created topic atguigu1
（2）查看 Leader 分布情况
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1 Topic: atguigu1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824 Topic: atguigu1 Partition: 0 Leader: 3 Replicas: 3,0,2,1 Isr: 3,0,2,1 Topic: atguigu1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,2,3,0 Topic: atguigu1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,3,1,2 Topic: atguigu1 Partition: 3 Leader: 2 Replicas: 2,1,0,3 Isr: 2,1,0,3
（3）停止掉 hadoop105 的 kafka 进程，并查看 Leader 分区情况
bin/kafka-server-stop.sh bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1 Topic: atguigu1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824 Topic: atguigu1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,2,1 Topic: atguigu1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,2,0 Topic: atguigu1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,1,2 Topic: atguigu1 Partition: 3 Leader: 2 Replicas: 2,1,0,3 Isr: 2,1,0
（4）停止掉 hadoop104 的 kafka 进程，并查看 Leader 分区情况
bin/kafka-server-stop.sh bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1 Topic: atguigu1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824 Topic: atguigu1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,1 Topic: atguigu1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,0 Topic: atguigu1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,1 Topic: atguigu1 Partition: 3 Leader: 1 Replicas: 2,1,0,3 Isr: 1,0
（5）启动 hadoop105 的 kafka 进程，并查看 Leader 分区情况
bin/kafka-server-start.sh -daemon config/server.properties bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1 Topic: atguigu1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824 Topic: atguigu1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,1,3 Topic: atguigu1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,0,3 Topic: atguigu1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,1,3 Topic: atguigu1 Partition: 3 Leader: 1 Replicas: 2,1,0,3 Isr: 1,0,3
（6）启动 hadoop104 的 kafka 进程，并查看 Leader 分区情况
bin/kafka-server-start.sh -daemon config/server.properties bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1 Topic: atguigu1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824 Topic: atguigu1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,1,3,2 Topic: atguigu1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,0,3,2 Topic: atguigu1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,1,3,2 Topic: atguigu1 Partition: 3 Leader: 1 Replicas: 2,1,0,3 Isr: 1,0,3,2
（7）停止掉 hadoop103 的 kafka 进程，并查看 Leader 分区情况
bin/kafka-server-stop.sh bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1 Topic: atguigu1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824 Topic: atguigu1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,3,2 Topic: atguigu1 Partition: 1 Leader: 2 Replicas: 1,2,3,0 Isr: 0,3,2 Topic: atguigu1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,3,2 Topic: atguigu1 Partition: 3 Leader: 2 Replicas: 2,1,0,3 Isr: 0,3,2
3.3Leader 和 Follower 故障处理细节
Follower故障处理细节
