面试专题--Kafka

Follower

生产者数据可靠性如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。该时间阈值由replica.lag.timme.max.ms参数设定，默认30s。如果分区副本设置为2个，或者ISR里应答的最小副本数量(min.insync.replicas默认为1)设置为1，和ack=1的效果是样的，仍然有丢数的风险所以数据可靠性设置为：1：acks设置为-12：分区副本数AR大于等于33：ISR应答的副本数大于等于是2消息顺序性和一致性由下游消费者自己控制

sender读取数据

HW

main线程应用程序运行的线程，当主线程调用 send() 方法发送消息时，会依次触发拦截器（Interceptor）、序列化器（Serializer）和分区器（Partitioner），然后将处理后的消息放入缓冲区（RecordAccumulator）。sender 线程在 Kafka 中，sender 线程是生产者客户端的一部分，负责从生产者的缓冲区中提取消息并发送到 Kafka 集群。可以说，sender 线程是 Kafka Producer 的一个内部线程，专门用于处理消息的网络传输职责1：从缓冲区中读取消息2：选择分区和目标 Broker，根据消息的分区策略决定目标分区，并找到该分区的 Leader Broker。3：组装请求，将多个消息组合成一个批次（Batch），然后组装成 Kafka 请求（ProduceRequest）。4：发送数据，通过网络发送请求到目标 Kafka Broker。5：处理响应，接收 Kafka Broker 的响应，包括成功或失败的信息，并将结果通知回调函数（如果有）。batch.size只有数据积累到batch.size之后，sender才会读取数据并发送。默认16klingerms如果数据迟迟未达到batch.size，sender等待linger.ms设置的时间到了之后就会发送数据。单位ms，默认值是0ms，表示没有延迟。NetworkClientKafka 中，Sender 线程使用的 NetworkClient 是一个核心组件，负责与 Kafka Broker 之间进行网络通信。它是 Kafka Producer 的内部类，用于管理网络连接、发送请求和接收响应。作用1：建立和管理网络连接，负责与 Kafka 集群的 Broker 建立并维护 TCP 连接。2：发送请求，Sender 线程会通过 NetworkClient 将请求（例如 ProduceRequest）发送到目标 Broker，它支持异步非阻塞的网络 I/O，提高了吞吐量。3：接收响应，NetworkClient 监听 Kafka Broker 返回的响应，例如 ProduceResponse。Sender 线程使用这些响应来判断消息是否成功写入 Broker 或是否需要重试。4：元数据管理，在发送消息之前，NetworkClient 还需要与 Kafka 集群交互，获取元数据（例如 Topic 的分区和 Leader 信息）。如果发现分区的 Leader Broker 变化，它会更新这些元数据信息。5：处理错误和重试，如果发生网络错误或 Kafka 返回异常（如分区不可用），NetworkClient 会支持重试逻辑。SelectorKafka 的 Selector 是 NetworkClient 的核心组件，负责底层的非阻塞网络通信。它通过多路复用机制，高效管理到 Kafka Broker 的网络连接，支持异步的请求发送和响应接收。这种设计使 Kafka Producer 可以以高吞吐量和低延迟处理大量的消息。ACKS0: 生产者发送过来的数据，不需要等数据落盘应答。1: 生产者发送过来的数据，Leader收到数据后应答。-1(all):生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答。-1和all等价。

TopicA-Partition0（Leaders）

main线程

LEO

Consumer

应答ACKS

Kafka Cluster

0

Consumer3

/brokers/topics/first/partitions/0/state\"leader\

Leaders

分区器 Partitioner

业务方法

ms

TopicA-Partition0（Follower）

onSuccess

发送数据

TopicB-Partition0（Leaders）

complete Fetch

Selector IO多路复用

complete Fetch Queue

Broker2

是

RecordAccumulator 缓冲区（默认32m）

ConsumerNetworkClient

Consumer拉取消息消费

Controller

Consumer1

Producer

controller\"brokerid” :0

Kafka Broker 工作流程

1

TopicB-Partition0（Follower）

request1

拦截器 Interceptor

重试

Coordinator

Fetch.min.bytes每批次最小抓取大小，默认1字节Fetch.max.bytes每批次最大抓取大小，默认50mfetch.max.wait.ms一批数据最小值未达到的超时时间，默认500ms

2

request2

Broker0

DQueue

1：设置消费者组内的消费者数量不小于分区的数量2：提高每批次拉取的数量3：如果是生产者消息大量积压，比如积压了好几天，需要采取临时队列新建一个 Topic，设置为 20 个 PartitionConsumer 不再处理业务逻辑了，只负责搬运，把消息放到临时 Topic 中这 20 个 Partition 可以有 20个 Consumer 了，它们来处理原来的业务逻辑。

HW用于标识消费者可以读取的最大消息位置，LEO用于标识消息追加到文件的最后位置。消息发送成功，不代表消费者可以消费这条消息。只有消息写入leader成功，这条消息也同步到ISR，且更新HW值，此时消费者才可以看到该消息。

Producer发送消息

NetworkClient

Broker3

Consumer Group

Leader故障处理1：Leader发生故障之后，会从ISR中选出一个新的Leader2：为保证多个副本之间的数据一致性，其余的Follower会先将各自的1og文件高于HW的部分截掉，然后从新的Leader同步数据。注意:这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。Follower故障1：Follower发生故障后会被临时踢出ISR2：这个期间Leader和Follower继续接收数据3：待该Follower恢复后，Follower会读取本地磁盘记录的上次的HW，并将log文件高于HW的部分截取掉，从HW开始向Leader进行同步。4：等该Follower的LEO大于等于该Partition的HW，即Follower追上Leader之后，就可以重新加入ISR了。

反序列化

inFlightRequests

3

Zookeeper

调用send（）

拦截器

分区分配策略以及再平衡

处理数据

到达batch.size或者linger.ms

Broker1

sender线程

4

Consumer2

成功

1.Range（默认）Range策略是kafka默认的消费者分区分配策略，它是针对topic维度的，首先对同一个topic里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。缺点：容易产生数据倾斜，如果是针对少量的topic而言C0多消费一个分区的数据影响不大，但是针对成百上千个topic那么C0就要多消费成百上千的分区数。2.RoundRobin（用的最多）RoundRobin是针对所有topic分区。它是采用轮询分区策略，是把所有的partition和所有的consumer列举出来，然后按照hashcode进行排序，最后再通过轮询算法来分配partition给每个消费者。再平衡消费者被coordinator认为是dead状态，这可能是由于消费者发送心跳超时(session.timeout.ms=45s)或者处理消息时间过长(max.poll.interval.ms=5分钟)

Kafka基本组成服务代理(节点Broker）、话题（Topic）、生产者（Producer）、消费者（Consumer）、消费者组（Consumer Group）话题（Topic）是特定类型的消息流。比如服务器的性能话题、日志话题消费者组Kafka 消费者是消费组的一部分，具有相同 group.id 的消费者实例，当多个消费者形成一个消费组来消费主题时，每个消费者会收到不同分区的消息。同一Topic的一条消息只能被同一个Consumer Group内的一个Consumer消费，但多个Consumer Group可同时消费这一消息分区(Partition)一个Partition是一个队列(先进先出)，话题可看做Partition的集合。每个Partition在物理上对应一个文件夹，该文件夹下存储这个Partition的所有消息(后缀\".log\")和索引文件(后缀 \".index\" )。创建一个topic时，同时可以指定分区数目，分区数越多，其吞吐量也越大，但是需要的资源也越多。kafka在接收到生产者发送的消息之后，会根据均衡策略将消息存储到不同的分区中。如果 Partition 规则设置的合理，所有消息可以均匀分布到不同的 Partition中。因为每条消息都被append到该Partition中，属于顺序写磁盘，因此效率非常高，顺序写磁盘效率比随机写内存要高，这是Kafka高吞吐率的一个很重要的保AR（Assigned Replication）分区中的所有副本统称为AR（Assigned Replicas）ISR（In-Sync Replicas）同步副本集合 ISR是指当前与主副本保持同步的副本集合。当主副本发生故障时，Kafka会从ISR中选举一个新的主副本来接管工作。因此，ISR的大小对于分区的可用性和性能至关重要。如果ISR太小，那么当主副本故障时，选举新的主副本可能会导致数据丢失或延迟；如果ISR太大，那么同步数据的成本会变得很高，影响分区的性能。OSR（Out-of-Sync Replicas）异步副本集合 OSR是指当前与主副本不保持同步的副本集合。这些副本可能由于网络故障或其他原因而与主副本失去同步。OSR的存在不会影响分区的可用性和性能，但是如果OSR过大，那么可能会占用过多的磁盘空间和网络带宽。HW（High Watermark）高水位 HW是指已经被所有副本复制的最高偏移量。当消费者从分区中读取消息时，它会记录当前已经读取到的偏移量，并将该偏移量作为下一次读取的起始位置。如果消费者读取到的偏移量小于HW，那么它只能读取到已经被所有副本复制的消息；如果消费者读取到的偏移量大于HW，那么它可能会读取到未被所有副本复制的消息。LEO（Log End Offset）日志末尾偏移量 LEO是指分区中最后一条消息的偏移量。当生产者向分区中写入消息时，它会将该消息的偏移量记录在LEO中。消费者从分区中读取消息时，它可以通过LEO来判断是否已经读取了所有的消息。 

消费者组初始化流程

清理

FetchedRecords从队列中抓取数据Max.poll.records一次拉取数据返回消息的最大条数，默认500条

消息积压处理

1: Broker启动后在zk中注册2: Controller谁先注册，谁说了算3: 由选举出来的Controller监听Brokers节点变化4: Controller决定Leader选举5: Controller将节点信息上传到ZK6: 其他Contorller从zk同步相关信息7: 假设Broker0中Leader挂了8: Controller监听到节点变化9: 获取ISR10: 选举新的Leader(在ISR中存活为前提，按照AR中排在前面的优先)

coordinator 辅助实现消费者组的初始化和分区的分配。 coordinator节点选择=groupid的hashcode值%50(consumer offsets的分区数量)例如:groupid的hashcode值=1，1%50=1，那么__consumer_offsets主题的1号分区，在哪个broker上，就选择这个节点的coordinator作为这个消费者组的老大。消费者组下的所有的消费者提交offset的时候就往这个分区去提交offset。1：每个consumer都发送JoinGroup请求2：选出一个consumer作为leader3：把要消费的topic情况发送给leader 消费者4：leader会负责制定消费方案5：把消费方案发给coordinator6：Coordinator就把消费方案下发给各个consumer7：每个消费者都会和coordinator保持心跳(默认3s)，一旦超时(session.timeout.ms=45s)，该消费者会被移除，并触发再平衡;或者消费者处理消息的时间过长(max.poll.interval.ms5分钟)，也会触发再平衡

生产者提高吞吐量1: batch.size:批次大小，默认16k2: linger.ms:等待时间，修改为5-100ms3: 压缩消息4: RecordAccumulator:缓冲区大小，修改为64m

否

TopicA-Partition1（Follower）

sendFetchs发送消费请求

send

TopicA-Partition1（Leaders）

logStartOffset

序列化器 Serializer

ControllerKafka Broker 集群中有一个专门的组件叫做 Controller，它是集群管理的核心角色，用于协调和管理集群的元数据、分区副本分配、以及故障恢复等。在 Kafka 集群中，只有一个 Broker 会被选举为 Controller。在 Kafka 使用 ZooKeeper 的架构中，Controller 是通过 ZooKeeper 的临时节点选举的。某个 Broker 成为 Controller 后，会在 ZooKeeper 上创建一个临时节点 /controller。如果当前 Controller 崩溃，其他 Broker 会重新选举新的 Controller。作用1：分区副本管理，负责管理每个 Topic 的分区及其副本状态以及Leader 选举2：元数据管理，维护集群元数据的全局视图，包括 Topic 的分区信息、副本分布、Leader 信息等，当集群的元数据发生变化（如创建或删除 Topic），Controller 会将更新后的元数据广播给所有 Broker 和客户端3：故障检测与恢复，如果某个 Broker 崩溃，Controller 会重新分配该 Broker 上的分区和副本。