14-Kafka-Day02

第 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 bigdata04

xsync.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,1
        Topic: bigdata2301      Partition: 1    Leader: 2       Replicas: 2,3,1,0   Isr: 2,3,1,0
        Topic: bigdata2301      Partition: 2    Leader: 3       Replicas: 3,1,0,2   Isr: 3,1,0,2
        Topic: bigdata2301      Partition: 3    Leader: 1       Replicas: 1,0,2,3   Isr: 1,0,2,3

?（3）停止掉 hadoop13 的 kafka 进程，并查看 Leader 分区情况

bin/kafka-server-stop.sh
 bin/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: 3

bin/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.index

kafka-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

?