seo怎么做整站排名,简单网页制作素材,网站后台管理系统制作软件,网络优化报告Kafka#xff1a;分布式消息系统的核心原理与安装部署-CSDN博客
自定义 Kafka 脚本 kf-use.sh 的解析与功能与应用示例-CSDN博客
Kafka 生产者全面解析#xff1a;从基础原理到高级实践-CSDN博客
Kafka 生产者优化与数据处理经验-CSDN博客
Kafka 工作流程解析#xff1a…
Kafka分布式消息系统的核心原理与安装部署-CSDN博客
自定义 Kafka 脚本 kf-use.sh 的解析与功能与应用示例-CSDN博客
Kafka 生产者全面解析从基础原理到高级实践-CSDN博客
Kafka 生产者优化与数据处理经验-CSDN博客
Kafka 工作流程解析从 Broker 工作原理、节点的服役、退役、副本的生成到数据存储与读写优化-CSDN博客
Kafka 消费者全面解析原理、消费者 API 与Offset 位移-CSDN博客
Kafka 分区分配及再平衡策略深度解析与消费者事务和数据积压的简单介绍-CSDN博客
Kafka 数据倾斜原因、影响与解决方案-CSDN博客
Kafka 核心要点解析_kafka mirrok-CSDN博客
Kafka 核心问题深度解析全面理解分布式消息队列的关键要点_kafka队列日志-CSDN博客 目录
一、Kafka Broker 工作流程
一Zookeeper 存储的 Kafka 信息
二Kafka Broker 总体工作流程
注册与选举
数据处理
二、生产经验 —— 节点服役和退役
一服役新节点
新节点准备
执行负载均衡操作
二退役旧节点
执行负载均衡操作
执行停止命令
三、Kafka 副本
一副本基本信息
二Leader 选举流程
三Leader 和 Follower 故障处理细节
四分区副本分配
五生产经验 —— 手动调整分区副本的存储
六生产经验 ——Leader Partition 负载平衡
七生产经验 —— 增加副本因子
四、文件存储
一文件存储机制
Topic 数据存储
数据位置查找
二文件清理策略
delete 日志删除
compact 日志压缩
五、高效读写数据
分区技术与并行度
稀疏索引与快速定位
顺序写磁盘
页缓存 零拷贝技术
六、总结 在大数据处理领域Kafka 作为一款高性能的分布式消息队列系统扮演着至关重要的角色。它能够高效地处理大规模的实时数据为众多大数据应用提供了可靠的数据传输和存储基础。本文将深入探讨 Kafka 的核心组件 ——Kafka Broker包括其工作流程、副本机制、文件存储以及高效读写数据的原理并分享一些生产环境中的实用经验。 一、Kafka Broker 工作流程 一Zookeeper 存储的 Kafka 信息 Kafka 与 Zookeeper 紧密协作Zookeeper 存储了许多关键的 Kafka 信息。启动 Zookeeper 客户端后通过ls /kafka命令可以查看相关信息。其中consumers节点在老版本0.9 版本之前存放消费者的偏移量而在新版本中偏移量直接存储在集群中。我们还可以借助工具如 “漂亮的 zoo” 通过图形化界面更直观地查看 Zookeeper 中的信息。
通过网盘分享的文件prettyZoo-win.msi 二Kafka Broker 总体工作流程 注册与选举 每一个 Broker 上线时会在 Zookeeper 中进行注册。每个 Broker 中都有一个 Controller它们会争先抢占 Zookeeper 中controller节点的注册权率先抢到的 Broker 中的 Controller 将在选举等事务中拥有决策权。 选举规则以在 ISR与 Leader 保持同步的 Follower 集合中存活为前提按照 AR分区中的所有副本集合中排在前面的优先。例如对于ar[1,0,2]isr[1,0,2]Leader 会按照102的顺序进行轮询选举。选举出来的新节点会注册到 Zookeeper 中并将信息记录下来其他 Controller 则同步 Zookeeper 中的信息假定broker中的leader挂掉了会进行重新的选举。 数据处理 客户端发送消息给 LeaderLeader 负责记录数据并落盘形成 Log。Log 底层采用 Segment 机制每一个 Segment 大小为 1G为了实现数据的快速查找在 1G 内的数据又分成了log和index两个文件。 二、生产经验 —— 节点服役和退役 一服役新节点 新节点准备 1关闭 bigdata03进行一个快照并右键执行克隆操作。
2开启 bigdata04并修改 IP 地址。
vi /etc/sysconfig/network-scripts/ifcfg-ens33修改完记得重启网卡
systemctl restart network
3在 bigdata04 上修改主机名称为 bigdata04。
hostname bigdata04 # 临时修改
[rootbigdata04 ~]# vim /etc/hostname
bigdata04
还要记得修改 /etc/hosts文件并进行同步
修改bigdata01的hosts 文件修改完之后记得同步一下192.168.52.11 bigdata01
192.168.52.12 bigdata03
192.168.52.13 bigdata02
192.168.52.14 bigdata04xsync.sh /etc/hosts
scp -r /etc/hosts rootbigdata04:/etc/
4重新启动 bigdata03、bigdata04。
5修改 bigdata04 中 kafka 的 broker.id 为 3。
进入bigdata04的kafka中修改里面的配置文件 config/server.properties
6删除 bigdata04 中 kafka 下的 datas 和 logs。
rm -rf datas/* logs/*
7启动 bigdata01、bigdata02、bigdata03 上的 kafka 集群。
先启动zk集群
xcall.sh zkServer.sh stop
xcall.sh zkServer.sh start
启动kafka集群只能启动三台
kf.sh start
8单独启动 bigdata04 中的 kafka。
bin/kafka-server-start.sh -daemon ./config/server.properties
查看kafka集群first主题的详情 bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --topic first --describe 发现副本数并没有增加。 由于我之前创建first这个主题的时候只有一个副本不是三个副本所以呢演示效果不佳。 kafka-topics.sh --bootstrap-server bigdata01:9092 --topic third --create --partitions 3 --replication-factor 3 执行负载均衡操作 1在kafka下创建一个要均衡的主题
创建一个文件vi topics-to-move.json
写上如下代码如果多个topic 可以使用,分隔
{topics: [{topic: third}],version: 1
}
2生成一个负载均衡的计划
在创建的时候记得启动bigdata04节点否则计划中还是没有bigdata04
bin/kafka-reassign-partitions.sh --bootstrap-server bigdata01:9092 --topics-to-move-json-file topics-to-move.json --broker-list 0,1,2,3 --generate 未来的分区策略拷贝一份
{version:1,partitions:[{topic:abc,partition:0,replicas:[2,0,1],log_dirs:[any,any,any]},{topic:abc,partition:1,replicas:[3,1,2],log_dirs:[any,any,any]},{topic:abc,partition:2,replicas:[0,2,3],log_dirs:[any,any,any]}]}
3创建副本存储计划所有副本存储在 broker0、broker1、broker2、broker3 中。
vi increase-replication-factor.json
{version:1,partitions:[{topic:first,partition:0,replicas:[3,2,0],log_dirs:[any,any,any]},{topic:first,partition:1,replicas:[0,3,1],log_dirs:[any,any,any]},{topic:first,partition:2,replicas:[1,0,2],log_dirs:[any,any,any]}]}
以上这个内容来自于第二步的执行计划。
4执行副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server bigdata01:9092 --reassignment-json-file increase-replication-factor.json --execute
5验证副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server bigdata01:9092 --reassignment-json-file increase-replication-factor.json --verify
如果不相信添加成功可以查看first节点的详情 二退役旧节点 执行负载均衡操作 先按照退役一台节点生成执行计划然后按照服役时操作流程执行负载均衡。
1创建一个要均衡的主题
kafka下添加文件vim topics-to-move.json
添加如下内容
{topics: [{topic: abc}],version: 1
}
2创建执行计划。
bin/kafka-reassign-partitions.sh --bootstrap-server bigdata01:9092 --topics-to-move-json-file topics-to-move.json --broker-list 0,1,2 --generate
3创建副本存储计划所有副本存储在 broker0、broker1、broker2 中。
添加文件 vi increase-replication-factor.json
添加如下代码
{version:1,partitions:[{topic:first,partition:0,replicas:[0,2,1],log_dirs:[any,any,any]},{topic:first,partition:1,replicas:[1,0,2],log_dirs:[any,any,any]},{topic:first,partition:2,replicas:[2,1,0],log_dirs:[any,any,any]}]}
4执行副本存储计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop11:9092 --reassignment-json-file increase-replication-factor.json --execute
5验证副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop11:9092 --reassignment-json-file increase-replication-factor.json --verify 执行停止命令 在要退役的节点如bigdata04上执行bin/kafka-server-stop.sh命令停止 Kafka 服务。 三、Kafka 副本 一副本基本信息 1Kafka 副本作用提高数据可靠性。
2Kafka 默认副本 1 个生产环境一般配置为 2 个保证数据可靠性太多副本会增加磁盘存储空间增加网络上数据传输降低效率。
3Kafka 中副本分为Leader 和 Follower。Kafka 生产者只会把数据发往 Leader 然后 Follower 找 Leader 进行同步数据。
4Kafka 分区中的所有副本包含Leader统称为 ARAssigned Repllicas。
AR ISR OSR ISR表示和 Leader 保持同步的 Follower 集合。如果 Follower 长时间未向 Leader 发送通信请求或同步数据则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms 参数设定默认 30s。Leader 发生故障之后就会从 ISR 中选举新的 Leader。 OSR表示 Follower 与 Leader 副本同步时延迟过多的副本. 二Leader 选举流程 Kafka 集群中有一个 Broker 的 Controller 会被选举为 Controller Leader负责管理集群 Broker 的上下线、所有 topic 的分区副本分配和 Leader 选举等工作其信息同步依赖于 Zookeeper。例如创建一个新的 topic如bigdata2401设置 4 个分区和 4 个副本通过停止某些节点上的 Kafka 进程可以观察到 Leader 的选举是按照 AR 进行的而不是 ISR。
1创建一个新的 topic4 个分区4 个副本
bin/kafka-topics.sh --bootstrap-server hadoop11:9092 --create --topic bigdata2401 --partitions 4 --replication-factor 4
2查看 Leader 分布情况 3停止掉 hadoop13 的 kafka 进程并查看 Leader 分区情况
bin/kafka-server-stop.shbin/kafka-topics.sh --bootstrap-server bigdata01:9092 --describe
--topic bigdata2305 4停止掉 hadoop14 的 kafka 进程并查看 Leader 分区情况 通过以上演示大家可以发现选举是按照AR(跟Replicas一样)进行的而不是ISR 三Leader 和 Follower 故障处理细节 LEO演示-- 每一个副本最后的偏移量offset 1 HW(高水位线 High Water) 演示所有副本中最小的LEO 由于数据同步的时候先进入Leader,随后同步给Follower假如Follower挂掉了Leader和其他的Follower 继续往前存储数据挂掉的节点从ISR集合中剔除此时挂掉的Follower又重启了它会先从上一次挂掉的节点的HW开始同步数据直到追上最后一个Follower为止,此时会重新回归ISR。 四分区副本分配 如果 Kafka 服务器只有有限的节点数在创建 topic 时分区数和副本数的设置需要合理。例如创建 16 分区、3 个副本的second主题Kafka 在初始化时会按照特定算法选举第一 Leader保障 Leader 不在一个 Broker 里面。
1创建 16 分区3 个副本
1创建一个新的 topic名称为 second。
bin/kafka-topics.sh --bootstrap-server hadoop11:9092 --create --partitions 16 --replication-factor 3 --topic second bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --create --topic bigdata230801 --partitions 3 --replication-factor 4 假如你有3个broker ,却创建4个副本报错 Error while executing topic command : Replication factor: 4 larger than available brokers: 3. [2023-09-13 18:43:47,458] ERROR org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 4 larger than available brokers: 3 bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --create --topic bigdata23 --partitions 4 --replication-factor 2 假如你有3个broker ,却创建4个分区是可以的。 以上错误的意思是目前只有2台服务器却要创建3个副本创建不了。
2查看分区和副本情况
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic second kafka在进行初始化的时候选举谁当第一Leader是有一定的算法的。算法保障了Leader不在一个broker里面。 五生产经验 —— 手动调整分区副本的存储 在生产环境中由于服务器配置和性能差异可能需要手动调整分区副本的存储。例如创建three主题4 个分区、2 个副本并将其所有副本存储到broker0和broker1两台服务器上。通过创建副本存储计划并执行kafka-reassign-partitions.sh命令来实现。 1创建一个新的 topic名称为 three。
bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --create --partitions 4 --replication-factor 2 --topic three
2查看分区副本存储情况。
bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --describe --topic three
3创建副本存储计划所有副本都指定存储在 broker0、broker1 中。
vi 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执行副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server bigdata01:9092 --reassignment-json-file increase-replication-factor.json --execute
5验证副本存储计划。
bin/kafka-reassign-partitions.sh -- bootstrap-server bigdata01:9092 --reassignment-json-file increase-replication-factor.json --verify
6查看分区副本存储情况
bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --describe --topic three 六生产经验 ——Leader Partition 负载平衡 Kafka 通常会自动将 Leader Partition 均匀分散在各个机器上但在某些 Broker 宕机后可能导致负载不均衡。相关参数如auto.leader.rebalance.enable默认是true、leader.imbalance.per.broker.percentage默认是 10%和leader.imbalance.check.interval.seconds默认值 300 秒控制着 Leader Partition 的平衡机制。在生产环境中auto.leader.rebalance.enable常被修改为false以避免资源浪费。 七生产经验 —— 增加副本因子 在生产环境中若某个主题重要性提升可能需要增加副本。例如创建four主题初始副本数为 1之后通过手动创建副本存储计划并执行kafka-reassign-partitions.sh命令来增加副本。需要注意的是副本创建后不能直接通过命令修改只能通过这种手动计划的方式增加。
1创建 topic
bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --create --partitions 3 --replication-factor 1 --topic four
通过命令行修改副本是否成功
分区是可以通过语句修改的只能改多不能改少副本创建以后就不能直接修改了。
bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --alter --partitions 3 --replication-factor 3 --topic four
没办法使用命令修改的。
2手动增加副本存储
通过命令查看副本情况
1创建副本存储计划所有副本都指定存储在 broker0、broker1、broker2 中。
vi 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执行副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop11:9092 --reassignment-json-file increase-replication-factor.json --execute
查看副本情况 四、文件存储 一文件存储机制 Topic 数据存储 Topic 是逻辑概念partition 是物理概念每个 partition 对应一个log文件Producer 生产的数据追加到log文件末端。为防止log文件过大导致数据定位效率低下Kafka 采用分片和索引机制将每个 partition 分为多个 segment每个 segment 包括.index文件、.log文件和.timeindex等文件这些文件位于以topic名称分区序号命名的文件夹下如first - 0。 数据位置查找 启动生产者发送消息后可以在/opt/installs/kafka3/datas/下对应主题分区文件夹中查看相关文件。直接查看log日志可能是乱码可通过kafka-run-class.sh kafka.tools.DumpLogSegments命令查看index和log信息。Kafka 使用稀疏索引这使得数据读取速度较快。 二文件清理策略 delete 日志删除 Kafka 中默认的日志即 Segment保存时间为 7 天可通过log.retention.hours最低优先级小时默认 7 天、log.retention.minutes分钟设置后小时设置不起作用、log.retention.ms最高优先级毫秒设置后分钟设置不起作用等参数修改保存时间。log.retention.check.interval.ms负责设置检查周期默认 5 分钟。 delete 日志删除策略有基于时间默认打开以 segment 中所有记录中的最大时间戳作为该文件时间戳和基于大小默认关闭超过设置的所有日志总大小删除最早的 segmentlog.retention.bytes默认等于 - 1表示无穷大两种方式。若一个 segment 中有部分数据过期部分未过期基于时间的删除策略会根据最大时间戳判断整个 segment 是否过期。 compact 日志压缩 log.cleanup.policy compact启用压缩策略对于相同 key 的不同 value 值只保留最后一个版本。压缩后的 offset 可能不连续这种策略适用于特殊场景如消息的 key 是用户 IDvalue 是用户资料等。 五、高效读写数据 分区技术与并行度 Kafka 本身是分布式集群采用分区技术提高了并行度能够同时处理多个分区的数据从而提升整体性能。 稀疏索引与快速定位 读数据采用稀疏索引可以快速定位要消费的数据。与 MySQL 中索引过多会影响写入速度不同Kafka 的稀疏索引在保证快速定位数据的同时不会对写入性能造成过大影响。 顺序写磁盘 Kafka 的 Producer 生产数据时写入log文件的过程是顺序写即一直追加到文件末端。由于磁盘的机械结构特点顺序写能大幅提高写入速度官网数据表明同样的磁盘顺序写能达到 600M/s而随机写只有 100K/s。 页缓存 零拷贝技术 零拷贝技术使得 Kafka 的数据加工处理操作交由生产者和消费者处理Kafka Broker 应用层不关心存储的数据无需经过应用层提高了传输效率。 PageCache 页缓存功能被 Kafka 重度依赖。当有写操作时操作系统将数据写入 PageCache读操作时先从 PageCache 中查找找不到再从磁盘读取。这相当于将空闲内存作为磁盘缓存使用进一步提升了读写性能。 六、总结 本文全面深入地探讨了 Kafka Broker 的各个方面包括其工作流程、节点管理、副本机制、文件存储以及高效读写数据的原理和生产环境中的应用经验。理解这些知识对于在大数据项目中正确使用和优化 Kafka 具有极为重要的意义。无论是在构建实时数据处理管道还是大规模数据存储系统时Kafka 的这些特性都能为我们提供可靠、高效的解决方案。在实际应用中我们需要根据具体的业务需求和硬件环境合理配置 Kafka 的各项参数以充分发挥其优势应对复杂的数据处理挑战。
