六、常见问题QA

6.1 无消息丢失如何配置

producer

1. 调用方式

   （1）网络抖动导致消息丢失，Producer 端可以进行重试。

   （2）消息大小不合格，可以进行适当调整，符合 Broker 承受范围再发送。

   不要使用 producer.send(msg)，而要使用 producer.send(msg, callback)。记住，一定要使用带有回调通知的 send 方法。在剖析 Producer 端丢失场景的时候， 我们得出其是通过「异步」方式进行发送的，所以如果此时是使用「发后即焚」的方式发送，即调用 Producer.send(msg) 会立即返回，由于没有回调，可能因网络原因导致 Broker 并没有收到消息，此时就丢失了。

2. ACK 确认机制

   将 request.required.acks 设置为 -1/ all。acks 是 Producer 的一个参数，代表了你对“已提交”消息的定义。如果设置成 all，则表明所有副本 Broker 都要接收到消息，该消息才算是“已提交”。这是最高等级的“已提交”定义。

3. 重试次数 retries

   设置 retries 为一个较大的值。这里的 retries 同样是 Producer 的参数，对应前面提到的 Producer 自动重试。当出现网络的瞬时抖动时，消息发送可能会失败，此时配置了 retries > 0 的 Producer 能够自动重试消息发送，避免消息丢失。

4. 重试时间 retry.backoff.ms

   发送超时的时候两次发送的间隔，避免太过频繁重试，默认值为100ms, 推荐设置为300ms。

5. 启用幂等传递的方法配置

   enable.idempotence = true

consumer

1. **确保消息消费完成再提交。Consumer 端有个参数 enable.auto.commit，最好把它设置成 false，并采用手动提交位移的方式。**就像前面说的，这对于单 Consumer 多线程处理的场景而言是至关重要的。

broker配置

1. 设置 unclean.leader.election.enable = false。它控制的是哪些 Broker 有资格竞选分区的 Leader。如果一个 Broker 落后原先的 Leader 太多，那么它一旦成为新的 Leader，必然会造成消息的丢失。故一般都要将该参数设置成 false，即不允许这种情况的发生。

2. 设置 replication.factor >= 3。该参数表示分区副本的个数。建议设置 replication.factor >=3, 这样如果 Leader 副本异常 Crash 掉，Follower 副本会被选举为新的 Leader 副本继续提供服务。

3. ISR配置>1, 即设置参数 min.insync.replicas > 1。这依然是 Broker 端参数，控制的是消息至少要被写入到多少个副本才算是“已提交”。设置成大于 1 可以提升消息持久性。在实际环境中千万不要使用默认值 1。

   我们还需要确保一下 replication.factor > min.insync.replicas, 如果相等，只要有一个副本异常 Crash 掉，整个分区就无法正常工作了，因此推荐设置成： replication.factor = min.insync.replicas +1, 最大限度保证系统可用性。

4. 确保 replication.factor > min.insync.replicas。如果两者相等，那么只要有一个副本挂机，整个分区就无法正常工作了。我们不仅要改善消息的持久性，防止数据丢失，还要在不降低可用性的基础上完成。推荐设置成 replication.factor = min.insync.replicas + 1。

6.2 producer在接收到ack后会刷盘吗？有可能丢消息吗？

producer在接收到broker的ack后不会刷盘，有可能丢消息。

1. ack=all或-1，配置成所有ISR同步后才会返回ack，保证broker中有多个副本中都存有这条消息，即使leader挂了，也不会丢消息。
2. 通过配置ISR>1, 至少还有一个以上的副本保存了消息。

kafka 通过「多 Partition （分区）多 Replica（副本）机制」已经可以最大限度的保证数据不丢失，如果数据已经写入 PageCache 中但是还没来得及刷写到磁盘，此时如果所在 Broker 突然宕机挂掉或者停电，极端情况还是会造成数据丢失。

更多配置见上面。

6.3 如何保证消息的顺序存储

1. 全局有序只能让一个Topic只有一个Partition。可以通过生产者在写消息的时候指定一个key，通过partitionKey将某类消息写入同一个Partition中。
2. Kafka通过引入生产者的幂等性来解决重传机制引起的消息乱序。每一个新的生产者在初始化的时候会生成一个Producer ID，即PID。对于每一个PID，该Producer发送消息的每个<Topic,Partition> 都对应一个单调递增的Sequence Number。broker端会缓存这个序号。每收到一条消息，判断该消息的序号是否比缓存的序号大1，是则接收，不是则丢弃。

6.4 顺序消费

1. Kafka 只保证单Partition有序，而不保证主题中不同分区的顺序。每个 partition 分区按照key值排序足以满足大多数应用程序的需求。但如果你需要总记录在所有记录的上面，可使用仅有一个分区的主题来实现，这意味着每个消费者组只有一个消费者进程。
2. 使用Kafka key+offset做到业务有序，后根据key+offset存储到时序数据库或ES中。 key区分相同的业务，offset区分前后进入的顺序。使用场景比如用户画像分析，用户id做key，用户的操作先后根据offset排序，后对用户的行为进行分析。

6.5 消息优先级

实际上，这个需求也在Kafka的官方需求中(KIP-349: Priorities for Source Topics)，目前的状态是Under Vote，这个Proposal是2019年提出来的，看来官方的方案是指望不上了，只能找些第三方的轮子，或者自己来实现。

在每个Topic中，Kafka顺序写以获得尽可能获得高吞吐，使用Index文件来维护Consumer的消息拉取，维护维度是Offset。Offset不包含优先级语义，但需要顺序语义，优先级语义本身包含非顺序语义，因此就语义来看，以Offset为维度的拉模型MQ和优先级需求本质是冲突的。所以对于单个Topic，在Kafka原生实现消息优先级可行性不高。
因此很自然的，我们能够想到，可以创建多个Topic，每个Topic代表一个优先级。

在生产者端，引入优先级字段，以数字来表示，数值越高优先级越高。在向broker推消息时，根据其优先级推送到不同的topic中。
在消费者端，通过实现对不同优先级Topic的消费，以实现消息的优先消费。
对于消息的生产，实现起来比较简单，问题的难点在于消费者端如何消费不同Topic的消息，以实现高优先级的消息能够被优先处理？

这里大致有三种方案

1. 对于不同的topic，各个consumer分别拉取，拉去后在服务内部使用优先队列进行缓冲。

   为了避免OOM，需要使用有界优先队列。然而对于有界优先队列，在消息消费逻辑复杂，处理速度不够快时，可能会导致优先队列的阻塞。这里一个可能的做法是不在服务内部的优先队列中维护，而是将消息再放到Redis Zset中进行排序。这样会避免OOM和阻塞的问题，但是会增加系统的复杂度。

2. 使用一个consumer，优先拉取高优先级消息，没有的话再拉去次优先级消息。

   每次都要串行的判断各个优先级数据是否存在，实际的场景中往往是高优消息时比较少的，每次轮询到较低优先级才拉取到消息，一个consumer在性能上可能存在问题。

3. 权重消费。使用不同的consumer分别拉取各个topic，但是拉取的消息数量不同，对于高优先级的消息，拉取的“配额”更多。

   有一个开源的实现flipkart-incubator/priority-kafka-client。对于每次拉取的数量，按照优先级的“权重”不同，分配到不同的topic上。默认的分配策略是按照指数分配。

4. 桶优先级消费。开源实现：prioritization-kafka

权重消费

同时拉取多Topic，“权重”不同

对于每次拉取的数量，按照优先级的“权重”不同，分配到不同的topic上。默认的分配策略是按照指数分配。

比如对于每次拉取50个记录，3个优先级的情况下，三个优先级的比例按指数分布，为1:2:4，实际的配额即为7:14:29。
这里有一个很明显的问题是对于高优先级的数据，如果每次拉取不到指定的数量，这部分配额相当于被浪费掉了，这样会影响整体的拉取性能。

对于这种情况，代码中为每个优先级维护了一个“滑动窗口”来记录近期拉取的数量的历史记录，在拉取前，会根据历史拉取情况来进行配额的rebalance，以此来实现配额的动态分配。

桶优先级实现

自定义一个Partitioner：BucketPriorityPartitioner， 优先级高的有更多数量的Partition, 消费的时候，更多的消费者去消费。这种模式允许消费者在给定时间内消费所有优先级消息。只有给定优先级的消费者数量会发生变化。

在设定优先级时，可以设置不同的topic比例

configs.setProperty(BucketPriorityConfig.BUCKETS_CONFIG, "Platinum, Gold");
configs.setProperty(BucketPriorityConfig.ALLOCATION_CONFIG, "70%, 30%");

6.6 kafka为什么不支持主从读写

主从的模式带来的数据延迟，从节点总是会落后主节点ms级别，甚至s级别。但是在kafka除了用做削峰，异步的中间件外，它还是流式处理中间件，比如Flink,Spark，Spark的实时性要求不高，它是一批一批处理，减少批次间的间隔来完成假的实时功能；但Flink对实时性要求比较高。在实时性要求高的场景下，如果出现秒级甚至由于网络的原因，出现了分区级别的延迟，这是不能接受的。

6.7 数据积压，如何提高吞吐量

broker端

1. 动态调整Broker参数，调整Broker端各种线程池大小。增加网络线程数和I/O线程数，快速消耗一些积压。当突发流量过去后，我们也能将线程数调整回来，减少对资源的浪费。整个过程都不需要重启Broker。你甚至可以将这套调整线程数的动作，封装进定时任务中，以实现自动扩缩容。
2. 考虑增长partition数据

消费者

1. 如果是kafka的消费能力不足，则可以考虑增加topic的分区数，并且同时提升消费组的消费者数量，消费者数=分区数，两者缺一不可
2. 如果是下游的数据处理不及时，提高每批次拉取的数量。 批次拉取数量过少(拉取数据/处理时间 < 生产速度)，使处理的数据小于生产的数据，也会造成数据积压

6.8 消息默认保存多久

默认保存7天。kafka支持消息持久化，消费端为拉模型来拉取数据，消费状态和订阅关系有客户端负责维护，消息消费完后，不会立即删除，会保留历史消息。因此支持多订阅时，消息只会存储一份就可以了。

6.9 哪些情况会落后于leader？

• 副本是新加入的，直到它同步与leader一致才能加入ISR
• 副本卡死，副本机器死亡或GC卡死
• 副本同步因为网络或其它原因导致延迟

6.10 怎么判断副本落后？

replica.lag.time.max.ms参数：如果副本落后超过这个时间就判定为落后了，直到它回来。消息复制分为异步和同步，ISR是动态的，有进有出。

6.11 哪些情况consumer会发生再均衡的情形呢？

1. 同一组中消费者数量发生变化
2. 主题的分区数发生变化，加减分区
3. 组协调者节点下线

6.12 in-memory cache 和 in-process cache

in-memory cache（内存缓存）是位于应用程序和数据库之间的数据存储层，通过存储来自早期请求的数据或直接从数据库复制的数据来高速传递响应。当构建在基于磁盘的数据库上的应用程序必须在处理之前从磁盘检索数据时，内存中的缓存可以消除性能延迟。从内存读取数据比从磁盘读取数据快。内存缓存避免了延迟，提高了在线应用程序的性能。

in-process cache（进程内缓存）是在与应用程序相同的地址空间中构建的对象缓存，在使用进程内缓存时，缓存元素只存在于应用程序的单个实例中。谷歌Guava库提供了一个简单的进程内缓存API，这就是一个很好的例子。另一方面，分布式缓存位于应用程序外部，很可能部署在多个节点上，形成一个大型逻辑缓存。Memcached是一种流行的分布式缓存。来自Terracotta的Ehcache是一款可以配置为任意一种功能的产品。

6.13 一个Partition就是一个queue吗？

和java中的queue没有关系，一个Partition是由一个或多个segment文件组成。

6.14 新增一个consumer是会发生什么？

What happens when a new consumer joins the group in Kafka?

6.15 一个消费者可以订阅和消费多个topic?

消费者组中的2个或多个消费者可以订阅不同的topic。同一个topic下的某一个分区只能被某个组中的同一个消费者所消费。

6.17 什么是事件流(event streaming)?

event streaming 是一个动态的概念，它描述了一个个 event ( “something happened” in the world ) 在不同主体间连续地、正确地流动的状态(state)。event source 产生 event，event source 可以是数据库、传感器、移动设备、应用程序，等等。

event source 产生 event，event source 可以是数据库、传感器、移动设备、应用程序，等等。
event broker 持久化 event，以备 event sink 可以随时获取它们。
event sink 实时或回顾性地从 broker 中获取 event 进行处理。

producer 发布 event，broker 持久化 even，consumer 订阅 event。

事件流应用场景

我们可以在很多的应用场景中找到 event streaming 的身影，例如：
实时处理支付、金融交易、客户订单等等；
实时跟踪和监控物流进度；
持续捕获和分析来自物联网设备或其他设备的传感器数据；
不同数据源的数据连接；
作为数据平台、事件驱动架构和微服务等的技术基础；

我们来看一段英文What is Event Streaming?的关于事件流的解释

An event is defined as a change in state such as a transaction or a prospect navigating to your website. Businesses want to be able to react to these crucial business moments in real time.

Event-processing programs aggregate information from distributed systems in real time, applying rules that reveal key patterns, relationships or trends. An “event stream” is a sequence of business events ordered by time. With event stream processing you connect to all data sources and normalize, enrich and filter the data.

Event streaming platforms process the inbound data while it is in flight, as it streams through the server. It performs ultra-fast, continuous computations against high-speed streaming data, and uses a continuous query engine that drives real time alerts and actions as well as live, user-configured visualizations.

事件流是一个或多个状态的改变，它具有实时性，它的来源多种多样，可以是事务记录、IOT数据、业务和操作的metrics等，而这些数据来源，即事件具有一系列的连续性。

6.18 直接io和裸io区别

直接io是经过文件系统但不经过page cache，直接io的一个比较常见的用法就是在备份文件的时候会用直接io，这样它不会污染文件系统的cache，造成缓存命中率的下降。

裸io是绕过了文件系统直接操作磁盘，数据库实际上就是裸io ，因为它不需要经过文件系统的cache那一层，也不需要经过文件系统传递，它能够自己去保存这种读取到的内存资源。

裸io指direct io。有两种绕过文件系统的方法： 1、禁止掉文件系统的io缓存。这种做法实际上不是绕过文件系统，而是不采纳文件系统的io缓存算法。因为数据库可能自己有自己的缓存算法，如果文件系统也有缓存，就比较累赘，浪费了宝贵的内存空间，同样的内存空间给数据库扩大缓存空间更好。 2、直接裸写磁盘分区。这时不存在文件系统，也就是说磁盘分区不需要格式化。这种做法在对象存储系统中用得更多一点，在数据库中用得不多。

6.19 follower如何同步leader? 如何保证数据强一致性？

Kafka中partition replication之间同步数据，从partition的leader复制数据到follower只需要一个线程(ReplicaFetcherThread)，实际上复制是follower(一个follower相当于consumer)主动从leader批量拉取消息的。Kafka中每个Broker启动时都会创建一个副本管理服务(ReplicaManager)，该服务负责维护ReplicaFetcherThread与其他Broker链路连接关系，该Broker中存在多少Follower的partitions，就会创建相同数量的ReplicaFetcherThread线程同步对应partition数据，Kafka中partition间复制数据是由follower(扮演consumer角色)主动向leader获取消息，follower每次读取消息都会更新HW状态。

当Producer发送消息到leader partition所在Broker时，首先保证leader commit消息成功，然后创建一个“生产者延迟请求任务”，并判断当前partiton的HW是否大于等于logEndOffset，如果满足条件即表示本次Producer请求partition replicas之间数据已经一致，立即向Producer返回Ack。否则待Follower批量拉取Leader的partition消息时，同时更新Leader ISR中HW，然后检查是否满足上述条件，如果满足向Producer返回Ack。

6.20 堆外内存

kafka并没有直接使用堆外内存，但由于kafka的网络IO使用了java的nio中的DirectMemory的方式,而这个申请的是堆外内存。在对于HeapByteBuffer进行读写操作时，需要开辟堆外内存空间作为中间数据交换的空间。而这部分堆外内存并非由Kafka直接申请，而是由JVM申请。

如果在jvm参数里，-XX:MaxDirectMemorySize参数配置的过小，kafka可能会出现java.lang.OutOfMemoryError: Direct buffer memory的错误。

6.21 Kafka判断一个节点是否活着有两个条件

1. 节点必须可以维护和ZooKeeper的连接，Zookeeper通过心跳机制检查每个节点的连接。
2. 如果节点是个follower,他必须能及时的同步leader的写操作，延时不能太久。

符合以上条件的节点准确的说应该是“同步中的（in sync）”，而不是模糊的说是“活着的”或是“失败的”。Leader会追踪所有“同步中”的节点，一旦一个down掉了，或是卡住了，或是延时太久，leader就会把它移除。至于延时多久算是“太久”，是由参数replica.lag.max.messages决定的，怎样算是卡住了，怎是由参数replica.lag.time.max.ms决定的。

6.22 Kafa consumer是否可以消费指定分区消息？

在某些业务场景下，比如上游生产者希望通过分区将不同类型的业务数据发送到不同的分区，而对下游的消费者来说，就需要从指定的分区消费数据；或者在另一种业务情况下，消费者希望能够顺序消费，那么就可以通过生产端将消息发送到指定的分区下即可；

6.23 Kafka存储在硬盘上的消息格式是什么？

Segment由log、index、timeindex三个文件组成，index和timeindex分别是一些索引信息。

消息由一个固定长度的头部和可变长度的字节数组组成。头部包含了一个版本号、CRC32校验码、消息长度、具体的消息。

6.24 Kafka创建Topic时如何将分区放置到不同的Broker中

实现类AdminUtils

1. 副本因⼦不能⼤于 Broker的个数；
2. 第⼀个分区（编号为 0 ）的第⼀个副本放置位置是随机从 brokerList 选择的；
3. 其他分区的第⼀个副本放置位置相对于第 0 个分区依次往后移。也就是如果我们有 5 个Broker， 5个分区，假设第⼀个分区放在第四个 Broker 上，那么第⼆个分区将会放在第五个 Broker 上；第三个分区将会放在第⼀个 Broker 上；第四个分区将会放在第⼆个Broker 上，依次类推；
4. 剩余的副本相对于第⼀个副本放置位置其实是由 nextReplicaShift 决定的，⽽ 这个数也是随机产⽣的

6.25 Kafka新建的分区会在哪个目录下创建

配置好 log.dirs 参数，其值是 Kafka 数据的存放目录。

Kafka 会在含有分区目录最少的文件夹中创建新的分区目录，分区目录名为 Topic名+分区 ID。注意，是分区文件夹总数最少的目录，而不是磁盘使用量最少的目录！也就是说，如果你给 log.dirs 参数新增了一个新的磁盘，新的分区目录肯定是先在这个新的磁盘上创建直到这个新的磁盘目录拥有的分区目录不是最少为止。

6.26 Kafka再平衡机制

所谓的再平衡，指的是在kafka consumer所订阅的topic发生变化时发生的一种分区重分配机制。一般有三种情况会触发再平衡：

• consumer group中的新增或删除某个consumer，导致其所消费的分区需要分配到组内其他的consumer上；
• consumer订阅的topic发生变化，比如订阅的topic采用的是正则表达式的形式，如test-*此时如果有一个新建了一个topic test-user，那么这个topic的所有分区也是会自动分配给当前的consumer的，此时就会发生再平衡；
• consumer所订阅的topic发生了新增分区的行为，那么新增的分区就会分配给当前的consumer，此时就会触发再平衡。

Kafka提供的再平衡策略主要有三种：Round Robin，Range和Sticky，默认使用的是Range。这三种分配策略的主要区别在于：

• Round Robin：会采用轮询的方式将当前所有的分区依次分配给所有的consumer；

• Range：首先会计算每个consumer可以消费的分区个数，然后按照顺序将指定个数范围的分区分配给各个consumer；

• Sticky：这种分区策略是最新版本中新增的一种策略，其主要实现了两个目的：

• • 将现有的分区尽可能均衡的分配给各个consumer，存在此目的的原因在于Round Robin和Range分配策略实际上都会导致某几个consumer承载过多的分区，从而导致消费压力不均衡；
  • 如果发生再平衡，那么重新分配之后在前一点的基础上会尽力保证当前未宕机的consumer所消费的分区不会被分配给其他的consumer上；

6.27 kafka为什么比rocketmq支持的单机partion要少?

kafka是一个分区一个文件，当topic过多，分区的总量也会增加，kafka中存在过多的文件，当对消息刷盘时，就会出现文件竞争磁盘，出现性能的下降。

rocketMq中，所有的队列都存储在一个文件中，每个队列的存储的消息量也比较小，因此topic的增加对rocketMq的性能的影响较小。也从而rocketMq可以存在的topic比较多，可以适应比较复杂的业务。

6.28 partition分区数设置多少合适

没有一个统一的标准答案，可以根据当前这个topic所属业务的某个维度来，也可以按照你kafka的broker数量来决定，还可以根据kafka系统默认的分区数量来。

在回答前，先说明partition数量对业务和性能的影响：

1. 越多的partition可以提供更高的吞吐量，同时IO压力也会增加。越多的分区需要打开更多的文件句柄。
2. 建立topic时最好要确定partition的数量，虽然后期可以动态增加，但原本小的partition数改大，必定会导致宽依赖的产生，而宽依赖则一定会产生shuffle，既然产生shuffle，那数据处理效率也就必然下降。

那么可以从2方面出发：

1. 性能。如果一次有大批量的数据过来，需要在短时间内存储，那么在资源请允许的情况下，尽量设置多一些，增加吞吐量，partition数量>=consumer数量，这些也影响comsumer的消费速度；如果在资源有限，一旦单个broker的partition过多，则顺序写将退化为随机写，Page Cache脏页过多，频繁触发缺页中断，性能大幅下降。
2. 业务的角度。有多少数据源，或者按业务的维度去划分更有利于消费，比如topic省市来划分partition等

6.28 kafka删除topic

6.29 consumer 再平衡步骤

coordinator:协调者，负责消费者组内成员的leader选举、组内再平衡、组offset提交等功能。

再平衡触发三种情况

• 组成员发生变更(新consumer加入组、已有consumer主动离开组或已有consumer崩溃了）
• 订阅主题数发生变更——这当然是可能的，如果你使用了正则表达式的方式进行订阅，那么新建匹配正则表达式的topic就会触发rebalance
• 订阅主题的分区数发生变更

consumer 再平衡步骤

1. 每个consumer都发送JoinGroup请求
2. coordinator选出一个 consumer作为 leader。如果消费组内没有leader，那么第一个加入消费组的消费者就是消费者 leader，如果这个时候leader消费者退出了消费组，那么重新选举一个
3. 把要消费的 topic 情况 发送给leader 消费者
4. 消费者leader会负责制定消费方案
5. 把消费方案发给 coordinator，即消费者leader向协调者发送SyncGroup请求
6. Coordinator 就把消费方案下发给各个consumer
7. 每个消费者都会和 coordinator 保持心跳（ 默认 3s ），一旦超时 session.timeout.ms= 45s ），该消费者会被移除，并触发再平衡；或者消费者处理消息的过长（max.poll.interval.ms 5 分钟），也会触发再 平衡

分区分配策略的选择

每个消费者都可以设置自己的分区分配策略，消费组内的各个消费者会通过投票来决定

1. 在joingroup阶段，每个consumer都会把自己支持的分区分配策略发送到coordinator
2. coordinator收集到所有消费者的分配策略，组成一个候选集
3. 每个消费者需要从候选集里找出一个自己支持的策略，并且为这个策略投票
4. 最终计算候选集中各个策略的选票数，票数最多的就是当前消费组的分配策略

6.30 KafkaProducer何时创建TCP连接？

Apache Kafka的所有通信都是基于TCP的，而不是基于HTTP或其他协议。

try (Producer<String, String> producer = new KafkaProducer<>(props)) {producer.send(new ProducerRecord<String, String>(……), callback);……
}

1. 生产者应用在创建KafkaProducer实例时是会建立与Broker的TCP连接的。其实这种表述也不是很准确，应该这样说：在创建KafkaProducer实例时，生产者应用会在后台创建并启动一个名为Sender的线程，该Sender线程开始运行时首先会创建与Broker的连接。

   社区的官方文档中提及KafkaProducer类是线程安全的。KafkaProducer实例创建的线程和前面提到的Sender线程共享的可变数据结构只有RecordAccumulator类，故维护了RecordAccumulator类的线程安全，也就实现了KafkaProducer类的线程安全。了解RecordAccumulator类是做什么的，你只要知道它主要的数据结构是一个ConcurrentMap<TopicPartition, Deque>。TopicPartition是Kafka用来表示主题分区的Java对象，本身是不可变对象。而RecordAccumulator代码中用到Deque的地方都有锁的保护，所以基本上可以认定RecordAccumulator类是线程安全的。

2. TCP连接还可能在两个地方被创建：一个是在更新元数据后，另一个是在消息发送时。

   为什么说是可能？因为这两个地方并非总是创建TCP连接。当Producer更新了集群的元数据信息之后，如果发现与某些Broker当前没有连接，那么它就会创建一个TCP连接。同样地，当要发送消息时，Producer发现尚不存在与目标Broker的连接，也会创建一个。

Producer端关闭TCP连接的方式有两种：一种是用户主动关闭；一种是Kafka自动关闭。

Kafka帮你关闭，这与Producer端参数connections.max.idle.ms的值有关。默认情况下该参数值是9分钟，即如果在9分钟内没有任何请求“流过”某个TCP连接，那么Kafka会主动帮你把该TCP连接关闭。用户可以在Producer端设置connections.max.idle.ms=-1禁掉这种机制。一旦被设置成-1，TCP连接将成为永久长连接。

在第二种方式中，TCP连接是在Broker端被关闭的，但其实这个TCP连接的发起方是客户端，因此在TCP看来，这属于被动关闭的场景，即passive close。被动关闭的后果就是会产生大量的CLOSE_WAIT连接，因此Producer端或Client端没有机会显式地观测到此连接已被中断。如果设置该参数=-1，那么步骤1中创建的TCP连接将无法被关闭，从而成为“僵尸”连接

和生产者不同的是，构建KafkaConsumer实例时是不会创建任何TCP连接的

6.31 KafkaConsumer何时创建TCP连接？

1. 发起FindCoordinator请求时

   消费者程序会向集群中当前负载最小的那台Broker发送请求。就是看消费者连接的所有Broker中，谁的待发送请求最少。当然了，这种评估显然是消费者端的单向评估，并非是站在全局角度，因此有的时候也不一定是最优解。

2. 连接协调者时

3. 消费数据时

   举个例子，假设消费者要消费5个分区的数据，这5个分区各自的领导者副本分布在4台Broker上，那么该消费者在消费时会创建与这4台Broker的Socket连接。

当第三类TCP连接成功创建后，消费者程序就会废弃第一类TCP连接，之后在定期请求元数据时，它会改为使用第三类TCP连接。也就是说，最终你会发现，第一类TCP连接会在后台被默默地关闭掉。对一个运行了一段时间的消费者程序来说，只会有后面两类TCP连接存在。

6.32 comsumer自动提交位移的顺序

一旦设置了enable.auto.commit为true，Kafka会保证在开始调用poll方法时，提交上次poll返回的所有消息。从顺序上来说，poll方法的逻辑是先提交上一批消息的位移，再处理下一批消息，因此它能保证不出现消费丢失的情况。但自动提交位移的一个问题在于，它可能会出现重复消费。

6.33 KafkaConsumer是线程安全的吗

KafkaConsumer类不是线程安全的(thread-safe)。所有的网络I/O处理都是发生在用户主线程中，因此，你在使用过程中必须要确保线程安全。简单来说，就是你不能在多个线程中共享同一个KafkaConsumer实例，否则程序会抛出ConcurrentModificationException异常。

6.34 Leader总是-1，怎么破？

在生产环境中，你一定碰到过“某个主题分区不能工作了”的情形。使用命令行查看状态的话，会发现Leader是-1，于是，你使用各种命令都无济于事，最后只能用“重启大法”。不重启集群呢？

删除ZooKeeper节点/controller，触发Controller重选举。Controller重选举能够为所有主题分区重刷分区状态，可以有效解决因不一致导致的Leader不可用问题。

6.35 __consumer_offsets是做什么用的？

这是一个内部主题，公开的官网资料很少涉及到。因此，我认为，此题属于面试官炫技一类的题目。你要小心这里的考点：该主题有3个重要的知识点，你一定要全部答出来，才会显得对这块知识非常熟悉。

• 它是一个内部主题，无需手动干预，由Kafka自行管理。当然，我们可以创建该主题。
• 它的主要作用是负责注册消费者以及保存位移值。可能你对保存位移值的功能很熟悉，但其实该主题也是保存消费者元数据的地方。千万记得把这一点也回答上。另外，这里的消费者泛指消费者组和独立消费者，而不仅仅是消费者组。
• Kafka的GroupCoordinator组件提供对该主题完整的管理功能，包括该主题的创建、写入、读取和Leader维护等。

6.36 分区Leader选举策略有几种？

分区的Leader副本选举对用户是完全透明的，它是由Controller独立完成的。你需要回答的是，在哪些场景下，需要执行分区Leader选举。每一种场景对应于一种选举策略。当前，Kafka有4种分区Leader选举策略。

• OfflinePartition Leader选举：每当有分区上线时，就需要执行Leader选举。所谓的分区上线，可能是创建了新分区，也可能是之前的下线分区重新上线。这是最常见的分区Leader选举场景。
• ReassignPartition Leader选举：当你手动运行kafka-reassign-partitions命令，或者是调用Admin的alterPartitionReassignments方法执行分区副本重分配时，可能触发此类选举。假设原来的AR是[1，2，3]，Leader是1，当执行副本重分配后，副本集合AR被设置成[4，5，6]，显然，Leader必须要变更，此时会发生Reassign Partition Leader选举。
• PreferredReplicaPartition Leader选举：当你手动运行kafka-preferred-replica-election命令，或自动触发了Preferred Leader选举时，该类策略被激活。所谓的Preferred Leader，指的是AR中的第一个副本。比如AR是[3，2，1]，那么，Preferred Leader就是3。
• ControlledShutdownPartition Leader选举：当Broker正常关闭时，该Broker上的所有Leader副本都会下线，因此，需要为受影响的分区执行相应的Leader选举。

这4类选举策略的大致思想是类似的，即从AR中挑选首个在ISR中的副本，作为新Leader。当然，个别策略有些微小差异。不过，回答到这种程度，应该足以应付面试官了。毕竟，微小差别对选举Leader这件事的影响很小。

6.37 Kafka的哪些场景中使用了零拷贝（Zero Copy）？

Zero Copy是特别容易被问到的高阶题目。在Kafka中，体现Zero Copy使用场景的地方有两处：基于mmap的索引和日志文件读写所用的TransportLayer。

先说第一个。索引都是基于MappedByteBuffer的，也就是让用户态和内核态共享内核态的数据缓冲区，此时，数据不需要复制到用户态空间。不过，mmap虽然避免了不必要的拷贝，但不一定就能保证很高的性能。在不同的操作系统下，mmap的创建和销毁成本可能是不一样的。很高的创建和销毁开销会抵消Zero Copy带来的性能优势。由于这种不确定性，在Kafka中，只有索引应用了mmap，最核心的日志并未使用mmap机制。

再说第二个。TransportLayer是Kafka传输层的接口。它的某个实现类使用了FileChannel的transferTo方法。该方法底层使用sendfile实现了Zero Copy。对Kafka而言，如果I/O通道使用普通的PLAINTEXT，那么，Kafka就可以利用Zero Copy特性，直接将页缓存中的数据发送到网卡的Buffer中，避免中间的多次拷贝。相反，如果I/O通道启用了SSL，那么，Kafka便无法利用Zero Copy特性了。

6.38 如何调优Kafka？

回答任何调优问题的第一步，就是确定优化目标，并且定量给出目标！这点特别重要。对于Kafka而言，常见的优化目标是吞吐量、延时、持久性和可用性。每一个方向的优化思路都是不同的，甚至是相反的。

确定了目标之后，还要明确优化的维度。有些调优属于通用的优化思路，比如对操作系统、JVM等的优化；有些则是有针对性的，比如要优化Kafka的TPS。我们需要从3个方向去考虑。

• Producer端：增加batch.size、linger.ms，启用压缩，关闭重试等。
• Broker端：增加num.replica.fetchers，提升Follower同步TPS，避免Broker Full GC等。
• Consumer：增加fetch.min.bytes等

6.39 Controller发生网络分区（Network Partitioning）时，Kafka会怎么样？

这道题目能够诱发我们对分布式系统设计、CAP理论、一致性等多方面的思考。不过，针对故障定位和分析的这类问题，我建议你首先言明“实用至上”的观点，即不论怎么进行理论分析，永远都要以实际结果为准。一旦发生Controller网络分区，那么，第一要务就是查看集群是否出现“脑裂”，即同时出现两个甚至是多个Controller组件。这可以根据Broker端监控指标ActiveControllerCount来判断。

现在，我们分析下，一旦出现这种情况，Kafka会怎么样。

由于Controller会给Broker发送3类请求，即LeaderAndIsrRequest、StopReplicaRequest和UpdateMetadataRequest，因此，一旦出现网络分区，这些请求将不能顺利到达Broker端。这将影响主题的创建、修改、删除操作的信息同步，表现为集群仿佛僵住了一样，无法感知到后面的所有操作。因此，网络分区通常都是非常严重的问题，要赶快修复。

6.40 Java Consumer为什么采用单线程来获取消息？

在回答之前，如果先把这句话说出来，一定会加分：Java Consumer是双线程的设计。一个线程是用户主线程，负责获取消息；另一个线程是心跳线程，负责向Kafka汇报消费者存活情况。将心跳单独放入专属的线程，能够有效地规避因消息处理速度慢而被视为下线的“假死”情况。

单线程获取消息的设计能够避免阻塞式的消息获取方式。单线程轮询方式容易实现异步非阻塞式，这样便于将消费者扩展成支持实时流处理的操作算子。因为很多实时流处理操作算子都不能是阻塞式的。另外一个可能的好处是，可以简化代码的开发。多线程交互的代码是非常容易出错的。

6.41 简述Follower副本消息同步的完整流程

首先，Follower发送FETCH请求给Leader。接着，Leader会读取底层日志文件中的消息数据，再更新它内存中的Follower副本的LEO值，更新为FETCH请求中的fetchOffset值。最后，尝试更新分区高水位值。Follower接收到FETCH响应之后，会把消息写入到底层日志，接着更新LEO和HW值。

Leader和Follower的HW值更新时机是不同的，Follower的HW更新永远落后于Leader的HW。这种时间上的错配是造成各种不一致的原因。

七、参考

Kafka：这次分享我只想把原理讲清楚
kafka中文网
关于OS Page Cache的简单介绍
《图解系统》笔记（一）
zookeeper和Kafka的关系
Kafka源码深度解析－序列4 －Producer －network层核心原理
Reactor模式介绍
Reactor模型
图解Kafka服务端网络模型]
Kafka 核心技术与实战
Kafka 核心源码解读

转载请注明：arthur.dy.lee_kafka常见问题QA（六）