3、Kafka Broker

野性酷女 2024-03-27 18:42 80阅读 0赞

4.1 Kafka Broker 工作流程  
4.1.1 Zookeeper 存储的 Kafka 信息  
（1）启动 Zookeeper 客户端。

[hadoop102 zookeeper-3.5.7]$ bin/zkCli.sh

（2）通过 ls 命令可以查看 kafka 相关信息。

[zk: localhost:2181(CONNECTED) 2] ls /kafka

![在这里插入图片描述][b8d5b420ace94402a2d25891a0c54420.png]  
4.1.2 Kafka Broker 总体工作流程  
![在这里插入图片描述][dbbd767e6ab0429eae1fbb7253fb4c0b.png]  
1）模拟 Kafka 上下线，Zookeeper 中数据变化  
（1）查看/kafka/brokers/ids 路径上的节点。

[zk: localhost:2181(CONNECTED) 2] ls /kafka/brokers/ids
    [0, 1, 2]

（2）查看/kafka/controller 路径上的数据。

[zk: localhost:2181(CONNECTED) 15] get /kafka/controller
    {
        "version":1,"brokerid":0,"timestamp":"1637292471777"}

（3）查看/kafka/brokers/topics/first/partitions/0/state 路径上的数据

[zk: localhost:2181(CONNECTED) 16] 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。

[hadoop104 kafka]$ bin/kafka-server-stop.sh

（5）再次查看/kafka/brokers/ids 路径上的节点。

[zk: localhost:2181(CONNECTED) 3] ls /kafka/brokers/ids
    [0, 1]

（6）再次查看/kafka/controller 路径上的数据。

[zk: localhost:2181(CONNECTED) 15] get /kafka/controller
    {
        "version":1,"brokerid":0,"timestamp":"1637292471777"}

（7）再次查看/kafka/brokers/topics/first/partitions/0/state 路径上的数据。

[zk: localhost:2181(CONNECTED) 16] get 
    /kafka/brokers/topics/first/partitions/0/state
    {
        "controller_epoch":24,"leader":0,"version":1,"leader_epoch":18,"
    isr":[0,1]}

（8）启动 hadoop104 上的 kafka。

[hadoop104 kafka]$ bin/kafka-server-start.sh -
    daemon ./config/server.properties

（9）再次观察（1）、（2）、（3）步骤中的内容  
4.1.3 Broker 重要参数  
![在这里插入图片描述][4a8638eb17f142a9b4eac0bdf484a847.png]  
2. 节点服役和退役  
2.1 服役新节点  
1）新节点准备

（1）关闭 hadoop104，并右键执行克隆操作。  
（2）开启 hadoop105，并修改 IP 地址。

[root@hadoop104 ~]# vim /etc/sysconfig/network-scripts/ifcfgens33
    DEVICE=ens33
    TYPE=Ethernet
    ONBOOT=yes
    BOOTPROTO=static
    NAME="ens33"
    IPADDR=192.168.10.105
    PREFIX=24
    GATEWAY=192.168.10.2
    DNS1=192.168.10.2

（3）在 hadoop105 上，修改主机名称为 hadoop105。

[root@hadoop104 ~]# vim /etc/hostname
    hadoop105

（4）重新启动 hadoop104、hadoop105。  
（5）修改 haodoop105 中 kafka 的 broker.id 为 3。  
（6）删除 hadoop105 中 kafka 下的 datas 和 logs。

[hadoop105 kafka]$ rm -rf datas/* logs/*

（7）启动 hadoop102、hadoop103、hadoop104 上的 kafka 集群。

[hadoop102 ~]$ zk.sh start
    [hadoop102 ~]$ kf.sh start

（8）单独启动 hadoop105 中的 kafka。

[hadoop105 kafka]$ bin/kafka-server-start.sh -
    daemon ./config/server.properties

2）执行负载均衡操作  
（1）创建一个要均衡的主题。

[hadoop102 kafka]$ vim topics-to-move.json
    {
        
     "topics": [
     {
        "topic": "first"}
     ],
     "version": 1
    }

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

[hadoop102 kafka]$ 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,"replic
    as":[0,2,1],"log_dirs":["any","any","any"]},{"topic":"first","par
    tition":1,"replicas":[2,1,0],"log_dirs":["any","any","any"]},{"to
    pic":"first","partition":2,"replicas":[1,0,2],"log_dirs":["any","
    any","any"]}]}
    Proposed partition reassignment configuration
    {
        "version":1,"partitions":[{
        "topic":"first","partition":0,"replic
    as":[2,3,0],"log_dirs":["any","any","any"]},{"topic":"first","par
    tition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"to
    pic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","
    any","any"]}]}

（3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2、broker3 中）。

[hadoop102 kafka]$ vim increase-replication-factor.json

输入如下内容：

{
        "version":1,"partitions":[{
        "topic":"first","partition":0,"replic
    as":[2,3,0],"log_dirs":["any","any","any"]},{"topic":"first","par
    tition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"to
    pic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","
    any","any"]}]}

（4）执行副本存储计划。  
\[atguigu@hadoop102 kafka\]$ bin/kafka-reassign-partitions.sh –  
bootstrap-server hadoop102:9092 --reassignment-json-file  
increase-replication-factor.json --execute  
（5）验证副本存储计划。

[hadoop102 kafka]$ 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

4.2.2 退役旧节点  
1）执行负载均衡操作  
先按照退役一台节点，生成执行计划，然后按照服役时操作流程执行负载均衡。  
（1）创建一个要均衡的主题。

[hadoop102 kafka]$ vim topics-to-move.json
    {
        
     "topics": [
     {
        "topic": "first"}
     ],
     "version": 1
    }

（2）创建执行计划。

[hadoop102 kafka]$ 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,"replic
    as":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","par
    tition":1,"replicas":[3,1,2],"log_dirs":["any","any","any"]},{"to
    pic":"first","partition":2,"replicas":[0,2,3],"log_dirs":["any","
    any","any"]}]}
    Proposed partition reassignment configuration
    {
        "version":1,"partitions":[{
        "topic":"first","partition":0,"replic
    as":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","par
    tition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"to
    pic":"first","partition":2,"replicas":[1,2,0],"log_dirs":["any","
    any","any"]}]}

（3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2 中）。

[hadoop102 kafka]$ vim increase-replication-factor.json
    {
        "version":1,"partitions":[{
        "topic":"first","partition":0,"replic
    as":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","par
    tition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"to
    pic":"first","partition":2,"replicas":[1,2,0],"log_dirs":["any","
    any","any"]}]}

（4）执行副本存储计划。

[hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --
    bootstrap-server hadoop102:9092 --reassignment-json-file 
    increase-replication-factor.json --execute

（5）验证副本存储计划。

2）执行停止命令  
在 hadoop105 上执行停止命令即可。

[hadoop105 kafka]$ 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 分区中的所有副本统称为 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 的  
![在这里插入图片描述][4c137984b74a41a199b591bce02eae42.png]  
（1）创建一个新的 topic，4 个分区，4 个副本

[hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server 
    hadoop102:9092 --create --topic atguigu1 --partitions 4 --replication-factor 
    4
    Created topic atguigu1.

（2）查看 Leader 分布情况

[hadoop102 kafka]$ 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 分区情况

[hadoop105 kafka]$ bin/kafka-server-stop.sh
    [hadoop102 kafka]$ 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
    pic: 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 分区情况

[hadoop104 kafka]$ bin/kafka-server-stop.sh
    [hadoop102 kafka]$ 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 分区情况

[hadoop105 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties
    [hadoop102 kafka]$ 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 分区情况

[hadoop104 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties
    [hadoop102 kafka]$ 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 分区情况

[hadoop103 kafka]$ bin/kafka-server-stop.sh
    [hadoop102 kafka]$ 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

4.3.3 Leader 和 Follower 故障处理细节  
![在这里插入图片描述][2ce01c009cf749b289ef107c67376a88.png]  
![在这里插入图片描述][212ddecc2ba940b6be8f5d32f85c4276.png]  
4.3.4 分区副本分配  
如果 kafka 服务器只有 4 个节点，那么设置 kafka 的分区数大于服务器台数，在 kafka  
底层如何分配存储副本呢？  
1）创建 16 分区，3 个副本  
（1）创建一个新的 topic，名称为 second。

[hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server 
    hadoop102:9092 --create --partitions 16 --replication-factor 3 --
    topic second

（2）查看分区和副本情况。

[hadoop102 kafka]$ 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

![在这里插入图片描述][ef1ce76f27d84ddbb6d7ed603a5cf63e.png]  
4.3.5 生产经验——手动调整分区副本存储  
生产经验——手动调整分区副本存储  
在生产环境中，每台服务器的配置和性能不一致，但是Kafka只会根据自己的代码规则创建对应的分区副  
本，就会导致个别服务器存储压力较大。所有需要手动调整分区副本的存储。

需求：创建一个新的topic，4个分区，两个副本，名称为three。将 该topic的所有副本都存储到broker0和  
broker1两台服务器上。  
![在这里插入图片描述][01f52df8c7b24dfa9f09050f6f847d7c.png]  
手动调整分区副本存储的步骤如下：  
（1）创建一个新的 topic，名称为 three。

[hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server 
    hadoop102:9092 --create --partitions 4 --replication-factor 2 --
    topic three

（2）查看分区副本存储情况。

[hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server 
    hadoop102:9092 --describe --topic three

（3）创建副本存储计划（所有副本都指定存储在 broker0、broker1 中）。

[hadoop102 kafka]$ vim 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）执行副本存储计划。

[hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --
    bootstrap-server hadoop102:9092 --reassignment-json-file 
    increase-replication-factor.json --execute

（5）验证副本存储计划。

[hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --
    bootstrap-server hadoop102:9092 --reassignment-json-file 
    increase-replication-factor.json --verify

（6）查看分区副本存储情况。

[hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server 
    hadoop102:9092 --describe --topic three

4.3.6 生产经验——Leader Partition 负载平衡  
![在这里插入图片描述][12c81f81364c4698bd82d6ce8063e70d.png]  
![在这里插入图片描述][ec6f18c5cd7b4b56a598b9319ffbe4a0.png]  
4.3.7 生产经验——增加副本因子  
在生产环境当中，由于某个主题的重要等级需要提升，我们考虑增加副本。副本数的  
增加需要先制定计划，然后根据计划执行。  
1）创建 topic

[hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server 
    hadoop102:9092 --create --partitions 3 --replication-factor 1 --
    topic four

2）手动增加副本存储  
（1）创建副本存储计划（所有副本都指定存储在 broker0、broker1、broker2 中）。

[hadoop102 kafka]$ vim increase-replication-factor.json

输入如下内容：

{
        "version":1,"partitions":[{
        "topic":"four","partition":0,"replica
    s":[0,1,2]},{"topic":"four","partition":1,"replicas":[0,1,2]},{"t
    opic":"four","partition":2,"replicas":[0,1,2]}]}

（2）执行副本存储计划。

[hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --
    bootstrap-server hadoop102:9092 --reassignment-json-file 
    increase-replication-factor.json --execute

4.4 文件存储  
4.4.1 文件存储机制  
1）Topic 数据的存储机制  
![在这里插入图片描述][591bf6154db84eb48716091682933c91.png]  
2）思考：Topic 数据到底存储在什么位置？  
（1）启动生产者，并发送消息。

[hadoop102 kafka]$ bin/kafka-console-producer.sh --
    bootstrap-server hadoop102:9092 --topic first
    >hello world

（2）查看 hadoop102（或者 hadoop103、hadoop104）的/opt/module/kafk  
a/datas/first-1（first-0、first-2）路径上的文件。

[hadoop104 first-1]$ ls
    00000000000000000092.index
    00000000000000000092.log
    00000000000000000092.snapshot
    00000000000000000092.timeindex
    leader-epoch-checkpoint
    partition.metadata

（3）直接查看 log 日志，发现是乱码。

（4）通过工具查看 index 和 log 信息。

[hadoop104 first-1]$ kafka-run-class.sh kafka.tools.DumpLogSegments 
    --files ./00000000000000000000.index 
    Dumping ./00000000000000000000.index
    offset: 3 position: 152
    [atguigu@hadoop104 first-1]$ kafka-run-class.sh kafka.tools.DumpLogSegments 
    --files ./00000000000000000000.log
    Dumping datas/first-0/00000000000000000000.log
    Starting offset: 0
    baseOffset: 0 lastOffset: 1 count: 2 baseSequence: -1 lastSequence: -1 producerId: -1 
    producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 
    0 CreateTime: 1636338440962 size: 75 magic: 2 compresscodec: none crc: 2745337109 isvalid: 
    true
    baseOffset: 2 lastOffset: 2 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 
    producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 
    75 CreateTime: 1636351749089 size: 77 magic: 2 compresscodec: none crc: 273943004 isvalid: 
    true
    baseOffset: 3 lastOffset: 3 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 
    producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 
    152 CreateTime: 1636351749119 size: 77 magic: 2 compresscodec: none crc: 106207379 isvalid: 
    true
    baseOffset: 4 lastOffset: 8 count: 5 baseSequence: -1 lastSequence: -1 producerId: -1 
    producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 
    229 CreateTime: 1636353061435 size: 141 magic: 2 compresscodec: none crc: 157376877 isvalid: 
    true
    baseOffset: 9 lastOffset: 13 count: 5 baseSequence: -1 lastSequence: -1 producerId: -1 
    producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 
    370 CreateTime: 1636353204051 size: 146 magic: 2 compresscodec: none crc: 4058582827 isvalid: 
    true

3）index 文件和 log 文件详解  
![在这里插入图片描述][2f0452a9e3ac41439a5a161f470b0fbe.png]  
![在这里插入图片描述][7927c8c9140341e4b9a08a4efceacf8d.png]  
4.4.2 文件清理策略  
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，表示无穷大。  
思考：如果一个 segment 中有一部分数据过期，一部分没有过期，怎么处理？  
![在这里插入图片描述][18c9bff84b3141f9a0a87f628d8d0704.png]  
2）compact 日志压缩  
![在这里插入图片描述][17bed295fe274a81aa5fa7d5cb034455.png]  
4.5 高效读写数据  
1）Kafka 本身是分布式集群，可以采用分区技术，并行度高  
2）读数据采用稀疏索引，可以快速定位要消费的数据  
3）顺序写磁盘  
Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程是一直追加到文件末端，  
为顺序写。官网有数据表明，同样的磁盘，顺序写能到 600M/s，而随机写只有 100K/s。这  
与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。  
![在这里插入图片描述][25a9bba903dd421192408cb596b25c6d.png]

4）页缓存 + 零拷贝技术  
![在这里插入图片描述][bb3829989f6e4895b8814056da5450b2.png]  
![在这里插入图片描述][69fe82bacced4b168222728aedd7857e.png]

[b8d5b420ace94402a2d25891a0c54420.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/67e7a78484f24a12ba8cd531c4d85c5a.png
[dbbd767e6ab0429eae1fbb7253fb4c0b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/8f0b7fec4a854654860451c5aa5d3ac0.png
[4a8638eb17f142a9b4eac0bdf484a847.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/4e3f9accc3d448eeb11ae9e890410171.png
[4c137984b74a41a199b591bce02eae42.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/222b026170484859a546a0106f9f9e8b.png
[2ce01c009cf749b289ef107c67376a88.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/a898d33cdba14a0ba8468426402ca8f9.png
[212ddecc2ba940b6be8f5d32f85c4276.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/3272516e2add47e2a7043726887f24d4.png
[ef1ce76f27d84ddbb6d7ed603a5cf63e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/8e00a574527e47489f7d248e859ff364.png
[01f52df8c7b24dfa9f09050f6f847d7c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/1b1d6ddf165c4947b69a14a6dd378304.png
[12c81f81364c4698bd82d6ce8063e70d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/4f47beb977c34c3984982b272c28a1e0.png
[ec6f18c5cd7b4b56a598b9319ffbe4a0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/0c994775123a4428afa490e595e3645c.png
[591bf6154db84eb48716091682933c91.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/a7f2837d35054c0193f4a0b6f1c1e332.png
[2f0452a9e3ac41439a5a161f470b0fbe.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/dbd5bca7a1fc4a4f95197287f4cc17c5.png
[7927c8c9140341e4b9a08a4efceacf8d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/9bbe07f2513044ca86d1ac5798323c99.png
[18c9bff84b3141f9a0a87f628d8d0704.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/3d9cf8003d5946d19c0857a53d36c7bd.png
[17bed295fe274a81aa5fa7d5cb034455.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/cfdbb183f806426aa8011f5d923afb18.png
[25a9bba903dd421192408cb596b25c6d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/8aaf9b8846cc442b866e3e3509ddd575.png
[bb3829989f6e4895b8814056da5450b2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/da62ebdc78394ca89babc21e56912e92.png
[69fe82bacced4b168222728aedd7857e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/5929369e40c04c58ae2dfe27c660a5e7.png