kafka笔记（三）：broker-工作流程/节点服役和退役/副本/文件存储/高效读写

目录

kafka broker

kafka broker工作流程

zookeeper存储的kafka信息

kafka总体工作流程

broker重要参数

生产经验—节点服役和退役

服役新节点

退役旧节点

kafka副本

副本基本信息

leader选举流程

leader和follower故障处理细节

生产经验—手动调整分区副本存储

生产经验—leader partition负载均衡

生产经验—增加副本因子

文件存储

文件存储机制

文件清理策略

高效读写数据

kafka broker

kafka broker工作流程

zookeeper存储的kafka信息

（1）开启zookeeper，进入zookeeper，执行./zkCli.sh

 （2）查看信息

查看目录：ls /

其它的用ls也可以查看到

例：

 例：

 部分信息目录

查看controller路径上的数据

 get /controller

 查看/brokers/topics/first/partitions/0/state的数据

get /brokers/topics/first/partitions/0/state

kafka总体工作流程

broker重要参数

生产经验—节点服役和退役

服役新节点

（1）新节点准备

1）克隆出另一台机器

2）修改IP地址

vim /etc/sysconfig/network-scripts/ifcfg-eno16777736

3）修改主机名

vim /etc/hostname

4）修改新节点的kafka的broker.id

vim config/server.properties

5）删除新节点kafka下的datas和logs

rm -rf datas/* logs/*

6）启动zookeeper和kafka集群（启动过的不用重复启动）

（2）执行负载均衡

1）创建一个要负载均衡的主题

vim topics-to-move.json

内容： 

{
 "topics": [
 {"topic": "first"}
 ],
 "version": 1
}

2）生成一个负载均衡的计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop01:9092 
--topics-to-move-json-file topics-to-move.json 
--broker-list "0,1,2,3" --generate

3）创建副本存储计划

上一步操作会出现Proposed partition reassignment configuration

创建一个json文件，把Proposed partition reassignment configuration下的内容粘贴到这个文件中

4）执行副本存储计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop01:9092 
--reassignment-json-file xxxxxx.json --verify

5)验证副本存储计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop01:9092 
--reassignment-json-file XXXX.json --verify

退役旧节点

（1）执行负载均衡

1）创建一个要负载均衡的主题

{
 "topics": [
 {"topic": "first"}
 ],
 "version": 1
}

2）创建执行计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop01:9092 
--topics-to-move-json-file topics-to-move.json 
--broker-list "0,1,2" --generate

3)创建副本存储计划

上一步操作会出现Proposed partition reassignment configuration

创建一个json文件，把Proposed partition reassignment configuration下的内容粘贴到这个文件中

4）执行副本存储计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop01:9092 
--reassignment-json-file xxxxxx.json --verify

5)验证副本存储计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop01:9092 
--reassignment-json-file XXXX.json --verify

（2）执行停止命令

在要退出的节点上执行

 bin/kafka-server-stop.sh

kafka副本

副本基本信息

（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。

bin/kafka-topics.sh --bootstrap-server hadoop01:9092 
--create --topic first1 --partitions 4 --replication-factor 4

 bin/kafka-topics.sh --bootstrap-server hadoop01:9092 --describe 
--topic first1

#停止（测试节点）
bin/kafka-server-stop.sh
 
#查看分区情况（主节点）
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe 
--topic first1

bin/kafka-topics.sh --bootstrap-server hadoop01:9092 
--create --partitions 4 --replication-factor 2 --topic three

bin/kafka-topics.sh --bootstrap-server hadoop01:9092 --describe --topic three

{
"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]}]
}

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop01:9092 
--reassignment-json-file xxxxxx.json --execute

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop01:9092 
--reassignment-json-file xxxxxx.json --verify

bin/kafka-topics.sh --bootstrap-server hadoop01:9092 --describe --topic three

bin/kafka-topics.sh --bootstrap-server hadoop01:9092 
--create --partitions 3 --replication-factor 1 --topic four

{
"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]}]
}

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop01:9092 
--reassignment-json-file xxxxxx.json --execute

（2）topic存储数据的位置

1）查看hadoop01的topic的文件

 2）查看log日志

启动生产者发送数据

 bin/kafka-console-producer.sh --broker-list hadoop01:9092 --topic test

[root@hadoop01 test-0]# cat 00000000000000000002.log
=5V▒▒▒MwZ~▒MwZ~▒▒▒▒▒▒▒▒▒▒▒▒▒▒
hello=5▒Zh▒▒Mwh_▒Mwh_▒▒▒▒▒▒▒▒▒▒▒▒▒▒
spark=5]▒)0▒Mwp▒▒Mwp▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
kafka=5>{▒▒▒Mw~▒▒Mw~▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒

3）查看index（直接查看不能查看，要使用工具查看）

kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000002.index

[root@hadoop01 test-0]# kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000002.index
Dumping ./00000000000000000002.index
offset: 2 position: 0
 

4）使用工具查看log

kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000002.log

（3）index和log

index为稀疏索引，大约每往log文件写入4kb数据，会往index文件写入一条索引。

参数log.index.interval.bytes默认4kb。

Index文件中保存的offset为相对offset，这样能确保offset的值所占空间不会过大，因此能将offset的值控制在固定大小。

文件清理策略

Kafka中默认的日志保存时间为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，表示无穷大。

（2）compact日志压缩

compact日志压缩：对于相同key的不同value值，只保留最后一个版本。

log.cleanup.policy = compact：所有数据启用压缩策略

压缩后的offset可能是不连续的，这种策略只适合特殊场景，比如消息的key是用户ID，value是用户的资料，通过这种压缩策略，整个消息集里就保存了所有用户最新的资料。

高效读写数据

（1）kafka本身是分布式集群，可以采用分布式技术，并行度高

（2）读取数据采用稀疏索引，可以快速定位要消费的数据

（3）顺序写磁盘

Kafka的producer生产的数据要写入到log文件中，写的过程是一直追加到文件末端，为顺序写。

官网有数据表明，同样的磁盘，顺序写能到600M/s，而随机写只有100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。

（4）页缓存（PageCache）+零拷贝

零拷贝：Kafka的数据加工处理操作交由Kafka生产者和Kafka消费者处理。Kafka Broker应用层不关心存储的数据，所以就不用走应用层，传输效率高。

页缓存：Kafka重度依赖底层操作系统提供的PageCache功 能。当上层有写操作时，操作系统只是将数据写入PageCache。当读操作发生时，先从PageCache中查找，如果找不到，再去磁盘中读取。实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用。

本文为学习笔记！！！