【kafka】——Broker

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

Kfaka Broker

1. Zookeeper 存储的kafka信息

【kafka】——Broker

<1> /kafka/brokers/dis

存储broker的id,记录有哪些服务器
【kafka】——Broker

<2> /kafka/brokers/topics

存储topic 相关信息
【kafka】——Broker

【kafka】——Broker

【kafka】——Broker

【kafka】——Broker

<3> /kafka/consumers

Kafka 0.9 版本之前 用于保存offset信息
Kafka 0.9 版本之后offser存储在Kafka主题中

<4> /kafka/controller

辅助选举Leader
【kafka】——Broker

2. Broker工作流程

【kafka】——Broker
<1> Broker 启动后在Zookeeper中注册
<2> 每个节点中的Contoller 抢先在Zookeeper中注册,谁抢先注册,谁负责Leader选举
<3> 由选举出来的Controller 监听brokers节点的变化
<4> 选举规则:在ISR中存活为前提,按照AR(Kafka分区中的所有副本统称)中排在前面的优先,例如:AR[1,0,2],ISR[0,1,2],那么Leader就会按照1,0,2 的顺序轮询
<5> Controller将节点信息上传到ZK
<6> 其他Controller从zk同步相关信息
<7> 生产者发送数据到Leader ,Follower 主动从Leader 拉取信息。
<8> 底层通过Segment存储,一个Segment默认大小1G。.log文件存储信息。.index文件存储索引,加快检索速度。
<9> 如果Ledaer 宕机。Controller监听到brokers节点发生变化,从ZK拉取Leader 和ISR信息,然后重新选举Leader(ISR中存活为前提。按照AR中排在前面的有限)。更新Leader 和ISR

3. 模拟Kafka上下线

<1> 原始状态

  1. 查看/kafka/brokers/ids 路径上的节点。
    【kafka】——Broker

  2. 查看/kafka/controller 路径上的数据。
    【kafka】——Broker

  3. 查看/kafka/brokers/topics/test_topic/partitions/0/state 路径上的数据
    【kafka】——Broker

<2> 停止 Follower节点后状态

   【kafka】——Broker

  1. 再次查看/kafka/brokers/ids 路径上的节点
    【kafka】——Broker

  2. 查看/kafka/controller 路径上的数据。
    【kafka】——Broker

  3. 查看/kafka/brokers/topics/test_topic/partitions/0/state 路径上的数据
    【kafka】——Broker

<3> 停止 Leader节点后状态

【kafka】——Broker

  1. 再次查看/kafka/brokers/ids 路径上的节点
    【kafka】——Broker

  2. 查看/kafka/controller 路径上的数据。
    【kafka】——Broker

  3. 查看/kafka/brokers/topics/test_topic/partitions/0/state 路径上的数据
    【kafka】——Broker

4. 新增和减少节点

<1> 新节点准备

1. 修改IP 地址和主机名

【kafka】——Broker
【kafka】——Broker

2. 安装JDK和Kakfa

【kafka】——Broker

kafka 需要修改配置文件,brokerid 需要全局唯一
如果是从其他服务器克隆,则需要删除datas目录我logs目录下的内容。

3. 启动新增节点上的Kafka
bin/kafka-server-start.sh - daemon ./config/server.properties

<2> 新增节点

   如果只是作为新topic 使用,则不需要以下操作。如果想要将旧topic 的数据迁移到新增节点上,因为已经创建的topic主题的分区和副本只在旧的节点上,那么需要以下操作。可见 test_topic 的分区和副本都在 0,1,2 三个broker中。
   通过命令,可见 test_topic 的分区和副本都在 0,1,2 三个broker中。
【kafka】——Broker

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

在Kafka根目录下创建文件 topics-to-move.json

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

test_topic 为测试节点,如果是多个topic ,只需要在json串中并列添加。“version”: 1是固定写法

2. 生成一个负载均衡的计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate

3 为新增的节点的brokerID
【kafka】——Broker
Current partition replica assignment(当前的分区和副本规划)

{"version":1,"partitions":[{"topic":"test_topic","partition":0,"replicas":[0,2,1],"log_dirs":["any","any","any"]},{"topic":"test_topic","partition":1,"replicas":[1,2,0],"log_dirs":["any","any"test_topic","partition":2,"replicas":[2,0,1],"log_dirs":["any","any","any"]}]}

Proposed partition reassignment configuration(新提出的分区和副本计划)

{"version":1,"partitions":[{"topic":"test_topic","partition":0,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"topic":"test_topic","partition":1,"replicas":[0,1,2],"log_dirs":["any","any"test_topic","partition":2,"replicas":[1,2,3],"log_dirs":["any","any","any"]}]}

可见在新提出的分区和副本计划中添加了brokerid=3的新增节点。

3. 创建副本存储计划

在Kafka根目录下创建文件vim increase-replication-factor.json,将新提出的分区和副本计划写入文件。

{
    "version": 1,
    "partitions": [
        {
            "topic": "test_topic",
            "partition": 0,
            "replicas": [
                3,
                0,
                1
            ],
            "log_dirs": [
                "any",
                "any",
                "any"
            ]
        },
        {
            "topic": "test_topic",
            "partition": 1,
            "replicas": [
                0,
                1,
                2
            ],
            "log_dirs": [
                "any",
                "any",
                "any"
            ]
        },
        {
            "topic": "test_topic",
            "partition": 2,
            "replicas": [
                1,
                2,
                3
            ],
            "log_dirs": [
                "any",
                "any",
                "any"
            ]
        }
    ]
}

4. 执行副本存储计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --reassignment-json-file increase-replication-factor.json --execute

【kafka】——Broker

5. 验证副本存储计划

<1> 方式一

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --reassignment-json-file increase-replication-factor.json --verify

【kafka】——Broker

<2> 方式二

bin/kafka-topics.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --topic test_topic --describe

【kafka】——Broker

<3> 减少节点

当减少一个kafka集群节点的时候,不能简单粗暴的将节点停止,因为有的topic的分区和副本在节点上,如果直接停止会造成数据丢失,需要进行以下操作。

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

在Kafka根目录下创建文件 topics-to-move.json

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

test_topic 为测试节点,如果是多个topic ,只需要在json串中并列添加。“version”: 1是固定写法

2. 生成一个负载均衡的计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate

broker-list只保留要留存的节点
【kafka】——Broker

Current partition replica assignment(当前的分区和副本规划)

{"version":1,"partitions":[{"topic":"test_topic","partition":0,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"topic":"test_topic","partition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"test_topic","partition":2,"replicas":[1,2,3],"log_dirs":["any","any","any"]}]}

Proposed partition reassignment configuration(新提出的分区和副本计划)

{"version":1,"partitions":[{"topic":"test_topic","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"test_topic","partition":1,"replicas":[1,2,0],"log_dirs":["any","any","any"]},{"topic":"test_topic","partition":2,"replicas":[2,0,1],"log_dirs":["any","any","any"]}]}

可见在新提出的分区和副本计划中 只保留要留存的节点。

3. 创建副本存储计划

在Kafka根目录下创建文件vim increase-replication-factor.json,将新提出的分区和副本计划写入文件。

{
    "version": 1,
    "partitions": [
        {
            "topic": "test_topic",
            "partition": 0,
            "replicas": [
                0,
                1,
                2
            ],
            "log_dirs": [
                "any",
                "any",
                "any"
            ]
        },
        {
            "topic": "test_topic",
            "partition": 1,
            "replicas": [
                1,
                2,
                0
            ],
            "log_dirs": [
                "any",
                "any",
                "any"
            ]
        },
        {
            "topic": "test_topic",
            "partition": 2,
            "replicas": [
                2,
                0,
                1
            ],
            "log_dirs": [
                "any",
                "any",
                "any"
            ]
        }
    ]
}
4. 执行副本存储计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --reassignment-json-file increase-replication-factor.json --execute

【kafka】——Broker

5. 验证副本存储计划

<1> 方式一

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --reassignment-json-file increase-replication-factor.json --verify

【kafka】——Broker

<2> 方式二

bin/kafka-topics.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --topic test_topic --describe

【kafka】——Broker
【kafka】——Broker

6. 停止节点
bin/kafka-server-stop.sh

5. Kafka副本

<1> 基本信息

  1. 作用:提高数据可靠性
  2. 默认一个副本,生产环境一般配置2个。保证数据可靠性;太多副本会增加磁盘空间,增加网络上数据传输,降低效率。
  3. 分为Leader 和Follower。kafka生产者只会把数据发送给Leader。饭后Follower找Leader进行同步数据。
  4. ISR:标识和Leader保持同步的Follower集合,如果长时间未向Leader发送通讯请求或同步数据,则Follower将被提出ISR。该时间阈值由replica.time.max.ms 参数设定,默认30s。Leader发生故障之后,就会从ISR中选举新的Leader。
  5. SR:表示Follower与Leader副本同步时,延迟过多
  6. AR:Kafka分区中所有副本的统称 O的副本。

<2> Leader选举流程

Kafka集群中有个一broker的Controller 会被选举为Controller Leader。负责管理集群broker的上下线,所有topic的分区副本分配和Leader选举等工作。Controller的信息同步工作是依赖于Zookeeper的。
【kafka】——Broker

  1. Broker 启动后在Zookeeper中注册
  2. 每个节点中的Contoller 抢先在Zookeeper中注册,谁抢先注册,谁负责Leader选举
  3. 由选举出来的Controller 监听brokers节点的变化
  4. 选举规则:在ISR中存活为前提,按照AR(Kafka分区中的所有副本统称)中排在前面的优先,例如:AR[1,0,2],ISR[0,1,2],那么Leader就会按照1,0,2 的顺序轮询
  5. 如果Ledaer 宕机。Controller监听到brokers节点发生变化,从ZK拉取Leader 和ISR信息,然后重新选举Leader(ISR中存活为前提,按照AR中排在前面的有限)。

<3> Leader 和Follower 故障处理细节

1. LEO(Log End Offset)

每个副本的最后一个offset ,LEO其实就是最新的offset +1 。

2. HW(High Watermark)

所有副本中最小的LEO。

3. Follower 故障

【kafka】——Broker
<1> Follower 发生故障后,会被临时踢出ISR。
<2> 这个期间Leader和Follower继续接受数据。
<3> 待该Follower恢复后,Follower 会读取本地磁盘记录的上次的HW。并将log文件高于HW的部分截取掉,从HW开始向Leader进行同步。
<4> 待该Follower的LEO大于等于该Partition的HW。即Follower追上Leader之后,就可以重新加入ISR了。

4. Leader故障

<1> Leader发生故障后,会被临时踢出ISR。
<2> 从ISR 中选出一个新的Leader。
<3> 为保证多个副本之间的数据一致性,其余的Follower会先将各自的log 文件高于HW的部分接截掉,然后从新的Leader同步数据。(只能保证副本之间的数据一致的,并不能保证数据不丢失或者不重复)

<4> 分区副本分配

1. 默认分配

如果kafka分区数大于服务器台数默认情况下, kafka分区分配时候,底层实现是尽量均匀分布的。

2. 手动分配

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

3. 实例

创建一个新的主题,4个分区,每个分区2个副本。默认情况下分区和副本会分布在四个节点上,再手动改为broker0 和broker1上。

<1> 创建主题
bin/kafka-topics.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092,hadoop105:9092 --topic test_assign_topic --create --partitions 4 --replication-factor 2
<2> 查看分区和副本分布情况
bin/kafka-topics.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092,hadoop105:9092  --topic test_assign_topic --describe

【kafka】——Broker

<3> 创建副本存储计划

在Kafka根目录下创建文件vim increase-replication-factor.json,将新提出的分区和副本计划写入文件。

 {
    "version": 1,
    "partitions": [
        {
            "topic": "test_assign_topic",
            "partition": 0,
            "replicas": [
                1,
                0
            ],
            "log_dirs": [
                "any",
                "any"
            ]
        },
        {
            "topic": "test_assign_topic",
            "partition": 1,
            "replicas": [
                0,
                1
            ],
            "log_dirs": [
                "any",
                "any"
            ]
        },
        {
            "topic": "test_assign_topic",
            "partition": 2,
            "replicas": [
                1,
                0
            ],
            "log_dirs": [
                "any",
                "any"
            ]
        },
        {
            "topic": "test_assign_topic",
            "partition": 3,
            "replicas": [
                0,
                1
            ],
            "log_dirs": [
                "any",
                "any"
            ]
        }
    ]
}

<4> 执行副本存储计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092,hadoop105:9092 --reassignment-json-file increase-replication-factor.json --execute

【kafka】——Broker

<5> 验证副本存储计划

<1> 方式一

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092,hadoop105:9092 --reassignment-json-file increase-replication-factor.json --verify

【kafka】——Broker
<2> 方式二

bin/kafka-topics.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092,hadoop105:9092 --topic test_topic --describe

【kafka】——Broker

<5> Leader Partition 负载平衡

  正常情况下,Kafka本身会自动把Leader Partition均匀分散在各个机器上,来保证每台机器的读写吞吐量都是均匀的。但是如果某些broker宕机,会导致Leader Partition过于集中在其他少部分几台broker上,这会导致少数几台broker的读写请求压力过高,其他宕机的broker重启之后都是follower partition,读写请求很低,造成集群负载不均衡。

1. 自动平衡

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

<6> 增加副本

1. 创建主题
bin/kafka-topics.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092, hadoop105:9092 --topic test_add_replication_topic --create --partitions 4 --replication-factor 1
2. 查看分区和副本分布情况
bin/kafka-topics.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092,hadoop105:9092  --topic test_add_replication_topic --describe

【kafka】——Broker

3. 创建副本存储计划

在Kafka根目录下创建文件vim increase-replication-factor.json,将新提出的分区和副本计划写入文件。

{
    "version": 1,
    "partitions": [
        {
            "topic": "test_add_replication_topic",
            "partition": 0,
            "replicas": [
                1,
                0
            ],
            "log_dirs": [
                "any",
                "any"
            ]
        },
        {
            "topic": "test_add_replication_topic",
            "partition": 1,
            "replicas": [
                0,
                1
            ],
            "log_dirs": [
                "any",
                "any"
            ]
        },
        {
            "topic": "test_add_replication_topic",
            "partition": 2,
            "replicas": [
                2,
                3
            ],
            "log_dirs": [
                "any",
                "any"
            ]
        },
        {
            "topic": "test_add_replication_topic",
            "partition": 3,
            "replicas": [
                3,
                2
            ],
            "log_dirs": [
                "any",
                "any"
            ]
        }
    ]
}

4. 执行副本存储计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092,hadoop105:9092 --reassignment-json-file increase-replication-factor.json --execute

【kafka】——Broker

5. 验证副本存储计划

<1> 方式一

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092,hadoop105:9092 --reassignment-json-file increase-replication-factor.json --verify

【kafka】——Broker

<2> 方式二

bin/kafka-topics.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092,hadoop105:9092 --topic test_add_replication_topic --describe

【kafka】——Broker

6. 文件存储机制

<1> 概述

【kafka】——Broker

  1. Topic是逻辑上的概念,而partition是物理上的概念。
  2. 每个partition对应一个log文件(log文件也是虚拟概念),该log文件中存储的就是Producer生产的数据。Producer生产的数据会被不断追加到该log文件末端。
  3. 为防止log文件过大导致数据定位效率低下,Kafka采取了分片和索引机制,将每个partition分为多个segment。log.segment.bytes 指定segment的大小,默认值 1G。
  4. 每个segment包括:.index文件、.log文件和.timeindex等文件。位于同一目录下。
  5. 该文件夹的命名规则为:topic名称+分区序号,例如test_topic-0。
  6. Kafka 中的消息默认保存7天。

<2> 查看文件内容

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

kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.index
kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.logs

<3> index 文件和 log 文件详解

  1. index为稀疏索引,大约每往log 文件中写入4kb数据,会往index文件中写入一条索引,参数log.index.interval.bytes 默认4kb
  2. index文件中保存的offset为相对offset,这样能保证offset的值所占空间不会过大,因此能将offset 的值控制在固定大小。

<4> 查找过程

【kafka】——Broker

  1. 根据目标offset定位segment文件。
  2. 找到小于或等于目标offset的最大offset对那个的索引项。
  3. 定位到log文件。
  4. 向下遍历找到目标record。

7. 文件清理策略

<1> 概述

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

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

<2> 日志清理策略

1. delete

<1> log.cleanup.policy = delete 所有数据启用删除策略
<2> 基于时间:默认打开。以 segment 中所有记录中的最大时间戳作为该文件时间戳。
<3> 基于大小(log.retention.bytes):默认关闭。超过设置的所有日志总大小,删除最早的 segment。
<4> log.retention.bytes,默认等于-1,表示无穷大。

2. compact

【kafka】——Broker
<1> log.cleanup.policy = compact 所有数据启用压缩策略
<2> compact日志压缩:对于相同key的不同value值,只保留最后一个版本。
<3> 压缩后的offset可能是不连续的,比如上图中没有6,当从这些offset消费消息时,将会拿到比这个offset大 的offset对应的消息,实际上会拿到offset为7的消息,并从这个位置开始消费。
<4> 这种策略只适合特殊场景,比如消息的key是用户ID,value是用户的资料,通过这种压缩策略,整个消息 集里就保存了所有用户最新的资料。

8. 高效读写数据

<1> kafka 本省是分布式集群,可以采用分区奇数,并行度高。
<2> 读数据采用稀疏索引,可以快速定位到要消费的数据。
<3> 顺序写磁盘
  Kafka 的 producer 生产数据,要写入到 log文件中,写的过程是一直追加到文件末端,为顺序写。官网有数据表明,同样的磁盘,顺序写能到600M/s,而随机写只有100K/s。这与磁盘的机械机构有关,顺序写之所以快,是因为其省去了大量磁头寻址的时间。
【kafka】——Broker
<4> 页缓存 + 零拷贝技术
【kafka】——Broker
  零拷贝:Kafka的数据加工处理操作交由Kafka生产者和Kafka消费者处理。Kafka Broker应用层不关心存储的数据,所以就不用 走应用层,传输效率高。
  PageCache页缓存:Kafka重度依赖底层操作系统提供的PageCache功能。当上层有写操作时,操作系统只是将数据写入 PageCache。当读操作发生时,先从PageCache中查找,如果找不到,再去磁盘中读取。实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用。
【kafka】——Broker文章来源地址https://www.toymoban.com/news/detail-493976.html

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

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

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

相关文章

  • Kafka 实战 - Kafka Broker工作流程

    Apache Kafka Broker 在 Kafka 集群中扮演着核心角色,负责接收、存储、复制及分发消息。以下是 Kafka Broker 的工作流程概览: 1. 启动与初始化 加载配置 :Kafka Broker 从 server.properties 文件加载配置参数,包括 Broker ID、监听地址、日志目录、ZooKeeper 连接信息等。 注册到 ZooKeeper :

    2024年04月15日
    浏览(41)
  • 「Kafka」Broker篇

    主要讲解的是在 Kafka 中是怎么存储数据的,以及 Kafka 和 Zookeeper 之间如何进行数据沟通的。 Zookeeper 存储的 Kafka 信息 启动 Zookeeper 客户端: 通过 ls 命令可以查看 kafka 相关信息: Kafka Broker 总体工作流程 模拟 Kafka 上下线,Zookeeper 中数据变化: 查看 /kafka/brokers/ids 路径上的节

    2024年01月18日
    浏览(35)
  • Kafka - Broker 详解

    目录 零、前置 一、Kafka Broker 工作流程 1.Zookeeper 存储的 Kafka 信息 2.Kafka Broker 总体工作流程 模拟 Kafka 上下线,Zookeeper 中数据变化 3.Broker 重要参数 二、生产经验 节点服役和退役 1.服役新节点 新节点准备 执行负载均衡操作 生成负载均衡的计划 执行副本存储计划 验证副本存

    2024年01月24日
    浏览(35)
  • 深入Kafka broker

    颗粒度, PRODUCE和FETCH中支持topic,partion等层级的颗粒度; 测试友好, 基于session_id和epoch确定一条拉取链路的fetch session; 全量增量结合, FetchRequest中的全量拉取和增量拉取; 基本结构: header+body。 常见header: api_key, api_version, corelation_id, client_id。与网络协议类似, Kafka本身的协议也是分

    2024年01月22日
    浏览(72)
  • 【kafka】——Broker

    1 /kafka/brokers/dis 存储broker的id,记录有哪些服务器 2 /kafka/brokers/topics 存储topic 相关信息 3 /kafka/consumers Kafka 0.9 版本之前 用于保存offset信息 Kafka 0.9 版本之后offser存储在Kafka主题中 4 /kafka/controller 辅助选举Leader 1 Broker 启动后在Zookeeper中注册 2 每个节点中的Contoller 抢先在Zookeepe

    2024年02月09日
    浏览(34)
  • 四、Kafka Broker

    4.1.1 Zookeeper 存储的 Kafka 信息 4.1.2 Kafka Broker 总体工作流程 自己的理解:其实就是将kafka的分区,负载到集群中的各个节点上。 1、服役新节点 2、退役旧节点 1、副本的作用 2、Leader的选举流程 选举规则:在isr中存活为前提,按照AR中排在前面的优先。例如ar[1,0,2], isr [1,0,

    2024年02月11日
    浏览(31)
  • Kafka-Broker工作流程

     kafka集群在启动时,会将每个broker节点注册到zookeeper中,每个broker节点都有一个controller,哪个controller先在zookeeper中注册,哪个controller就负责监听brokers节点变化,当有分区的leader挂掉时,controller会监听到节点变化,然后去zookeeper中获取isr,选举新的leader,选举的规则是:在

    2024年02月14日
    浏览(43)
  • KafKa 3.x(二、Broker,消费者)

    4.1.1 Zoopkeeper存储的Kafka信息 启动Zookeeper客户端 通过ls命令查看kafka相关信息 在Zookeeper的服务端存储的Kafka相关信息 /kafka/brokers/ids [0,1,2] 记录那些服务器 /kafka/brokers/topics/first/partitions/0/state {“leader”:1,“isr”:[1,0,2]} 记录谁是leader,有哪些服务器可用 /kafka/controller {“brokerid”

    2024年02月12日
    浏览(35)
  • Kafka学习--3、Kafka Broker、节点服役和退役、Kafka 副本、Leader 选举流程、故障处理

    1.1 Kafka Broker工作流程 1.1.1 Zookeeper储存的Kafka信息 (1)启动Zookeeper集群、再启动Kafka集群,然后启动Zookeeper客户端 (2)通过ls命令可以查看kafka相关信息。 1.1.2 Kafka Broker总体工作流程 1、模拟Kafka上下线,Zookeeper中数据变化 (1)查看/kafka/brokers/ids 路径上的节点。 (2)查看

    2024年02月10日
    浏览(43)
  • Kafka3.0.0版本——Broker( 退役旧节点)示例

    三台服务器 原始服务器名称 原始服务器ip centos7虚拟机1 192.168.136.27 centos7虚拟机2 192.168.136.28 centos7虚拟机3 192.168.136.29 centos7虚拟机4 192.168.136.30 分别启动4台zookeeper 再分别启动4台kafka brokers中的ids=0、1、2、3都存在,说明4台kafka启动正常,如下图: 查看kafka中的创建过的名称为

    2024年02月13日
    浏览(35)

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

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

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

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

二维码1

领取红包

二维码2

领红包