学习链接：https://www.bilibili.com/video/BV1vr4y1677k?p=22&spm_id_from=pageDriver&vd_source=b9e4f35102d61e6d02e0a5e1bbfea480

1 Kafka Broker工作流程

1.1 Zookeeper存储的Kafka信息

  1. 启动Zookeeper客户端

    1. [qtbhy@hadoop104 zookeeper-3.5.7]$ bin/zkCli.sh

  2. ls命令查看kafka相关信息

    1. [zk: localhost:2181(CONNECTED) 0] ls /

    QQ截图20220704095508.png

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

    QQ截图20220704095702.png

    1. [zk: localhost:2181(CONNECTED) 3] ls /kafka/brokers
    2. [ids, seqid, topics]
    3. [zk: localhost:2181(CONNECTED) 4] ls /kafka/brokers/ids
    4. [0, 1, 2]
  • Zookeeper中存储的Kafka信息

QQ截图20220704100250.png

  • 查看zookeeper工具——prettyzoo

QQ截图20220704102151.png

1.2 Kafka Broker总体工作流程

QQ截图20220704101536.png

  • Kafka上下线

    • 停hadoop104上的kafka

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

      ids中Hadoop102对应的2没了
      QQ截图20220704102345.png

    • 开启hadoop104上的kafka

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

      2正常
      QQ截图20220704102528.png

      2 节点服役和退役

      2.1 服役新节点

  1. 关闭hadoop104，克隆，拷贝成hadoop105

  2. 开启hadoop105，并修改IP地址 IPADDR=192.168.10.105

    1. [root@hadoop104 ~]# vim /etc/sysconfig/network-scripts/ifcfg-ens33

  3. 在hadoop105上，修改主机名称为hadoop105

    1. [root@hadoop104 ~]# vim /etc/hostname

  4. 重启hadoop104、hadoop105

  5. 修改hadoop105中的kafka的broker.id为3

在/opt/module/kafka/config中修改server.properties 

  1. [qtbhy@hadoop105 config]$ vim server.properties

  1. 删除hadoop105中kafka下的datas和logs

    1. [qtbhy@hadoop105 datas]$ rm -rf datas/* logs/*

  2. 启动集群和zookeeper

    1. [qtbhy@hadoop104 zookeeper-3.5.7]$ bin/zkServer.sh start

  3. 启动hadoop105的kafka

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

  4. 负载均衡

    1. 创建一个要均衡的主题

      1. [qtbhy@hadoop102 kafka]$ vim topics-to-move.json

      QQ截图20220704113256.png

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

      1. [qtbhy@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

      QQ截图20220704113455.png

    3. 创建副本存储计划，所有副本存储在 broker0、broker1、broker2、broker3 中

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

      QQ截图20220704113707.png

    4. 执行副本存储计划

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

    5. 验证副本存储计划 ```shell [qtbhy@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

  1. <a name="Mg2By"></a>
  2. ## 2.2 退役旧节点
  4. 1. 执行负载均衡操作，先按照退役一台节点，生成执行计划
  5.    1. 创建一个要均衡的主题
  6. ```shell
  7. [qtbhy@hadoop102 kafka]$ vim topics-to-move.json

QQ截图20220704113256.png

  1. 创建执行计划 ```shell [qtbhy@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,”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”]}]}

Proposed partition reassignment configuration {“version”:1,”partitions”:[{“topic”:”first”,”partition”:0,”replicas”:[1,0,2],”log_dirs”:[“any”,”any”,”any”]},{“topic”:”first”,”partition”:1,”replicas”:[2,1,0],”log_dirs”:[“any”,”any”,”any”]},{“topic”:”first”,”partition”:2,”replicas”:[0,2,1],”log_dirs”:[“any”,”any”,”any”]}]}

  2.    3. 创建副本存储计划 所有副本存储在 broker0、broker1、broker2 中
  3. ```shell
  4. [qtbhy@hadoop102 kafka]$ vim increase-replication-factor.json
  1. {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[1,0,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[2,1,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,2,1],"log_dirs":["any","any","any"]}]}
  1. 执行副本存储计划 ```shell [qtbhy@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh —bootstrap-server hadoop102:9092 —reassignment-json-file increase-replication-factor.json —execute Current partition replica assignment

{“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”]}]}

Save this to use as the —reassignment-json-file option during rollback Successfully started partition reassignments for first-0,first-1,first-2

  2.    5. 验证副本存储计划
  3. ```shell
  4. [qtbhy@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify
  5. Status of partition reassignment:
  6. Reassignment of partition first-0 is complete.
  7. Reassignment of partition first-1 is complete.
  8. Reassignment of partition first-2 is complete.
  10. Clearing broker-level throttles on brokers 0,1,2,3
  11. Clearing topic-level throttles on topic first

  1. 在hadoop105上执行停止命令

    1. [qtbhy@hadoop105 kafka]$ bin/kafka-server-stop.sh

    3 Kafka副本

    3.1 副本基本信息

  2. Kafka 副本作用：提高数据可靠性。

  3. Kafka 默认副本 1 个，生产环境一般配置为 2 个，保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率。
  4. Kafka 中副本分为：Leader 和 Follower。Kafka 生产者只会把数据发往 Leader，然后 Follower 找 Leader 进行同步数据。
  5. 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 的。

  • Leader选举流程

QQ截图20220704213116.png

  1. 创建一个新的topic，4个分区，4个副本

    1. [qtbhy@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic second --partitions 4 --replication-factor 4
    2. Created topic second.

  2. 查看Leader分布情况

    1. [qtbhy@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic second
    2. Topic: second    TopicId: HaoBRpXNTluDEx0bS9U0Dg    PartitionCount: 4    ReplicationFactor: 4    Configs: segment.bytes=1073741824
    3.  Topic: second    Partition: 0    Leader: 3    Replicas: 3,1,0,2 Isr: 3,1,0,2
    4.  Topic: second    Partition: 1    Leader: 1    Replicas: 1,0,2,3 Isr: 1,0,2,3
    5.  Topic: second    Partition: 2    Leader: 0    Replicas: 0,2,3,1 Isr: 0,2,3,1
    6.  Topic: second    Partition: 3    Leader: 2    Replicas: 2,3,1,0 Isr: 2,3,1,0

  3. 停hadoop105

    1. [qtbhy@hadoop105 kafka]$ bin/kafka-server-stop.sh

    QQ截图20220704210712.png

  4. 停hadoop104

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

    QQ截图20220704210852.png

  5. 恢复hadoop105

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

    QQ截图20220704211021.png

  6. 恢复hadoop104

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

    QQ截图20220704211133.png
    在Isr中存活且在AR中排在最前

    3.3 Leader和Follower故障处理

    LEO（Log End Offset）：每个副本的最后一个offset，LEO其实就是最新的offset + 1。
    HW（High Watermark）：所有副本中最小的LEO 。
    1）Follower故障
    QQ截图20220704213505.png
    （1）Follower发生故障后会被临时踢出ISR
    （2）这个期间Leader和Follower继续接收数据
    （3）待该Follower恢复后，Follower会读取本地磁盘记录的上次的HW，并将log文件高于HW的部分截取掉，从HW开始向Leader进行同步。
    （4）等该Follower的LEO大于等于该Partition的HW，即Follower追上Leader之后，就可以重新加入ISR了。
    2）Leader故障
    QQ截图20220704213625.png
    （1）Leader发生故障之后，会从ISR中选出一个新的Leader
    （2）为保证多个副本之间的数据一致性，其余的Follower会先将各自的log文件高于HW的部分截掉，然后从新的Leader同步数据。
    注意：这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。

    3.4 分区副本分配

    分区数大于服务器台数：

  7. 创建一个新的topic，16个分区，3个副本

    1. [qtbhy@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic third --partitions 16 --replication-factor 3
    2. Created topic third.

  8. 查看分区和副本情况

    1. [qtbhy@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic third
    2. Topic: third    TopicId: j9fu0f50QkCgRIfFlhNwLg    PartitionCount: 16    ReplicationFactor: 3    Configs: segment.bytes=1073741824
    3.  Topic: third    Partition: 0    Leader: 1    Replicas: 1,0,2    Isr: 1,0,2
    4.  Topic: third    Partition: 1    Leader: 0    Replicas: 0,2,3    Isr: 0,2,3
    5.  Topic: third    Partition: 2    Leader: 2    Replicas: 2,3,1    Isr: 2,3,1
    6.  Topic: third    Partition: 3    Leader: 3    Replicas: 3,1,0    Isr: 3,1,0
    7.  Topic: third    Partition: 4    Leader: 1    Replicas: 1,2,3    Isr: 1,2,3
    8.  Topic: third    Partition: 5    Leader: 0    Replicas: 0,3,1    Isr: 0,3,1
    9.  Topic: third    Partition: 6    Leader: 2    Replicas: 2,1,0    Isr: 2,1,0
    10.  Topic: third    Partition: 7    Leader: 3    Replicas: 3,0,2    Isr: 3,0,2
    11.  Topic: third    Partition: 8    Leader: 1    Replicas: 1,3,0    Isr: 1,3,0
    12.  Topic: third    Partition: 9    Leader: 0    Replicas: 0,1,2    Isr: 0,1,2
    13.  Topic: third    Partition: 10    Leader: 2    Replicas: 2,0,3    Isr: 2,0,3
    14.  Topic: third    Partition: 11    Leader: 3    Replicas: 3,2,1    Isr: 3,2,1
    15.  Topic: third    Partition: 12    Leader: 1    Replicas: 1,0,2    Isr: 1,0,2
    16.  Topic: third    Partition: 13    Leader: 0    Replicas: 0,2,3    Isr: 0,2,3
    17.  Topic: third    Partition: 14    Leader: 2    Replicas: 2,3,1    Isr: 2,3,1
    18.  Topic: third    Partition: 15    Leader: 3    Replicas: 3,1,0    Isr: 3,1,0

    QQ截图20220704213955.png

    3.5 手动调整分区副本存储

    在生产环境中，每台服务器的配置和性能不一致，但是Kafka只会根据自己的代码规则创建对应的分区副本，就会导致个别服务器存储压力较大。所有需要手动调整分区副本的存储。
    需求：创建一个新的topic，4个分区，两个副本，名称为three。将 该topic的所有副本都存储到broker0和broker1两台服务器上。
    QQ截图20220704214132.png

  9. 创建一个新的topic，fourth

    1. [qtbhy@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic fourth --partitions 4 --replication-factor 2
    2. Created topic fourth.

  10. 查看分区副本存储情况

    1. [qtbhy@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic fourth
    2. Topic: fourth    TopicId: -Q_9bxfjTi6Fl3AjFtoSXA    PartitionCount: 4    ReplicationFactor: 2    Configs: segment.bytes=1073741824
    3.  Topic: fourth    Partition: 0    Leader: 0    Replicas: 0,1    Isr: 0,1
    4.  Topic: fourth    Partition: 1    Leader: 2    Replicas: 2,0    Isr: 2,0
    5.  Topic: fourth    Partition: 2    Leader: 3    Replicas: 3,2    Isr: 3,2
    6.  Topic: fourth    Partition: 3    Leader: 1    Replicas: 1,3    Isr: 1,3

  11. 创建副本存储计划，副本指定存在broker0、broker1中

    1. [qtbhy@hadoop102 kafka]$  vim increase-replication-factor.json
    1. {
    2. "version":1,
    3. "partitions":[{"topic":"fourth","partition":0,"replicas":[0,1]},
    4. {"topic":"fourth","partition":1,"replicas":[0,1]},
    5. {"topic":"fourth","partition":2,"replicas":[1,0]},
    6. {"topic":"fourth","partition":3,"replicas":[1,0]}] 
    7. }

  12. 执行副本存储计划 ```shell [qtbhy@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh —bootstrap-server hadoop102:9092 —reassignment-json-file increase-replication-factor.json —execute Current partition replica assignment

{“version”:1,”partitions”:[{“topic”:”fourth”,”partition”:0,”replicas”:[0,1],”log_dirs”:[“any”,”any”]},{“topic”:”fourth”,”partition”:1,”replicas”:[2,0],”log_dirs”:[“any”,”any”]},{“topic”:”fourth”,”partition”:2,”replicas”:[3,2],”log_dirs”:[“any”,”any”]},{“topic”:”fourth”,”partition”:3,”replicas”:[1,3],”log_dirs”:[“any”,”any”]}]}

Save this to use as the —reassignment-json-file option during rollback Successfully started partition reassignments for fourth-0,fourth-1,fourth-2,fourth-3

  2. 5. 验证副本存储计划
  3. ```shell
  4. [qtbhy@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify
  5. Status of partition reassignment:
  6. Reassignment of partition fourth-0 is complete.
  7. Reassignment of partition fourth-1 is complete.
  8. Reassignment of partition fourth-2 is complete.
  9. Reassignment of partition fourth-3 is complete.
  11. Clearing broker-level throttles on brokers 0,1,2,3
  12. Clearing topic-level throttles on topic fourth
  1. 查看分区副本存储情况
    1. [qtbhy@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic fourth
    2. Topic: fourth    TopicId: -Q_9bxfjTi6Fl3AjFtoSXA    PartitionCount: 4    ReplicationFactor: 2    Configs: segment.bytes=1073741824
    3.  Topic: fourth    Partition: 0    Leader: 0    Replicas: 0,1    Isr: 0,1
    4.  Topic: fourth    Partition: 1    Leader: 0    Replicas: 0,1    Isr: 0,1
    5.  Topic: fourth    Partition: 2    Leader: 1    Replicas: 1,0    Isr: 1,0
    6.  Topic: fourth    Partition: 3    Leader: 1    Replicas: 1,0    Isr: 1,0

    3.6 Leader Partition负载平衡

    正常情况下，Kafka本身会自动把Leader Partition均匀分散在各个机器上，来保证每台机器的读写吞吐量都是均匀的。但是如果某些broker宕机，会导致Leader Partition过于集中在其他少部分几台broker上，这会导致少数几台broker的读写请求压力过高，其他宕机的broker重启之后都是follower partition，读写请求很低，造成集群负载不均衡。
    QQ截图20220705104555.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负载是否平衡的间隔时间。

QQ截图20220705104740.png
针对broker0节点，分区2的AR优先副本是0节点，但是0节点却不是Leader节点，不平衡数加1，AR副本总数是4=>broker0节点不平衡率为1/4>10%，需要再平衡

3.7 增加副本因子

  1. 创建topic

    1. [qtbhy@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic fifth --partitions 3 --replication-factor 1
    2. Created topic fifth.

    :::danger 不能通过命令行增加副本 :::

  2. 手动增加副本存储

    1. 创建副本存储计划

      1. [qtbhy@hadoop102 kafka]$ vim increase-replication-factor.json
      1. {"version":1,"partitions":[{"topic":"fifth","partition":0,"replicas":[0,1,2]},{"topic":"fifth","partition":1,"replicas":[0,1,2]},{"topic":"fifth","partition":2,"replicas":[0,1,2]}]}

    2. 执行副本存储计划 ```shell [qtbhy@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh —bootstrap-server hadoop102:9092 —reassignment-json-file increase-replication-factor.json —execute Current partition replica assignment

{“version”:1,”partitions”:[{“topic”:”fifth”,”partition”:0,”replicas”:[0],”log_dirs”:[“any”]},{“topic”:”fifth”,”partition”:1,”replicas”:[2],”log_dirs”:[“any”]},{“topic”:”fifth”,”partition”:2,”replicas”:[1],”log_dirs”:[“any”]}]}

Save this to use as the —reassignment-json-file option during rollback Successfully started partition reassignments for fifth-0,fifth-1,fifth-2

  1. <a name="BrTRl"></a>
  2. # 4 文件存储
  3. <a name="Bguvw"></a>
  4. ## 4.1 文件存储机制
  6. 1. Topic数据的存储机制
  8. Topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是Producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制， 将每个partition分为多个segment。每个segment包括：“.index”文件、“.log”文件和.timeindex等文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号，例如：first-0。<br />![QQ截图20220705114043.png](https://cdn.nlark.com/yuque/0/2022/png/26273875/1656992456733-123c7889-6b4b-429b-817b-04288959516c.png#clientId=u6bed6d30-c4eb-4&crop=0&crop=0&crop=1&crop=1&from=ui&id=u3d4ebe7f&margin=%5Bobject%20Object%5D&name=QQ%E6%88%AA%E5%9B%BE20220705114043.png&originHeight=536&originWidth=1220&originalType=binary&ratio=1&rotation=0&showTitle=false&size=109307&status=done&style=none&taskId=ub79b307b-5ded-4e2f-8895-c34d80b6b36&title=)
  10. 2. Topic数据存储位置
  11.    1. 查看hadoop102的/opt/moudule/kafka/datas/first-0的index
  12. ```shell
  13. [qtbhy@hadoop102 kafka]$ bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files ./datas/first-0/00000000000000000000.index
  14. Dumping ./datas/first-0/00000000000000000000.index
  15. offset: 0 position: 0
  1. 查看log信息
    1. [qtbhy@hadoop102 kafka]$ bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files ./datas/first-0/00000000000000000000.log
    2. Dumping ./datas/first-0/00000000000000000000.log
    3. Starting offset: 0
    4. baseOffset: 0 lastOffset: 0 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 1 isTransactional: false isControl: false position: 0 CreateTime: 1656647101364 size: 73 magic: 2 compresscodec: none crc: 1441072200 isvalid: true
    5. baseOffset: 1 lastOffset: 1 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 1 isTransactional: false isControl: false position: 73 CreateTime: 1656647253396 size: 73 magic: 2 compresscodec: none crc: 396373320 isvalid: true
    6. baseOffset: 2 lastOffset: 6 count: 5 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 1 isTransactional: false isControl: false position: 146 CreateTime: 1656675667168 size: 126 magic: 2 compresscodec: none crc: 2968610005 isvalid: true
    7. baseOffset: 7 lastOffset: 7 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 1 isTransactional: false isControl: false position: 272 CreateTime: 1656676628375 size: 74 magic: 2 compresscodec: none crc: 2942310478 isvalid: true
    8. baseOffset: 8 lastOffset: 8 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 1 isTransactional: false isControl: false position: 346 CreateTime: 1656676628408 size: 74 magic: 2 compresscodec: none crc: 685671724 isvalid: true
    9. baseOffset: 9 lastOffset: 13 count: 5 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 1 isTransactional: false isControl: false position: 420 CreateTime: 1656850856607 size: 126 magic: 2 compresscodec: none crc: 3542517371 isvalid: true
    10. baseOffset: 14 lastOffset: 18 count: 5 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 1 isTransactional: false isControl: false position: 546 CreateTime: 1656850920514 size: 126 magic: 2 compresscodec: none crc: 2371857795 isvalid: true
    11. baseOffset: 19 lastOffset: 23 count: 5 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 1 isTransactional: false isControl: false position: 672 CreateTime: 1656851851995 size: 131 magic: 2 compresscodec: none crc: 2420501260 isvalid: true
    12. baseOffset: 24 lastOffset: 28 count: 5 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 1 isTransactional: false isControl: false position: 803 CreateTime: 1656851864298 size: 136 magic: 2 compresscodec: none crc: 786779293 isvalid: true
  1. index文件和log文件

QQ截图20220705114700.png

参数 描述
log.segment.bytes Kafka 中 log 日志是分成一块块存储的，此配置是指 log 日志划分成块的大小，默认值 1G。
log.index.interval.bytes 默认 4kb，kafka 里面每当写入了 4kb 大小的日志（.log），然后就往 index 文件里面记录一个索引。 稀疏索引。

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，表示无穷大。
2）compact日志压缩
compact日志压缩：对于相同key的不同value值，只保留最后一个版本
log.cleanup.policy = compact 所有数据启用压缩策略
QQ截图20220705115708.png
压缩后的offset可能是不连续的，比如上图中没有6，当从这些offset消费消息时，将会拿到比这个offset大 的offset对应的消息，实际上会拿到offset为7的消息，并从这个位置开始消费。
这种策略只适合特殊场景，比如消息的key是用户ID，value是用户的资料，通过这种压缩策略，整个消息集里就保存了所有用户最新的资料。

5 高效读写数据

1）Kafka 本身是分布式集群，可以采用分区技术，并行度高
2）读数据采用稀疏索引，可以快速定位要消费的数据
3）顺序写磁盘
Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到 600M/s，而随机写只有 100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。
QQ截图20220705115924.png
4）页缓存 + 零拷贝技术
QQ截图20220705120005.png

参数 描述
log.flush.interval.messages 强制页缓存刷写到磁盘的条数，默认是 long 的最大值，9223372036854775807。一般不建议修改，交给系统自己管理。
log.flush.interval.ms 每隔多久，刷数据到磁盘，默认是 null。一般不建议修改， 交给系统自己管理。