「Kafka」Broker篇

这篇具有很好参考价值的文章主要介绍了「Kafka」Broker篇。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

「Kafka」Broker篇

主要讲解的是在 Kafka 中是怎么存储数据的,以及 Kafka 和 Zookeeper 之间如何进行数据沟通的。

Kafka Broker 总体工作流程

Zookeeper 存储的 Kafka 信息

  • 启动 Zookeeper 客户端:

    [atguigu@hadoop102 zookeeper-3.5.7]$ bin/zkCli.sh
    
  • 通过 ls 命令可以查看 kafka 相关信息:

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

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

Kafka Broker 总体工作流程

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

模拟 Kafka 上下线,Zookeeper 中数据变化:

  1. 查看 /kafka/brokers/ids 路径上的节点:

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

  2. 查看 /kafka/controller 路径上的数据:

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

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

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

  4. 停止 hadoop104 上的 kafka: 「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

  5. 再次查看 /kafka/brokers/ids 路径上的节点

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

  6. 再次查看 /kafka/controller 路径上的数据

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

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

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

  8. 启动 hadoop104 上的 kafka

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

  9. 再次观察 1、2、3 步骤中的内容。

Broker 重要参数

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

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

服役新节点

新节点准备

  1. 关闭 hadoop104,并右键执行克隆操作

  2. 开启 hadoop105,并修改 IP 地址

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

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

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

  4. 重新启动 hadoop104、hadoop105

  5. 修改 haodoop105 中 kafka 的 broker.id 为 3保证唯一

    [atguigu@hadoop105 config]$ vim server.properties
    

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

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

    [atguigu@hadoop105 kafka]$ rm -rf datas/* logs/*
    
  7. 启动 hadoop102、hadoop103、hadoop104 上的 kafka 集群

    [atguigu@hadoop102 ~]$ zk.sh start
    [atguigu@hadoop102 ~]$ kf.sh start
    
  8. 单独启动 hadoop105 中的 kafka

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

我们先来看一下 first 主题的信息:

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

目前 first 主题的信息仍然只存在 broker0、1、2上,但 broker3 并没有帮助分担历史数据,所以我们需要负载均衡的操作。

执行负载均衡操作

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

    [atguigu@hadoop102 kafka]$ vim topics-to-move.json
    
    {
    	"topics": [
    		{"topic": "first"}
    	],
    	"version": 1
    }
    
  2. 生成一个负载均衡的计划

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

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

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

    输入以下内容(刚生成的计划):

    {"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"]}]}
    
  4. 执行副本存储计划:

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

  5. 验证副本存储计划:

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

退役旧节点

执行负载均衡操作

先按照退役一台节点,生成执行计划,然后按照服役时操作流程执行负载均衡

把要退役节点的数据导入到其他节点上。

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

    [atguigu@hadoop102 kafka]$ vim topics-to-move.json
    
    {
    	"topics": [
    		{"topic": "first"}
    	],
        "version": 1
    }
    
  2. 创建执行计划

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

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

    [atguigu@hadoop102 kafka]$ vim increase-replication-factor.json
    
    {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,2,0],"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. 验证副本存储计划

    [atguigu@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
    

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

执行停止命令

在 hadoop105 上执行停止命令即可:

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

Kafka 副本

副本基本信息

  • Kafka 副本作用:提高数据可靠性。

  • Kafka 默认副本 1 个,生产环境一般配置为 2 个,保证数据可靠性;

    • 太多副本会增加磁盘存储空间,增加网络上数据传输,降低效率。
  • Kafka 中副本分为:Leader 和 Follower。

    • Kafka 生产者只会把数据发往 Leader,然后 Follower 找 Leader 进行同步数据。
  • Kafka 分区中的所有副本统称为 AR(Assigned Repllicas)。

A R = I S R + O S R AR = ISR + OSR AR=ISR+OSR

I S R ISR ISR,表示和 Leader 保持同步的 Follower 集合。如果 Follower 长时间未向 Leader 发送通信请求或同步数据,则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms 参数设定,默认 30s。Leader 发生故障之后,就会从 ISR 中选举新的 Leader。

O S R OSR OSR,表示 Follower 与 Leader 副本同步时,延迟过多的副本。

Leader 选举流程

Kafka 集群中有一个 broker 的 Controller 会被选举为 Controller Leader,负责管理集群 broker 的上下线,所有 topic 的分区副本分配 Leader 选举等工作。

Controller 的信息同步工作是依赖于 Zookeeper 的。

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

Leader 选举会按照 AR 的顺序进行选取,就是下图中的 Replicas 顺序:

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

Leader 和 Follower 故障处理细节

Follower 故障处理细节

消费者可见的数据最大 offset 就是 4, H W − 1 HW - 1 HW1

该 Follower 先被踢出 ISR 队列,然后其余的 Leader、Follower继续接受数据。如果该 Follower 恢复了,会读取本地磁盘上次记录的 HW,并裁剪掉 高于 HW 的数据,从 HW 开始向 Leader 进行同步数据。

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

待该 Follower 的 LEO 大于等于该 Partition 的 HW,即 Follower 追上了 Leader,

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

Leader 故障处理细节

broker0 一开始是 Leader,然后挂掉了,选举 broker1 为新的 Leader,然后其余的 Follower 会把各自 log 文件高于 HW 的部分裁剪掉,然后从新的 Leader 同步数据。

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

分区副本分配

如果 kafka 服务器只有 4 个节点,那么设置 kafka 的分区数大于服务器台数,在 kafka 底层如何分配存储副本呢?

创建 16 分区,3 个副本

  1. 创建一个新的 topic,名称为 second

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

  2. 查看分区和副本情况:

    「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

依次错开,让每一个副本负载均衡,均匀分配,也可以保证数据的可靠性。

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

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

在生产环境中,每台服务器的配置和性能不一致,但是Kafka只会根据自己的代码规则创建对应的分区副本,就会导致个别服务器存储压力较大。所有需要手动调整分区副本的存储。

需求:创建一个新的topic,4个分区,两个副本,名称为 three。将该 topic 的所有副本都存储到 broker0 和 broker1 两台服务器上。

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

手动调整分区副本存储的步骤如下:

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

生产经验—Leader Partition 负载平衡

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

真正生产环境建议关闭,或设置 percentage 为 20%、30%,不要频繁的触发自平衡,浪费集群大量性能。

生产经验—增加副本因子

在生产环境当中,由于某个主题的重要等级需要提升,我们考虑增加副本。副本数的增加需要先制定计划,然后根据计划执行。

  1. 创建 topic

    [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 3 --replication-factor 1 --topic four
    
  2. 手动增加副本存储

    1. 创建副本存储计划(所有副本都指定存储在 broker0、broker1、broker2 中)

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

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

文件存储

文件存储机制

Topic 数据的存储机制

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

kafka 中默认数据保存 7 天,通过 .timeindex 文件判断日志保存多久,过期会定时清理对应的数据,详情参考下方的 - 文件清理策略。

思考:Topic 数据到底存储在什么位置?

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件
「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件
「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

index 文件和 log 文件详解

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

说明:日志存储参数配置

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

文件清理策略

Kafka 中默认的日志保存时间为 7 天,可以通过调整如下参数修改保存时间:

  • log.retention.hours,最低优先级,小时,默认 7 天。
  • log.retention.minutes,分钟。
  • log.retention.ms,最高优先级,毫秒。
  • log.retention.check.interval.ms,负责设置检查周期,默认 5 分钟。

那么日志一旦超过了设置的时间,怎么处理呢?

Kafka 中提供的日志清理策略有 deletecompact 两种。

1)delete 日志删除:将过期数据删除
  • log.cleanup.policy = delete 所有数据启用删除策略(默认)

    1. 基于时间:默认打开。以 segment 中所有记录中的最大时间戳作为该文件时间戳。

    2. 基于大小:默认关闭。超过设置的所有日志总大小,删除最早的 segment。

      log.retention.bytes,默认等于 -1,表示无穷大,其实就是关闭掉了。

思考:如果一个 segment 中有一部分数据过期,一部分没有过期,怎么处理?

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

以 segment 中所有记录中的最大时间戳作为该文件时间戳,进行删除。

也就是只要这个 segment 中有数据还未过期,就不进行删除操作。

2)compact 日志压缩

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

高效读写数据

分布式集群

Kafka 本身是分布式集群,可以采用分区技术,并行度高。

稀疏索引

读数据采用稀疏索引,可以快速定位要消费的数据。

顺序写磁盘

Kafka 的 producer 生产数据,要写入到 log 文件中,写的过程是一直追加到文件末端,为顺序写。官网有数据表明,同样的磁盘,顺序写能到 600M/s,而随机写只有 100K/s。这与磁盘的机械机构有关,顺序写之所以快,是因为其省去了大量磁头寻址的时间。

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

页缓存 + 零拷贝技术

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

「Kafka」Broker篇,Kafka,kafka,分布式,java,后端,中间件

笔记整理自b站尚硅谷视频教程:【尚硅谷】Kafka3.x教程(从入门到调优,深入全面)文章来源地址https://www.toymoban.com/news/detail-799922.html

到了这里,关于「Kafka」Broker篇的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 分布式 - 消息队列Kafka:Kafka 消费者的消费位移

    01. Kafka 分区位移 对于Kafka中的分区而言,它的每条消息都有唯一的offset,用来表示消息在分区中对应的位置。偏移量从0开始,每个新消息的偏移量比前一个消息的偏移量大1。 每条消息在分区中的位置信息由一个叫位移(Offset)的数据来表征。分区位移总是从 0 开始,假设一

    2024年02月12日
    浏览(46)
  • 分布式 - 消息队列Kafka:Kafka生产者架构和配置参数

    生产者发送消息流程参考图1: 先从创建一个ProducerRecord对象开始,其中需要包含目标主题和要发送的内容。另外,还可以指定键、分区、时间戳或标头。在发送ProducerRecord对象时,生产者需要先把键和值对象序列化成字节数组,这样才能在网络上传输。 接下来,如果没有显式

    2024年02月13日
    浏览(45)
  • 分布式 - 消息队列Kafka:Kafka消费者的分区分配策略

    Kafka 消费者负载均衡策略? Kafka 消费者分区分配策略? 1. 环境准备 创建主题 test 有5个分区,准备 3 个消费者并进行消费,观察消费分配情况。然后再停止其中一个消费者,再次观察消费分配情况。 ① 创建主题 test,该主题有5个分区,2个副本: ② 创建3个消费者CustomConsu

    2024年02月13日
    浏览(42)
  • 分布式 - 消息队列Kafka:Kafka生产者发送消息的方式

    不管是把Kafka作为消息队列、消息总线还是数据存储平台,总是需要一个可以往Kafka写入数据的生产者、一个可以从Kafka读取数据的消费者,或者一个兼具两种角色的应用程序。 Kafka 生产者是指使用 Apache Kafka 消息系统的应用程序,它们负责将消息发送到 Kafka 集群中的一个或多

    2024年02月13日
    浏览(40)
  • 分布式 - 消息队列Kafka:Kafka消费者分区再均衡(Rebalance)

    01. Kafka 消费者分区再均衡是什么? 消费者群组里的消费者共享主题分区的所有权。当一个新消费者加入群组时,它将开始读取一部分原本由其他消费者读取的消息。当一个消费者被关闭或发生崩溃时,它将离开群组,原本由它读取的分区将由群组里的其他消费者读取。 分区

    2024年02月12日
    浏览(37)
  • 分布式 - 消息队列Kafka:Kafka 消费者消息消费与参数配置

    01. 创建消费者 在读取消息之前,需要先创建一个KafkaConsumer对象。创建KafkaConsumer对象与创建KafkaProducer对象非常相似——把想要传给消费者的属性放在Properties对象里。 为简单起见,这里只提供4个必要的属性:bootstrap.servers、key.deserializer 和 value.deserializer。 ① bootstrap.servers 指

    2024年02月12日
    浏览(42)
  • 分布式 - 消息队列Kafka:Kafka 消费者消费位移的提交方式

    最简单的提交方式是让消费者自动提交偏移量,自动提交 offset 的相关参数: enable.auto.commit:是否开启自动提交 offset 功能,默认为 true; auto.commit.interval.ms:自动提交 offset 的时间间隔,默认为5秒; 如果 enable.auto.commit 被设置为true,那么每过5秒,消费者就会自动提交 poll() 返

    2024年02月12日
    浏览(43)
  • 分布式 - 消息队列Kafka:Kafka消费者和消费者组

    1. Kafka 消费者是什么? 消费者负责订阅Kafka中的主题,并且从订阅的主题上拉取消息。与其他一些消息中间件不同的是:在Kafka的消费理念中还有一层消费组的概念,每个消费者都有一个对应的消费组。当消息发布到主题后,只会被投递给订阅它的每个消费组中的一个消费者

    2024年02月13日
    浏览(41)
  • 分布式 - 消息队列Kafka:Kafka生产者发送消息的分区策略

    01. Kafka 分区的作用 分区的作用就是提供负载均衡的能力,或者说对数据进行分区的主要原因,就是为了实现系统的高伸缩性。不同的分区能够被放置到不同节点的机器上,而数据的读写操作也都是针对分区这个粒度而进行的,这样每个节点的机器都能独立地执行各自分区的

    2024年02月13日
    浏览(50)
  • 分布式应用之Zookeeper和Kafka

    1.定义 2.特点 3.数据结构 4.选举机制 第一次选举 非第一次选举 5.部署 1.概念 中间件是一种独立的系统软件或服务程序,分布式应用软件借助这种软件在不同的技术之间共享资源。 2.消息队列型 3.Web应用型(代理服务器) 1.为什么需要MQ 2.消息队列作用 3.消息队列模式 ①点对

    2024年02月15日
    浏览(45)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包