网站备备份教程,网站开发多少钱一单,外汇交易网站开发,怎么提升网站排名目录
介绍
KafKa 相关术语
编辑 Kafka的工作流程
生产者向kafka发送数据的流程
Kafka选择分区的模式
Kafka选择分区的模式
数据消费
kafka的文件存储机制
topic、partition和segment
存储和查找message的过程
数据写入过程
数据查找过程
注意事项
kafka管理UI …目录
介绍
KafKa 相关术语
编辑 Kafka的工作流程
生产者向kafka发送数据的流程
Kafka选择分区的模式
Kafka选择分区的模式
数据消费
kafka的文件存储机制
topic、partition和segment
存储和查找message的过程
数据写入过程
数据查找过程
注意事项
kafka管理UI
概述
部署
docker compose 单机部署
Kafka主要配置详解
基本配置
内存调优 介绍
Apache Kafka 是分布式的、基于 发布/订阅 的容错消息系统。 主要特性如下 高吞吐、低延迟可以做到每秒百万级的吞吐量并且时延极低 持久性、可靠性消息被持久化到本地磁盘支持数据备份防止数据丢失具备以时间复杂度为 O(1) 的方式提供消息持久化能力并且可以配置消息有效期以便消费者可以多次消费 KafKa 相关术语 Broker 一个 Kafka 实例就是一个 Broker 每个主机有一个或多个 kafka 的实例通常只有一个所以认为一个 Kafka 主机就是一个 Broker 多个 Broker 可以组成一个集群Cluster其中集群内某个 Broker 会成为集群控制器Cluster Controller它负责管理、统筹集群 Topic主题用来存储不同类别的消息存储消息时需要指定存储在哪个主题下面如发帖发哪种类型的 Partition分区每个topic包含一个或多个partition在创建topic时指定包含的partition数据目的是为了进行分布式存储分区可以提高负载每个分区是不同的磁盘所以会提高负载 Replication副本每个partition分区可以有多个副本分布在不同的Broker上。kafka会选出一个副本作为Leader所有的读写请求都会通过Leader完成Follower只负责备份数据所有Follower会自动从Leader中复制数据当Leader宕机后会从Follower中选出一个新的Leader继续提供服务实现故障自动转移 Message消息是通信数据的基本单位每个消息都属于一个Partition消息都是放在Partition里面的。消息也被称之为事件Event Producer消息的生产者向kafka的一个topic发布消息发布消息时需要指定发布到哪个topic Consumer消息的消费者订阅Topic并读取其发布的消息可以订阅多个主题类似订阅多个微信公众号 Consumer Group消费者组每个Consumer属于一个特定的Consumer Group多个Consumer可以属于同一个Consumer Group各个consumer可以组成一个组每个消息只能被组中的一个consumer消费如果一个消息可以被多个consumer消费的话那么这些consumer必须在不同的组。 ZooKeeper/Kraft协调Kafka的正常运行kafka将元数据信息保存在ZooKeeper/Kraft中但发送给Topic本身的消息数据并不存储在ZK/Kraft中而是存储在磁盘文件中。元数据信息包括kafka有多少个节点、有哪些主题主题叫什么有哪些分区的等消息自身的数据不在ZK中而是在磁盘上对应的分区中 Kafka的工作流程
生产者向kafka发送数据的流程 生产者查询Leaderproducer先从zookeeper的“/brokers/.../state”节点找到该partition的leader 找到Leader之后往Leader写数据producer将消息发送给该leader Leader落盘leader将消息写入本地log Leader通知Follower Follower从Leader中拉取数据replication写入到Follower的本地log后follower向leader发送ack Kafka向生产者回应ACKleader收到所有的replication的ack之后向producer发送ack
Kafka选择分区的模式 直接指定往哪个分区写 指定key然后kafka根据key做hash后决定写哪个分区 各个分区轮询 详细说明Kafka Partition详解 Kafka选择分区的模式 把数据发送给Leader就认为成功效率最高安全性低 把数据发送给Leader等待Leader回复Ack后则认为发送成功 把数据发送给Leader确保Follower从Leader拉取数据回复Ack给LeaderLeader再向生产者回复Ack才认为发送成功安全性最高
数据消费
多个消费者可以组成一个消费者组并用一个标签来标识这个消费者组 如果所有的消费者实例都在同一个消费者组中那么消息记录会被很好的均衡发送到每个消费者实例 如果所有的消费者实例都在不同的消费者组那么每一条消息记录会被广播到每一个消费者实例
各个consumer可以组成一个组每个消息只能被组中的一个consumer消费如果一个消息可以被多个consumer消费的话那么这些consumer必须在不同的组 每个消费者实例可以消费多个分区但是每一个分区最多只能被消费者组中的一个实例消费 kafka的文件存储机制
topic、partition和segment 在kafka文件存储中同一个topic下有多个不同的partition 每个partition就是一个目录partition的命名规则为topic名称有序序号 第一个partition序号从0开始序号最大值为partition数量减一 每个partition的目录下面会有多组segment文件 每个partition相当于一个巨型大文件被平均分配到多个大小都相等的segment数据文件中但每个segment file消息数量不一定相等这种特性方便old segment file快速被删除 每组segment文件包含.index文件、.log文件、.timeindex文件.log文件就是实际存储message的地方.index和.timeindex文件为索引文件用于检索消息 每个partition只需要支持顺序读写就行了segment文件生命周期由服务端配置参数决定 这样做能快速删除无用文件有效提高磁盘利用率 segment文件 segment文件由2大部分组成分别为index file和data file此2个文件一一对应成对出现后缀.index和.log分别表示为segment索引文件、数据文件 segment文件命名规则partion全局的第一个segment从0开始后续每个segment文件名为上一个segment文件最后一条消息的offset值。数值最大为64位long大小19位数字字符长度没有数字用0填充
存储和查找message的过程
数据写入过程
每个Partition都是一个有序并且不可改变的消息记录集合每个partition都是一个有序队列当新的数据写入时就被追加到partition的末尾。
在每个partition中每条消息都会被分配一个顺序的唯一标识这个标识被称为Offset偏移量用于partition唯一标识一条消息。
数据查找过程
在partition中通过offset查找message 查找segment file每一个segment文件名都包含了上一个segment最后一条消息的offset值所以只要根据offset二分查找文件列表就能定位到具体segment文件 通过segment file查找message当定位到segment文件后可以通过对应的.index元数据文件在对应的.log文件中顺序查找对应的offset然后即可拿到数据
注意事项 kafka只能保证在同一个partition内部消息是有序的在不同的partition之间并不能保证消息有序 为什么kafka快因为它把对磁盘的随机读变成了顺序读
kafka管理UI
概述
kafka的管理UI这里主要推荐三个 kafka-ui官方文档地址为https://docs.kafka-ui.provectus.io/overview/readme。相对于EFAK功能显得极为简陋但基本够用最主要是他支持kraft。另外其github上的star数量也远超EFAK Kafdrop 其github地址为GitHub - obsidiandynamics/kafdrop: Kafka Web UI。是一个 Apache 2.0 许可项目在无数的开源选项中Kafdrop 以其简单、快速和易于使用而脱颖而出。同时它是一个开源 Web 项目允许查看来自 Kafka 代理的信息如现有主题、消费者甚至是发送的消息内容。 EFAK原名为kafka-eagle官方文档地址为https://www.kafka-eagle.org/articles/docs/documentation.html。优点是较为完备的管理功能相当酷炫的大盘和监控看板缺点是当前的3.0.1版本仍然不支持Kafka的kraft部署模式虽然官方说是支持了但并没有给出配置说明。 更多的kafka ui可以参考 Kafka ui 搭建以及使用 - 袋鼠社区-袋鼠云丨数栈丨数据中台丨数据治理丨湖仓一体丨数据开发丨基础软件 部署
docker compose 单机部署
services:kafka:image: bitnami/kafka:3.8.0container_name: kafkarestart: alwayshostname: slave02ports:- 9092:9092- 9094:9094environment:- KAFKA_CFG_NODE_ID0- KAFKA_CFG_PROCESS_ROLEScontroller,broker- KAFKA_CFG_LISTENERSPLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://0.0.0.0:9094- KAFKA_CFG_ADVERTISED_LISTENERSPLAINTEXT://slave02:9092,EXTERNAL://192.168.142.155:9094- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAPCONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT- KAFKA_CFG_CONTROLLER_QUORUM_VOTERS0slave02:9093- KAFKA_CFG_CONTROLLER_LISTENER_NAMESCONTROLLER- ALLOW_PLAINTEXT_LISTENERyes- KAFKA_HEAP_OPTS-Xmx512m -Xms512mvolumes:- kafka-conf:/bitnami/kafka/config- kafka-data:/bitnami/kafka/data- /etc/localtime:/etc/localtime:rokafka-ui:container_name: kafka-uiimage: provectuslabs/kafka-ui:latestrestart: alwaysports:- 8080:8080environment:DYNAMIC_CONFIG_ENABLED: trueKAFKA_CLUSTERS_0_NAME: kafka-devKAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092volumes:- kafkaui-app:/etc/kafkaui- /etc/localtime:/etc/localtime:ro
volumes:kafka-conf:kafka-data:kafkaui-app:
启动
docker compose up -d通过 ip 8080 端口访问 Kafka主要配置详解
基本配置
这里主要介绍kraft部署模式下的kafka的配置文件说明。
在KRaft模式下配置文件位于Kafk a目录中的config/kraft/server.properties常用配置说明如下 node.id 节点的**id**一个集群中每个节点id不能重复需要是不小于1的整数。类似Zookeeper的broker.id配置。 controller.quorum.voters 配置集群中Controller节点选举过程中的投票者集群中所有的Controller节点都需要被罗列在这个配置项中其配置格式为id1host1:port1,id2host2:port2,id3host3:port3...。所有的节点都是通过这个配置中的节点列表来得知所有的控制器节点信息以获取集群元数据并得到投票候选者的因此集群中所有节点不论是Broker还是Controller还是混合节点都需要配置这一项。 注意这里只需要写所有的Controller节点和混合节点的id、地址和端口即可这个配置中配置的端口当然是控制器端口。 listeners这个配置项用于指定Kafka服务器监听客户端连接的地址和端口当 Kafka 服务器启动时它将监听listeners配置项中指定的地址和端口等待客户端的连接请求。一般情况下这个配置以PLAINTEXT://或者CONTROLLER://开头意义如下 若这个节点是Broker节点则以PLAINTEXT://开头 若这个节点是Controller节点则以CONTROLLER://开头 若这个节点是混合节点则需要同时配置两者开头的地址 下面给出几个配置示例 PLAINTEXT://:9092 本节点作为Broker节点监听本机所有可用网卡的9092端口需要说明的是该监听会监听在hostname上也就是说如果使用默认监听客户端需要能解析broker的hostname PLAINTEXT://127.0.0.1:9092 本节点作为Broker节点监听本地的9092端口这样仅接受来自本地的请求 CONTROLLER://:10000 本节点作为Controller节点监听本机所有可用网卡的10000端口使用10000端口作为控制器端口 PLAINTEXT://:9092,CONTROLLER://:9093 本节点作为混合节点监听本机所有可用网卡的9092和9093端口其中9092作为客户端通信端口9093作为控制器端口 advertised.listeners 这个配置容易和listeners混淆事实上它们是有较大的区别的。该配置项指定Kafka服务器广播给客户端的地址和端口通常配置为Kafka所在服务器的直接提供给客户端访问的地址。当客户端生产者或消费者尝试连接到Kafka服务器时它首先会获取Kafka服务器广播的地址和端口也就是advertise.listeners配置所指定的地址和端口然后才会使用advertise.listeners配置所指定的地址和端口来建立与Kafka服务器的连接。 这里的问题是既然客户端要连接Kafka那一定是已经知道了Kafka对外的地址端口了那为什么连接的时候还需要获取一下广播的地址端口再进行连接呢 事实上Kafka设计这个配置是为了解决下面较为复杂的网络场景 多网络接口的主机部署在一个多网络接口的主机部署Kafka时Kafka服务器可能会监听多个地址和端口这些地址和端口可能与客户端实际访问的地址和端口不同advertise.listeners允许服务器指定一个公开的、可访问的地址和端口以便客户端能够正确连接 NAT/代理环境在某些网络环境下Kafka服务器位于一个私有网络中客户端位于一个公共网络中两者之间可能存在网络地址转换NAT或代理在这种情况下Kafka服务器的内部地址和端口对客户端来说是不可访问的。通过使用advertise.listenersKafka服务器可以将一个公共地址和端口广播给客户端使得客户端能够通过公共网络连接到服务器 容器环境例如你把Kafka放在Docker容器中运行按照默认配置Kafka服务端只会监听容器网络的9092端口我们知道外部不能直接访问容器的网络而是需要使用网络映射假设你把Kafka容器的9092端口映射至了宿主机9095端口也就是说外部需要通过9095端口访问到Kafka容器的9092端口那么你就配置advertise.listeners为PLAINTEXT://服务器外网地址:9095客户端就可以正确访问容器中的Kafka了 process.roles 这是KRaft模式下专门的配置用于配置这个节点的类型可以配置为下列值 broker 表示这个节点是Broker节点充当消息队列的角色 controller 表示这个节点是Controller节点充当元数据存放和管理的角色 broker,controller 表示这个节点同时担任Broker和Controller的角色也称作混合节点 如果没有配置这个选项则Kafka会以Zookeeper模式运行。 这里有下列注意事项 如果设定节点为controller 则不能配置advertised.listeners可以将其注释掉或者删掉 listeners需要配置为CONTROLLER://开头建议配置为CONTROLLER://:9093 如果设定节点为broker 则需要配置advertised.listeners为服务器外网地址和端口这和Zookeeper模式中相同 listeners需要配置为PLAINTEXT://开头建议配置为PLAINTEXT://:9092 如果设定节点为混合节点 同样需要配置advertised.listeners为服务器外网地址和端口 listeners需要同时配置CONTROLLER://和PLAINTEXT://建议配置为PLAINTEXT://:9092,CONTROLLER://:9093 在开发环境或者小规模集群可以全部使用混合节点如果是生产环境就建议设定好每个节点的类型了并且通常需要先启动Controller节点再启动Broker节点。 事实上我们发现Kafka的KRaft配置目录config/kraft下有三个配置文件其中server.properties是混合节点的配置模板而broker.properties和controller.properties分别是Broker节点和Controller节点的配置模板大家如果要设定节点类型可以直接使用对应的配置文件将对应配置文件需要修改的部分修改一下然后将上述格式化数据目录命令和启动命令中的配置文件路径改变一下即可这样可以省略我们设定process.roles和listeners或者控制器节点删除advertise.listeners配置的操作。 socket.send.buffer.bytes 每次发送的数据包的最大大小单位字节 socket.receive.buffer.bytes 每次接收的数据包的最大大小单位字节 socket.request.max.bytes 接收的最大请求大小单位字节 num.partitions 指定创建的Topic的默认分区数 auto.create.topics.enable当topic不存在时是否允许自动创建true/false default.replication.factor指定创建的topic默认的分区的副本数默认为1 logs.dir指定kafka的数据存储路径多个路径可以用逗号分隔示例 logs.dir: /data1,/data2,/data3,/data4 log.retention.hours168 设置消息过期时间全局 log.cleanup.policydelete 设置过期消息的处理策略默认为delete
内存调优
Kafka 是基于 Java 开发的因此它的内存配置需要通过 JVM 参数进行设置。在 Kafka 的启动脚本中有一个环境变量KAFKA_HEAP_OPTS可以用来设置 JVM 的内存参数。例如可以将 KAFKA_HEAP_OPTS设置为 -Xmx4g -Xms4gKafka对堆内存的占用相对不高一般建设6-8g就够了。
Kafka 内存配置 除了 JVM 参数之外还需要在 Kafka 的配置文件中配置内存。具体来说需要关注以下两个参数 log.retention.bytes用于设置 Kafka 存储消息的阈值当日志文件大小达到这个阈值时Kafka 会删除最旧的消息。因此这个参数应该根据存储需求和可用内存进行 log.segment.bytes这个参数设置了 Kafka 日志文件的大小。如果设置得太小会导致频繁的文件切换增加文件系统的开销如果设置得太大会占用过多的内存。因此这个参数也需要根据存储需求和可用内存进行调整
参考Kafka两种集群详解和搭建教程-阿里云开发者社区
