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 Group（CG）：消费者组，由多个 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、使用幂等性生产者（默认自动开启），幂等性主要是防止生产者重复发送消息，并且发送到一个分区，保证消息的有序性

一、分区内消息有序性

1. 分区机制

• • Kafka 的 Topic 是由一个或多个分区（Partition）组成的。在每个分区内部，消息是按照一定的顺序存储和追加的。这个顺序通常是由生产者发送消息的时间顺序或者由生产者指定的消息键（key）来决定的。
  • 当生产者发送消息时，如果只使用一个分区（即将 Topic 的分区数设置为 1），那么所有消息会按照发送的先后顺序依次追加到这个分区的日志文件中。
  • 但是，将 Topic 的分区数设置为 1 会限制系统的吞吐量。因为 Kafka 通过分区来实现并行处理，多个分区可以让多个生产者和消费者同时工作。所以这种方式在对消息顺序要求极高，且对吞吐量要求不高的场景下比较适用。

1. 消息键（key）的使用

• • 生产者可以为消息指定一个键（key）。当指定了键之后，Kafka 会根据这个键来计算消息应该被发送到哪个分区。通常是通过对键进行哈希运算，将具有相同键的消息发送到同一个分区。

二、幂等性生产者保证消息有序性

1.幂等性原理

• Kafka 从 0.11 版本开始引入了幂等性生产者。
• 幂等性是指生产者在发送消息时，对于具有相同消息键（key）和分区（Partition）的消息，无论发送多少次，在 Kafka 服务端只会持久化一次。
• 生产者在发送消息时，会为每个消息生成一个唯一的标识符（PID，Producer ID）和一个序列号（Sequence Number）。PID 是生产者的唯一标识，序列号用于标记每个消息的发送顺序。当 Kafka 服务端接收到消息时，会根据 PID 和序列号来判断消息是否已经接收过。

2.结合分区的作用

• 幂等性生产者和分区机制相结合可以更好地保证消息的有序性。当消息具有相同的键并且被发送到同一个分区时，幂等性可以防止因为生产者重试等原因导致的重复消息插入到分区中间，破坏消息顺序。

5、ISR、OSR、AR 是什么？

AR = ISR + OSR

AR（Assigned 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收到数据后应答

生产者把数据发送给了Leader，leadr收到数据后应答，生产者就以为数据发送成功了。但是此时，leader挂掉了，但是Follower 并没有同步数据过来，Follower此时变成了leader，这个时候数据就会丢失，因为此时的Leader的数据是follower没有同步完的数据，导致消息丢失。（图1）

-1（all）：生产者发送过来的数据，Leader和ISR队列里所有节点收齐数据后应答

leader收到数据后所有的follower开始同步数据，但是这时候有一个follower副本挂掉了，然后迟迟不能同步数据，但是leadr也挂断了然后选举了那个坏的follower作为leader导致数据丢失。（图2）

思考：Leader收到数据，所有Follower都开始同步数据， 但有一个Follower，因为某种故障，迟迟不能与Leader进行同步，那这个问题怎么解决呢？

解决方案：

Leader维护了一个动态的in-sync replica set（ISR），意味和 Leader保持同步的Follower+Leader集合(leader：0，ISR: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，和ack=1的效果是一样的，仍然有丢数的风险（Leader：0，ISR : 0）。

• 数据完全可靠条件 = ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

副本数是2，但是ISR中不一定有两个，因为会挂掉。

可靠性总结：

acks=0，生产者发送过来数据就不管了，可靠性差，效率高；

acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；

acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；

在生产环境中，acks=0很少使用；acks=1，一般用于传输普通日志，允许丢个别数据；

acks=-1，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。

8、Kafka中如何做到数据唯一，即数据去重？

重复原因：

当ask为-1的时候，可能出现数据重复问题，数据发送给了Leader，follower也同步成功了，此时准备应答 ask为-1的时候Leader挂了，Follower顶上，由于发送者不知道数据已经发送成功，会给新的Leader再发消息，此时数据重复。

去重：

Exactly Once：数据唯一

开启幂等性（幂等性，就是producer无论向broker发送多少重复的数据，只能是持久化的一条数据，保证数据的不重复，幂等性是默认开启的，类似于distinct）

数据唯一：幂等性+ack-1+分区副本数大于等于2+isr最下副本数大于等于2（不停机情况下）

但是幂等性只能保证在服务器不停机的情况下不会出现重复数据，当服务器运行的时候突然停机，就会出现leader重新阅读的情况导致数据丢失，这时候需要开启事务

事务：开启事务的前提要开启幂等性，生产者请求kafka服务器一个唯一id、这个id不能重复（幂等性需要），然后服务器返回这个id、，然后消费者发送信息请求给kafka，然后kafka给生产者一个返回值这时候生产者发送数据给topic，kafka事务协调器进行持久化请求给transacti分区（这个分区默认有50个分区，每个分区负责一部分事务）

当开始幂等性+事务+ack-1+分区副本大于等于2+isr最小分区副本大于等于2 就可以保证数据的唯一性(可能存在停机的情况下，用)

9、生产者如何提高吞吐量？

在消息发送过程中双端队列中调节batch.size：批次大小，默认16k（可以调大一点），linger.ms：信息等待时间调到（5~100ms），compression.type：压缩snappy （压缩类型有none、gzip、snappy、lz4 和 zstd），RecordAccumulator：缓冲区大小，修改为64m（默认32m）

10、zk在kafka集群中有何作用？

1. 集群元数据管理

• • 在 Kafka 集群中，Zookeeper（ZK）用于存储和管理集群的元数据信息。元数据包括主题（Topic）的信息，如主题的名称、分区（Partition）数量、每个分区的副本（Replica）分配情况等。
  • 同时，Zookeeper 还记录了 Broker 的信息，包括每个 Broker 的唯一标识（broker.id）、主机地址和端口号等。这些信息使得 Kafka 集群中的各个组件（如生产者、消费者）能够方便地发现和定位集群中的其他组件，以及了解集群的整体拓扑结构。

1. 集群成员管理

• • Zookeeper 用于维护 Kafka 集群中 Broker 的成员状态。当一个 Broker 加入或离开集群时，它会在 Zookeeper 中更新相应的状态信息。例如，通过在 Zookeeper 中创建或删除临时节点来表示 Broker 的加入或离开。
  • 这种成员管理方式对于集群的自动发现和动态扩展非常重要。其他 Broker 和客户端（生产者、消费者）可以通过监听 Zookeeper 中的节点变化来及时了解集群的成员动态。当新的 Broker 加入时，生产者和消费者可以根据新的集群状态调整消息的发送和接收策略，比如重新平衡消费者组内的分区分配。

1. 选举机制支持

• • 在 Kafka 中，一些关键的选举过程依赖于 Zookeeper。例如，当分区（Partition）的主副本（Leader）出现故障时，需要从副本（Follower）中选举出一个新的 Leader。这个选举过程通常是借助 Zookeeper 来实现的。
  • Zookeeper 提供了一种分布式协调服务，通过它可以实现基于多数投票或者其他选举算法的 Leader 选举。在选举过程中，各个副本会在 Zookeeper 中竞争创建一个特定的临时节点，成功创建节点的副本将成为新的 Leader。这种选举机制保证了在主副本故障时，能够快速、公平地选出新的 Leader 来维持集群的正常运行。

1. 消费者组管理与偏移量存储

• • 对于消费者组（Consumer Group），Zookeeper 用于协调消费者组内的消费者。它可以记录消费者组的成员信息，如每个消费者的唯一标识等。并且，在旧版本的 Kafka 中，消费者的消费偏移量（Offset）也是存储在 Zookeeper 中的。
  • 当消费者组内的消费者发生变化（如消费者加入或离开）时，Zookeeper 帮助协调分区的重新分配。同时，消费者会定期将自己的消费偏移量提交到 Zookeeper，这样在消费者重启或者重新加入消费者组时，可以从 Zookeeper 中获取上次的消费偏移量，从而继续从上次的位置进行消费。不过，在较新的 Kafka 版本中，也可以将偏移量存储在 Kafka 内部主题（__consumer_offsets）中，减少对 Zookeeper 的依赖。

11、简述kafka集群中的Leader选举机制

1. 控制器（Controller）选举

• • 在 Kafka 集群启动时，每个 Broker 都会尝试在 Zookeeper（ZK）中创建一个特殊的临时节点（例如 “/controller”）来竞争成为控制器。第一个成功创建该节点的 Broker 就会被选举为控制器（Controller Leader）。
  • 控制器负责整个集群的管理工作，包括监控 Broker 的上线和下线情况，以及管理所有 Topic 分区副本的 Leader 选举等诸多事务。例如，当有新的 Broker 加入集群时，控制器会感知到并进行相应的处理；当有 Broker 下线时，控制器也会协调相关的资源回收等操作。
  • 被选举为控制器的 Broker 会将自身的节点信息存储到 Zookeeper 的相关节点（如记录 ISR 等信息的节点）中，方便其他组件获取集群管理相关的信息。

1. 分区副本 Leader 选举（基于控制器）

• • 当一个分区的 Leader 副本出现故障时，控制器会负责协调选举新的 Leader。控制器首先会查看该分区的 ISR（In - Sync Replicas）列表。ISR 是与原 Leader 保持同步的副本集合，这些副本中的数据与 Leader 的数据在一定程度上是一致的，这一一致性程度是由replica.lag.time.max.ms参数控制的。
  • 选举新 Leader 的依据是在 ISR 中存活的副本，并且通常会按照 AR（Assigned 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中存活为前提，按照AR（kafka分区中的所有副本统称）中的排序优先选）例如AR【1，0，2】 ISR【1，0，2】那么选举顺序就是1，0，2轮询查询，最后更新Leader和ISR。

12、kafka是如何处理数据乱序问题的。

出现乱序的原因：

1）生产者在发送3请求的时候，发生异常，发生异常需要重新发送，所以排在了后面，在进行落盘的时候，先落盘1，2 ，落盘3的时候发现是4，需要等，等到3出现为止，然后将 3，4 ，5排序，排序后再进行落盘。

顺序错乱了，会自动排序（开启幂等性）。

kafka在1.0之前版本保证分区有序，是采用：max.in.flight.requests.per.connection=1（不需要考虑是否开启幂等性）

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选举有一定的规律，如何打破这个规律呢？

1. 调整副本列表顺序

• • 可以手动调整 AR（Assigned Replicas）或 ISR（In - Sync Replicas）中副本的顺序。例如，在创建主题或修改分区副本分配时，改变副本的排列顺序，使得选举时优先考虑不同的副本。

1. 修改选举策略相关参数

• • 改变replica.lag.time.max.ms参数，影响副本进入或退出 ISR 的条件。这样可能会改变选举时 ISR 中的副本集合，从而打破原有的选举顺序。例如，缩短这个时间，让一些原本可能在 ISR 中的副本被排除在外，改变了选举的候选范围。