14-Kafka-Day02

2023-12-15 12:52:02

第 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

image.png

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停止掉,会重新选举,两次查看的对比图如下:

image.png

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

image.png

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 分区情况

image.png

通过以上演示,大家可以发现,选举是按照AR(跟Replicas一样)进行的,而不是ISR

image.png

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

image.png

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

image.png

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

image.png

 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,读写请求很低,造成集群负载不均衡。

image.png

auto.leader.rebalance.enable,默认是true。 自动Leader Partition 平衡 
? leader.imbalance.per.broker.percentage, 默认是10%。每个broker允许的不平衡 的leader的比率。如果每个broker超过 了这个值,控制器会触发leader的平衡。 
? leader.imbalance.check.interval.seconds, 默认值300秒。检查leader负载是否平衡的间隔时间。

image.png

从以上可以看出,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 所有数据启用压缩策略

image.png

压缩后的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是把尽可能多的空闲内存都当做了磁盘缓存来使用

image.png

生产者将数据发送给kafka,kafka将数据交给Linux内核,Linux内核将数据放入自身操作系统的页缓存中,然后到一定值写入磁盘,假如消费者过来消费,直接从页缓存中,通过网卡发送给消费者,根本就没有去kafka的业务系统中获取数据,所以速度比较快。
跟这个问题非常像:mysql读取数据的速度为什么这么快!--buffer pool

?

文章来源:https://blog.csdn.net/qq_43819048/article/details/134878014
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。