添加网站图标,用phpcms建站的网站,设计网装修,企业系统建设1、kafka消息发送的流程#xff1f; 在消息发送过程中涉及到两个线程#xff1a;一个是 main 线程和一个 sender 线程。在 main 线程中创建了一个双端队列 RecordAccumulator。main 线程将消息发送给双端队列#xff0c;sender 线程不断从双端队列 RecordAccumulator 中拉取…1、kafka消息发送的流程 在消息发送过程中涉及到两个线程一个是 main 线程和一个 sender 线程。在 main 线程中创建了一个双端队列 RecordAccumulator。main 线程将消息发送给双端队列sender 线程不断从双端队列 RecordAccumulator 中拉取消息发送到 Kafka Broker。 外部数据生产者发送消息经过 main 线程其中数据在 main 线程中先经过拦截器、序列化器、分区器后被推送到双端队列中。双端队列RecordAccumulator缓冲区总大小默认是 32MB这个值可以调整。数据在双端队列中被分为多个双端队列的容器当双端队列的容器 batch.size 的数据累计到 16KB 的时候会自动发送到 sender 端还有一种情况是数据不满足 16KB 的时候可以设置 sender 等待 linger.ms 设置时间时间到了之后就会发送数据单位是 ms默认值是 0ms延迟建议设置在 5 - 100ms 之间。 当数据满足以上条件的时候sender 从双端队列 RecordAccumulator 去拉取数据。当数据到达 sender 线程的时候数据会被封装成多个分区的请求默认一个连接最多可以发送 5 个请求发送到 Kafka 集群。每发送一个请求sender 会等待 Kafka 集群返回一个确认ack。如果消息发送成功会收到 ack 确认如果没有收到 ack比如网络故障、Kafka 集群繁忙等情况为了重新发送消息有重试机制重试发送的时间间隔默认是 100ms并且可以设置重试次数默认是Integer.MAX_VALUE即 2147483647。经过这些步骤后消息就会进入 Kafka 集群。 2、Kafka 的设计架构你知道吗 kafka的框架主要有以下的几个方面构成 1、Producer消息生产者负责创建消息并将其发送到 Kafka 集群 2、Consumer消息消费者从 Kafka 集群中拉取消息进行处理 3、Consumer GroupCG消费者组由多个 consumer 消费者组成消费者组内的每个消费者负责消费不同分区的数据一个分区只能有一个组内的消费者消费消费者租之间互不影响。所有消费者组即消费者组是逻辑上的一个订阅者。 4、Broker一台kafka服务器就是一个broker一个集群由多个broker组成。 5、Topic队列生产者消费者都面向一个topic 6、Partition为了实现扩展性一个非常大的 topic 可以分布到多个 broker即服务器上一个 topic 可以分为多个 partition每个 partition 是一个有序的队列。 7、Replica副本每一个topic都有很多个副本一个Leader和多个Follower 8、Leader主副本 每个分区多个副本的“领导者”生产者发送数据的对象 和 消费者拉取数据的对象都属 Leader 9、follower从副本 保持和Leader 数据同步备份数据防止Leader崩溃数据丢失。 Leader 发生故障时某个 Follower 会成为新的 Leader。 3、Kafka 分区的目的 1便于合理使用存储资源每个Partition在一个Broker上存储可以把海量的数据按照分区切割成一块一块的数据存储在多台Broker上。合理控制分区的任务可以实现负载均衡的效果。 2提高并行度生产者可以以分区为单位发送数据消费者可以以分区为单位进行消费数据。 4、你知道 Kafka 是如何做到消息的有序性 保证消息的有序性需要依赖以下几个机制和策略 1、单一分区内的消息有序每一个topic内只有一个partation分区因为分区内有序可以将所有消息发送到一个分区内在创建一个topic的时候可以将分区数调为1 2、使用幂等性生产者默认自动开启幂等性主要是防止生产者重复发送消息并且发送到一个分区保证消息的有序性 一、分区内消息有序性 分区机制 Kafka 的 Topic 是由一个或多个分区Partition组成的。在每个分区内部消息是按照一定的顺序存储和追加的。这个顺序通常是由生产者发送消息的时间顺序或者由生产者指定的消息键key来决定的。当生产者发送消息时如果只使用一个分区即将 Topic 的分区数设置为 1那么所有消息会按照发送的先后顺序依次追加到这个分区的日志文件中。但是将 Topic 的分区数设置为 1 会限制系统的吞吐量。因为 Kafka 通过分区来实现并行处理多个分区可以让多个生产者和消费者同时工作。所以这种方式在对消息顺序要求极高且对吞吐量要求不高的场景下比较适用。 消息键key的使用 生产者可以为消息指定一个键key。当指定了键之后Kafka 会根据这个键来计算消息应该被发送到哪个分区。通常是通过对键进行哈希运算将具有相同键的消息发送到同一个分区。 二、幂等性生产者保证消息有序性 1.幂等性原理 Kafka 从 0.11 版本开始引入了幂等性生产者。幂等性是指生产者在发送消息时对于具有相同消息键key和分区Partition的消息无论发送多少次在 Kafka 服务端只会持久化一次。生产者在发送消息时会为每个消息生成一个唯一的标识符PIDProducer ID和一个序列号Sequence Number。PID 是生产者的唯一标识序列号用于标记每个消息的发送顺序。当 Kafka 服务端接收到消息时会根据 PID 和序列号来判断消息是否已经接收过。 2.结合分区的作用 幂等性生产者和分区机制相结合可以更好地保证消息的有序性。当消息具有相同的键并且被发送到同一个分区时幂等性可以防止因为生产者重试等原因导致的重复消息插入到分区中间破坏消息顺序。 5、ISR、OSR、AR 是什么 AR ISR OSR ARAssigned Repllicas是指一个分区的所有副本集合Kafka 分区中的所有副本包含Leader统称为 AR ISR是 AR 的一个子集表示和 Leader 保持同步的副本集合包括 Leader 本身。如果 Follower 长时间未向 Leader 发送通信请求或落后太多而不能在规定时间内同步数据则该 Follower 将被踢出 ISR。Leader 发生故障之后kafka就会从 ISR 中选举新的 Leader。 OSR是 AR 中除 ISR 之外的副本集合表示 Follower 与 Leader 副本同步时延迟过多而不能跟上 Leader 进度的副本。 判断一个 Follower 是否在 ISR 中的依据是 它在可配置的时间由 replica.lag.time.max.ms 参数设定默认 10s内能够跟上 Leader 副本的消息写入进度。这里的进度主要是通过比较副本最后一次写入消息的偏移量offset来衡量 6、Kafka 在什么情况下会出现消息丢失 0生产者发送过来的数据不需要等数据落盘应答 假如发送了Hello 和 World两个信息Leader直接挂掉数据就会丢失 生产者 --- Kafka集群 一放进去就跑 1生产者发送过来的数据Leader收到数据后应答 生产者把数据发送给了Leaderleadr收到数据后应答生产者就以为数据发送成功了。但是此时leader挂掉了但是Follower 并没有同步数据过来Follower此时变成了leader这个时候数据就会丢失因为此时的Leader的数据是follower没有同步完的数据导致消息丢失。图1 -1all生产者发送过来的数据Leader和ISR队列里所有节点收齐数据后应答 leader收到数据后所有的follower开始同步数据但是这时候有一个follower副本挂掉了然后迟迟不能同步数据但是leadr也挂断了然后选举了那个坏的follower作为leader导致数据丢失。图2 思考Leader收到数据所有Follower都开始同步数据 但有一个Follower因为某种故障迟迟不能与Leader进行同步那这个问题怎么解决呢 解决方案 Leader维护了一个动态的in-sync replica setISR意味和 Leader保持同步的FollowerLeader集合(leader0ISR:0,1,2) 如果Follower长时间未向Leader发送通信请求或同步数据则该Follower将被踢出ISR。该时间阈值由replica.lag.time.max.ms参数设定默认30s。 例如2超时(leader:0, isr:0,1)。 这样就不用等长期联系不上或者已经故障的节点。 7、怎么尽可能保证 Kafka 的可靠性 如果分区副本设置为1个只有一个leader或者ISR里应答的最小副本数量 min.insync.replicas 默认为1设置为1和ack1的效果是一样的仍然有丢数的风险Leader0ISR : 0。 • 数据完全可靠条件 ACK级别设置为-1 分区副本大于等于2 ISR里应答的最小副本数量大于等于2 副本数是2但是ISR中不一定有两个因为会挂掉。 可靠性总结 acks0生产者发送过来数据就不管了可靠性差效率高 acks1生产者发送过来数据Leader应答可靠性中等效率中等 acks-1生产者发送过来数据Leader和ISR队列里面所有Follwer应答可靠性高效率低 在生产环境中acks0很少使用acks1一般用于传输普通日志允许丢个别数据 acks-1一般用于传输和钱相关的数据对可靠性要求比较高的场景。 8、Kafka中如何做到数据唯一即数据去重 重复原因 当ask为-1的时候可能出现数据重复问题数据发送给了Leaderfollower也同步成功了此时准备应答 ask为-1的时候Leader挂了Follower顶上由于发送者不知道数据已经发送成功会给新的Leader再发消息此时数据重复。 去重 Exactly Once数据唯一 开启幂等性幂等性就是producer无论向broker发送多少重复的数据只能是持久化的一条数据保证数据的不重复幂等性是默认开启的类似于distinct 数据唯一幂等性ack-1分区副本数大于等于2isr最下副本数大于等于2不停机情况下 但是幂等性只能保证在服务器不停机的情况下不会出现重复数据当服务器运行的时候突然停机就会出现leader重新阅读的情况导致数据丢失这时候需要开启事务 事务开启事务的前提要开启幂等性生产者请求kafka服务器一个唯一id、这个id不能重复幂等性需要然后服务器返回这个id、然后消费者发送信息请求给kafka然后kafka给生产者一个返回值这时候生产者发送数据给topickafka事务协调器进行持久化请求给transacti分区这个分区默认有50个分区每个分区负责一部分事务 当开始幂等性事务ack-1分区副本大于等于2isr最小分区副本大于等于2 就可以保证数据的唯一性(可能存在停机的情况下用) 9、生产者如何提高吞吐量 在消息发送过程中双端队列中调节batch.size批次大小默认16k可以调大一点linger.ms信息等待时间调到5~100mscompression.type压缩snappy 压缩类型有none、gzip、snappy、lz4 和 zstdRecordAccumulator缓冲区大小修改为64m默认32m 10、zk在kafka集群中有何作用 集群元数据管理 在 Kafka 集群中ZookeeperZK用于存储和管理集群的元数据信息。元数据包括主题Topic的信息如主题的名称、分区Partition数量、每个分区的副本Replica分配情况等。同时Zookeeper 还记录了 Broker 的信息包括每个 Broker 的唯一标识broker.id、主机地址和端口号等。这些信息使得 Kafka 集群中的各个组件如生产者、消费者能够方便地发现和定位集群中的其他组件以及了解集群的整体拓扑结构。 集群成员管理 Zookeeper 用于维护 Kafka 集群中 Broker 的成员状态。当一个 Broker 加入或离开集群时它会在 Zookeeper 中更新相应的状态信息。例如通过在 Zookeeper 中创建或删除临时节点来表示 Broker 的加入或离开。这种成员管理方式对于集群的自动发现和动态扩展非常重要。其他 Broker 和客户端生产者、消费者可以通过监听 Zookeeper 中的节点变化来及时了解集群的成员动态。当新的 Broker 加入时生产者和消费者可以根据新的集群状态调整消息的发送和接收策略比如重新平衡消费者组内的分区分配。 选举机制支持 在 Kafka 中一些关键的选举过程依赖于 Zookeeper。例如当分区Partition的主副本Leader出现故障时需要从副本Follower中选举出一个新的 Leader。这个选举过程通常是借助 Zookeeper 来实现的。Zookeeper 提供了一种分布式协调服务通过它可以实现基于多数投票或者其他选举算法的 Leader 选举。在选举过程中各个副本会在 Zookeeper 中竞争创建一个特定的临时节点成功创建节点的副本将成为新的 Leader。这种选举机制保证了在主副本故障时能够快速、公平地选出新的 Leader 来维持集群的正常运行。 消费者组管理与偏移量存储 对于消费者组Consumer GroupZookeeper 用于协调消费者组内的消费者。它可以记录消费者组的成员信息如每个消费者的唯一标识等。并且在旧版本的 Kafka 中消费者的消费偏移量Offset也是存储在 Zookeeper 中的。当消费者组内的消费者发生变化如消费者加入或离开时Zookeeper 帮助协调分区的重新分配。同时消费者会定期将自己的消费偏移量提交到 Zookeeper这样在消费者重启或者重新加入消费者组时可以从 Zookeeper 中获取上次的消费偏移量从而继续从上次的位置进行消费。不过在较新的 Kafka 版本中也可以将偏移量存储在 Kafka 内部主题__consumer_offsets中减少对 Zookeeper 的依赖。 11、简述kafka集群中的Leader选举机制 控制器Controller选举 在 Kafka 集群启动时每个 Broker 都会尝试在 ZookeeperZK中创建一个特殊的临时节点例如 “/controller”来竞争成为控制器。第一个成功创建该节点的 Broker 就会被选举为控制器Controller Leader。控制器负责整个集群的管理工作包括监控 Broker 的上线和下线情况以及管理所有 Topic 分区副本的 Leader 选举等诸多事务。例如当有新的 Broker 加入集群时控制器会感知到并进行相应的处理当有 Broker 下线时控制器也会协调相关的资源回收等操作。被选举为控制器的 Broker 会将自身的节点信息存储到 Zookeeper 的相关节点如记录 ISR 等信息的节点中方便其他组件获取集群管理相关的信息。 分区副本 Leader 选举基于控制器 当一个分区的 Leader 副本出现故障时控制器会负责协调选举新的 Leader。控制器首先会查看该分区的 ISRIn - Sync Replicas列表。ISR 是与原 Leader 保持同步的副本集合这些副本中的数据与 Leader 的数据在一定程度上是一致的这一一致性程度是由replica.lag.time.max.ms参数控制的。选举新 Leader 的依据是在 ISR 中存活的副本并且通常会按照 ARAssigned Replicas即该分区所有分配的副本中的某种顺序进行选择。例如假设 AR 是 [1,0,2]ISR 也是 [1,0,2]那么按照这个顺序优先考虑副本 1 作为新的 Leader。如果副本 1 不可用就会考虑副本 0以此类推。这种轮询查询的方式确保了能够从与原 Leader 数据同步程度较高的副本中快速选出新的 Leader从而减少对消息服务的中断时间。控制器在选举出新的 Leader 后会向集群中的所有 Broker 发送通知告知它们该分区新的 Leader 是谁以便其他副本能够及时调整自己的角色从副本开始从新 Leader 处同步数据同时也让生产者和消费者知道可以和新的 Leader 进行消息交互。 当选举leader的时候kafka集群中有一个broker的controller会被选举为controller Leader负责管理集群broker的上下线所有topic的分区副本分配和Leader选举等工作。 contorllet依赖于zookeeper启动集群的时候会在zk中注册记录然后将节点信息上传到zk isr【】中谁先注册谁就是leader如果集群的某一个leader挂掉的话contorller会监听到broker节点变化然后选举新的leader选举规则在ISR中存活为前提按照ARkafka分区中的所有副本统称中的排序优先选例如AR【102】 ISR【102】那么选举顺序就是102轮询查询最后更新Leader和ISR。 12、kafka是如何处理数据乱序问题的。 出现乱序的原因 1生产者在发送3请求的时候发生异常发生异常需要重新发送所以排在了后面在进行落盘的时候先落盘12 落盘3的时候发现是4需要等等到3出现为止然后将 34 5排序排序后再进行落盘。 顺序错乱了会自动排序开启幂等性。 kafka在1.0之前版本保证分区有序是采用max.in.flight.requests.per.connection1不需要考虑是否开启幂等性 1.0及以后的版本采用开启幂等性然后设置max.in.flight.requests.per.connection需要设置小于等于5 未开启幂等性max.in.flight.requests.per.connection需要设置为1 启用幂等性kafka服务端会缓存producer的5个request的元数据这都可保证最近5个request的数据是有序的 13、kafka中节点如何服役和退役 服役 1、需要创建以下json创建一个均衡的主题然后生成一个负载均衡的计划kafka-reassign-partitions.sh --bootstrap-server bigdata01:9092 --topics-to-move-json-file topics-to-move.json --broker-list 0,1,2,3 --generate 2、执行之前写的json然后将生成的未来分区策略复制 3、创建副本存储计划所有副本存储在 broker0、broker1、broker2、broker3 中 4、然后执行副本存储计划 5、验证副本存储计划 退役先按照退役一台节点生成执行计划然后按照服役时操作流程执行负载均衡 1、创建json 创建一个负载均衡主题 2、创建执行计划 3、创建副本存储计划 4、执行副本 5、验证副本存储计划 总结 kafka节点的服役和退役的节点操作是相同均是将主题中的数据均匀的分配到新增或者删除节点之外的borker中。 14、Kafka中Leader挂了Follower挂了然后再启动数据如何同步 由于数据同步的时候先进入Leader,随后同步给Follower假如Follower挂掉了Leader和其他的Follower 继续往前存储数据挂掉的节点从ISR集合中剔除此时挂掉的Follower又重启了它会先从上一次挂掉的节点的HW开始同步数据直到追上最后一个Follower为止,此时会重新回归ISR。 Follower 恢复启动当 Follower 节点重新启动后它会向 Controller 发送请求获取自己所属分区的 leader 信息以及 ISR 集合信息。然后Follower 会根据这些信息与 leader 建立连接并从 leader 副本拉取自己落后的数据进行同步。在同步过程中Follower 会不断向 leader 发送 FetchRequest 请求获取消息数据并将其写入本地日志文件直到与 leader 的数据差距缩小到 ISR 所允许的范围内此时该 Follower 会重新加入 ISR 集合完成数据同步过程。 Leader 恢复启动如果 Leader 节点重新启动它首先会检查本地的日志数据是否完整。如果数据完整它会向 Controller 注册自己并等待 Controller 协调其他副本与它进行数据同步如果有需要。Controller 会根据之前记录的元数据信息通知其他 Follower 副本与新启动的 leader 进行同步操作确保各个副本的数据一致性。如果 leader 节点发现本地数据不完整例如在宕机前某些数据未持久化到磁盘则可能需要根据具体情况进行数据恢复操作如从其他副本或备份中获取缺失的数据然后再进行同步操作以保证整个分区的数据完整性和一致性。 15、kafka中初始化的时候Leader选举有一定的规律如何打破这个规律呢 调整副本列表顺序 可以手动调整 ARAssigned Replicas或 ISRIn - Sync Replicas中副本的顺序。例如在创建主题或修改分区副本分配时改变副本的排列顺序使得选举时优先考虑不同的副本。 修改选举策略相关参数 改变replica.lag.time.max.ms参数影响副本进入或退出 ISR 的条件。这样可能会改变选举时 ISR 中的副本集合从而打破原有的选举顺序。例如缩短这个时间让一些原本可能在 ISR 中的副本被排除在外改变了选举的候选范围。
