11.LEO、LSO、AR、ISR、HW 都表示什么含义？
讲真，我不认为这是炫技的题目，特别是作为 SRE 来讲，对于一个开源软件的原理以及概念的理解，是非常重要的。
- LEO(Log End Offset): 日志末端位移值或末端偏移量，表示日志下一条待插入消息的位移值。举个例子，如果日志有 10 条消息，位移值从 0 开始，那么，第 10 条消息的位移值就是 9。此时，LEO = 10。
- LSO(Log Stable Offset): 这是 Kafka 事务的概念。如果你没有使用到事务，那么这个值不存在（其实也不是不存在，只是设置成一个无意义的值）。该值控制了事务型消费者能够看到的消息范围。它经常与 Log Start Offset，即日志起始位移值相混淆，因为有些人将后者缩写成 LSO，这是不对的。在 Kafka 中，LSO 就是指代 Log Stable Offset。
- AR(Assigned Replicas): AR 是主题被创建后，分区创建时被分配的副本集合，副本个数由副本因子决定。
- ISR(In-Sync Replicas): Kafka 中特别重要的概念，指代的是 AR 中那些与 Leader 保持同步的副本集合。在 AR 中的副本可能不在 ISR 中，但 Leader 副本天然就包含在 ISR 中。
- HW(High watermark): 高水位值，这是控制消费者可读取消息范围的重要字段。一个普通消费者只能 “看到” Leader 副本上介于 Log Start Offset 和 HW（不含）之间的所有消息。水位以上的消息是对消费者不可见的。
需要注意的是，通常在 ISR 中，可能会有人问到为什么有时候副本不在 ISR 中，这其实也就是上面说的 Leader 和 Follower 不同步的情况，为什么我们前面说，短暂的不同步我们可以关注，但是长时间的不同步，我们需要介入排查了，因为 ISR 里的副本后面都是通过 replica.lag.time.max.ms，即 Follower 副本的 LEO 落后 Leader LEO 的时间是否超过阈值来决定副本是否在 ISR 内部的。
12.Kafka 能手动删除消息吗？
Kafka 不需要用户手动删除消息。它本身提供了留存策略，能够自动删除过期消息。当然，它是支持手动删除消息的。
- 对于设置了 Key 且参数
cleanup.policy=compact的主题而言，我们可以构造一条 的消息发送给 Broker，依靠 Log Cleaner 组件提供的功能删除掉该 Key 的消息。 - 对于普通主题而言，我们可以使用 kafka-delete-records 命令，或编写程序调用 Admin.deleteRecords 方法来删除消息。这两种方法殊途同归，底层都是调用 Admin 的 deleteRecords 方法，通过将分区 Log Start Offset 值抬高的方式间接删除消息。
13.__consumer_offsets 是做什么用的？
这是一个内部主题，主要用于存储消费者的偏移量，以及消费者的元数据信息 (消费者实例，消费者id 等等)
需要注意的是: Kafka 的 GroupCoordinator 组件提供对该主题完整的管理功能，包括该主题的创建、写入、读取和 Leader 维护等。
14. 分区 Leader 选举策略有几种？
分区的 Leader 副本选举对用户是完全透明的，它是由 Controller 独立完成的。你需要回答的是，在哪些场景下，需要执行分区 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。
15.Kafka 的哪些场景中使用了零拷贝（Zero Copy）
其实这道题对于 SRE 来讲，有点超纲了，不过既然 Zero Copy 是 kafka 高性能的保证，我们需要了解它。
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 特性了。
深度思考题
16.Kafka 为什么不支持读写分离？
这其实是分布式场景下的通用问题，因为我们知道 CAP 理论下，我们只能保证 C (可用性) 和 A (一致性) 取其一，如果支持读写分离，那其实对于一致性的要求可能就会有一定折扣，因为通常的场景下，副本之间都是通过同步来实现副本数据一致的，那同步过程中肯定会有时间的消耗，如果支持了读写分离，就意味着可能的数据不一致，或数据滞后。
Leader/Follower 模型并没有规定 Follower 副本不可以对外提供读服务。很多框架都是允许这么做的，只是 Kafka 最初为了避免不一致性的问题，而采用了让 Leader 统一提供服务的方式。
不过，自 Kafka 2.4 之后，Kafka 提供了有限度的读写分离，也就是说，Follower 副本能够对外提供读服务。
19.Java Consumer 为什么采用单线程来获取消息？
在回答之前，如果先把这句话说出来，一定会加分：Java Consumer 是双线程的设计。一个线程是用户主线程，负责获取消息；另一个线程是心跳线程，负责向 Kafka 汇报消费者存活情况。将心跳单独放入专属的线程，能够有效地规避因消息处理速度慢而被视为下线的 “假死” 情况。
单线程获取消息的设计能够避免阻塞式的消息获取方式。单线程轮询方式容易实现异步非阻塞式，这样便于将消费者扩展成支持实时流处理的操作算子。因为很多实时流处理操作算子都不能是阻塞式的。另外一个可能的好处是，可以简化代码的开发。多线程交互的代码是非常容易出错的。
20. 简述 Follower 副本消息同步的完整流程
首先，Follower 发送 FETCH 请求给 Leader。
接着，Leader 会读取底层日志文件中的消息数据，再更新它内存中的 Follower 副本的 LEO 值，更新为 FETCH 请求中的 fetchOffset 值。
最后，尝试更新分区高水位值。Follower 接收到 FETCH 响应之后，会把消息写入到底层日志，接着更新 LEO 和 HW 值。
Leader 和 Follower 的 HW 值更新时机是不同的，Follower 的 HW 更新永远落后于 Leader 的 HW。这种时间上的错配是造成各种不一致的原因。
因此，对于消费者而言，消费到的消息永远是所有副本中最小的那个 HW。
Kafka的分区数是不是越多越好？
分区多的优点
kafka使用分区将topic的消息打散到多个分区分布保存在不同的broker上，实现了producer和consumer消息处理的高吞吐量。Kafka的producer和consumer都可以多线程地并行操作，而每个线程处理的是一个分区的数据。因此分区实际上是调优Kafka并行度的最小单元。对于producer而言，它实际上是用多个线程并发地向不同分区所在的broker发起Socket连接同时给这些分区发送消息；而consumer，同一个消费组内的所有consumer线程都被指定topic的某一个分区进行消费。
所以说，如果一个topic分区越多，理论上整个集群所能达到的吞吐量就越大。
分区不是越多越好
分区是否越多越好呢？显然也不是，因为每个分区都有自己的开销：
一、客户端/服务器端需要使用的内存就越多 Kafka0.8.2之后，在客户端producer有个参数batch.size，默认是16KB。它会为每个分区缓存消息，一旦满了就打包将消息批量发出。看上去这是个能够提升性能的设计。不过很显然，因为这个参数是分区级别的，如果分区数越多，这部分缓存所需的内存占用也会更多。假设你有10000个分区，按照默认设置，这部分缓存需要占用约157MB的内存。而consumer端呢？我们抛开获取数据所需的内存不说，只说线程的开销。如果还是假设有10000个分区，同时consumer线程数要匹配分区数(大部分情况下是最佳的消费吞吐量配置)的话，那么在consumer client就要创建10000个线程，也需要创建大约10000个Socket去获取分区数据。这里面的线程切换的开销本身已经不容小觑了。
服务器端的开销也不小，如果阅读Kafka源码的话可以发现，服务器端的很多组件都在内存中维护了分区级别的缓存，比如controller，FetcherManager等，因此分区数越多，这种缓存的成本就越大。
二、文件句柄的开销 每个分区在底层文件系统都有属于自己的一个目录。该目录下通常会有两个文件： base_offset.log和base_offset.index。Kafak的controller和ReplicaManager会为每个broker都保存这两个文件句柄(file handler)。很明显，如果分区数越多，所需要保持打开状态的文件句柄数也就越多，最终可能会突破你的ulimit -n的限制。
三、降低高可用性 Kafka通过副本(replica)机制来保证高可用。具体做法就是为每个分区保存若干个副本(replica_factor指定副本数)。每个副本保存在不同的broker上。期中的一个副本充当leader 副本，负责处理producer和consumer请求。其他副本充当follower角色，由Kafka controller负责保证与leader的同步。如果leader所在的broker挂掉了，contorller会检测到然后在zookeeper的帮助下重选出新的leader——这中间会有短暂的不可用时间窗口，虽然大部分情况下可能只是几毫秒级别。但如果你有10000个分区，10个broker，也就是说平均每个broker上有1000个分区。此时这个broker挂掉了，那么zookeeper和controller需要立即对这1000个分区进行leader选举。比起很少的分区leader选举而言，这必然要花更长的时间，并且通常不是线性累加的。如果这个broker还同时是controller情况就更糟了。
如何确定分区数量呢
可以遵循一定的步骤来尝试确定分区数：创建一个只有1个分区的topic，然后测试这个topic的producer吞吐量和consumer吞吐量。假设它们的值分别是Tp和Tc，单位可以是MB/s。然后假设总的目标吞吐量是Tt，那么分区数 = Tt / max(Tp, Tc)
说明：Tp表示producer的吞吐量。测试producer通常是很容易的，因为它的逻辑非常简单，就是直接发送消息到Kafka就好了。Tc表示consumer的吞吐量。测试Tc通常与应用的关系更大， 因为Tc的值取决于你拿到消息之后执行什么操作，因此Tc的测试通常也要麻烦一些。
一条消息如何知道要被发送到哪个分区？
按照key值分配
默认情况下，Kafka根据传递消息的key来进行分区的分配，即hash(key) % numPartitions：
def partition(key: Any, numPartitions: Int): Int = {
Utils.abs(key.hashCode) % numPartitions
}
这保证了相同key的消息一定会被路由到相同的分区。
key为null时，从缓存中取分区id或者随机取一个
如果你没有指定key，那么Kafka是如何确定这条消息去往哪个分区的呢？
if(key == null) { // 如果没有指定key
val id = sendPartitionPerTopicCache.get(topic) // 先看看Kafka有没有缓存的现成的分区Id
id match {
case Some(partitionId) =>
partitionId // 如果有的话直接使用这个分区Id就好了
case None => // 如果没有的话，
val availablePartitions = topicPartitionList.filter(_.leaderBrokerIdOpt.isDefined) //找出所有可用分区的leader所在的broker
if (availablePartitions.isEmpty)
throw new LeaderNotAvailableException("No leader for any partition in topic " + topic)
val index = Utils.abs(Random.nextInt) % availablePartitions.size // 从中随机挑一个
val partitionId = availablePartitions(index).partitionId
sendPartitionPerTopicCache.put(topic, partitionId) // 更新缓存以备下一次直接使用
partitionId
}
}
不指定key时，Kafka几乎就是随机找一个分区发送无key的消息，然后把这个分区号加入到缓存中以备后面直接使用——当然了，Kafka本身也会清空该缓存（默认每10分钟或每次请求topic元数据时）。
Consumer个数与分区数有什么关系？
topic下的一个分区只能被同一个consumer group下的一个consumer线程来消费，但反之并不成立，即一个consumer线程可以消费多个分区的数据，比如Kafka提供的ConsoleConsumer，默认就只是一个线程来消费所有分区的数据。
即分区数决定了同组消费者个数的上限
image.png
所以，如果你的分区数是N，那么最好线程数也保持为N，这样通常能够达到最大的吞吐量。超过N的配置只是浪费系统资源，因为多出的线程不会被分配到任何分区。
Consumer消费Partition的分配策略
Kafka提供的两种分配策略： range和roundrobin，由参数partition.assignment.strategy指定，默认是range策略。
当以下事件发生时，Kafka 将会进行一次分区分配：
- 同一个 Consumer Group 内新增消费者
- 消费者离开当前所属的Consumer Group，包括shuts down 或 crashes
- 订阅的主题新增分区
将分区的所有权从一个消费者移到另一个消费者称为重新平衡（rebalance），如何rebalance就涉及到本文提到的分区分配策略。
下面我们将详细介绍 Kafka 内置的两种分区分配策略。本文假设我们有个名为 T1 的主题，其包含了10个分区，然后我们有两个消费者（C1，C2）
来消费这10个分区里面的数据，而且 C1 的 num.streams = 1，C2 的 num.streams = 2。
Range strategy
Range策略是对每个主题而言的，首先对同一个主题里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。在我们的例子里面，排完序的分区将会是0, 1, 2, 3, 4, 5, 6, 7, 8, 9；消费者线程排完序将会是C1-0, C2-0, C2-1。然后将partitions的个数除于消费者线程的总数来决定每个消费者线程消费几个分区。如果除不尽，那么前面几个消费者线程将会多消费一个分区。在我们的例子里面，我们有10个分区，3个消费者线程， 10 / 3 = 3，而且除不尽，那么消费者线程 C1-0 将会多消费一个分区，所以最后分区分配的结果看起来是这样的：
- C1-0 将消费 0, 1, 2, 3 分区
- C2-0 将消费 4, 5, 6 分区
- C2-1 将消费 7, 8, 9 分区
假如我们有11个分区，那么最后分区分配的结果看起来是这样的：
- C1-0 将消费 0, 1, 2, 3 分区
- C2-0 将消费 4, 5, 6, 7 分区
- C2-1 将消费 8, 9, 10 分区
假如我们有2个主题(T1和T2)，分别有10个分区，那么最后分区分配的结果看起来是这样的：
- C1-0 将消费 T1主题的 0, 1, 2, 3 分区以及 T2主题的 0, 1, 2, 3分区
- C2-0 将消费 T1主题的 4, 5, 6 分区以及 T2主题的 4, 5, 6分区
- C2-1 将消费 T1主题的 7, 8, 9 分区以及 T2主题的 7, 8, 9分区
可以看出，C1-0 消费者线程比其他消费者线程多消费了2个分区，这就是Range strategy的一个很明显的弊端。
RoundRobin strategy
使用RoundRobin策略有两个前提条件必须满足：
- 同一个Consumer Group里面的所有消费者的num.streams必须相等；
- 每个消费者订阅的主题必须相同。
所以这里假设前面提到的2个消费者的num.streams = 2。RoundRobin策略的工作原理：将所有主题的分区组成 TopicAndPartition 列表，然后对 TopicAndPartition 列表按照 hashCode 进行排序，看下面的代码应该会明白：
val allTopicPartitions = ctx.partitionsForTopic.flatMap { case(topic, partitions) =>
info("Consumer %s rebalancing the following partitions for topic %s: %s"
.format(ctx.consumerId, topic, partitions))
partitions.map(partition => {
TopicAndPartition(topic, partition)
})
}.toSeq.sortWith((topicPartition1, topicPartition2) => {
/*
* Randomize the order by taking the hashcode to reduce the likelihood of all partitions of a given topic ending
* up on one consumer (if it has a high enough stream count).
*/
topicPartition1.toString.hashCode < topicPartition2.toString.hashCode
})
最后按照round-robin风格将分区分别分配给不同的消费者线程。
在这个的例子里面，假如按照 hashCode 排序完的topic-partitions组依次为T1-5, T1-3, T1-0, T1-8, T1-2, T1-1, T1-4, T1-7, T1-6, T1-9，我们的消费者线程排序为C1-0, C1-1, C2-0, C2-1，最后分区分配的结果为：
- C1-0 将消费 T1-5, T1-2, T1-6 分区；
- C1-1 将消费 T1-3, T1-1, T1-9 分区；
- C2-0 将消费 T1-0, T1-4 分区；
- C2-1 将消费 T1-8, T1-7 分区；
多个主题的分区分配和单个主题类似。遗憾的是，目前我们还不能自定义分区分配策略，只能通过partition.assignment.strategy参数选择 range 或 roundrobin。
Kafka 的零拷贝技术
1.传统拷贝过程
? ? ? ?首先我们先来了解一下传统的拷贝流程。当消息从发送到写入磁盘，Broker 维护的消息日志本身就是文件目录形式，每个文件都是二进制保存，生产者和消费者使用相同的格式来处理。在消费者获取消息时，服务器先从磁盘读取数据到内存，然后把内存中的数据原封不动的通过 socket 的形式发送给消费者。虽然这个操作看起来简单，但是实际上中间经历了很多步骤。如下图所示：
? ? ? ?这个过程涉及到 4 次上下文切换以及 4 次数据的复制，并且有两次复制操作是由 CPU 完成。但是这个过程中，数据完全没有进行变化，仅仅是从磁盘复制到网卡缓冲区。?
? ? ? ?在这种情况下，如果能够减少用户空间与内核空间之间的切换，是不是会比传统拷贝快一点呢？如下图：
结果显而易见，毕竟少了 1 次传输过程，肯定会比传统的拷贝性能高。这样子首先数据被从磁盘读取到 Read Buffer 中，然后再发送到 Socket Buffer，最后才发送到网卡。虽然减少了用户空间和内核空间之间的数据交换，但依然存在多次数据复制。
? ? ? ?明显性能的开销，都消耗在彼此之间的数据复制过程中，那么进一步减少数据的复制过程，或者干脆没有数据复制这一过程，性能会明显增强。这里就需要介绍到 DMA 技术 了
2.DMA 技术的出现
? ? ? ?DMA(Direct Memory Access，直接内存存取) 是所有现代电脑的重要特色，它的出现就是为了解决批量数据的输入/输出问题。它允许不同速度的硬件装置来沟通，而不需要依赖于 CPU 的大量中断负载。否则，CPU 需要从来源把每一片段的资料复制到暂存器，然后把它们再次写回到新的地方。在这个时间中，CPU 对于其他的工作来说就无法使用。
? ? ? ?传统的内存访问，所有的请求都会发送到 CPU ，然后再由 CPU 来完成相关调度工作。如下图所示：
? ? ? ? 当?DMA 技术的出现，数据文件在各个层之间的传输，则可以直接绕过CPU，使得外围设备可以通过DMA控制器直接访问内存。与此同时，CPU可以继续执行程序。如下图：
?在现代电脑中，很多硬件都是支持 DMA 技术的，这里面其中就包括我们此处用到的网卡。还有其他硬件也都是支持 DMA 技术的，例如：磁盘、显卡、声卡等其他硬件。
3.零拷贝技术
? ? ? ?有了 DMA 技术的，通过网卡直接去访问系统的内存，就可以实现现绝对的零拷贝了。这样就可以最大程度提高传输性能。通过“零拷贝”技术，我们可以去掉那些没必要的数据复制操作， 同时也会减少上下文切换次数。
? ? ? ?现代的 Unix 操作系统提供 了一个优化的代码路径，用于将数据从页缓存直接传输到 Socket； 在 Linux 中，是通过 sendfile 系统调用来完成的。Java 提供了访问这个系统调用的方法：FileChannel.transferTo API 。使用 sendfile ，只需要一次拷贝就行，允许操作系统将数据直接从页缓存发送到网络上。所以在这个优化的路径中， 只有最后一步将数据拷贝到网卡缓存中是需要的。
4.Java零拷贝的实现
File file = new File("demo.zip");
RandomAccessFile raf = new RandomAccessFile(file, "rw");
FileChannel fileChannel = raf.getChannel();
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("", 1234));
// 直接使用了transferTo()进行通道间的数据传输
fileChannel.transferTo(0, fileChannel.size(), socketChannel);
? ? ? ?Java的零拷贝由 FileChannel.transferTo() 方法实现。transferTo() 方法将数据从 FileChannel 对象传送到可写的字节通道（如Socket Channel等）。在内部实现中，由 native 方法 transferTo() 来实现，它依赖底层操作系统的支持。在 UNIX 和 Linux系统中，调用这个方法将会引起 sendfile() 系统调用。
? ? 零拷贝的使用场景一般是：
较大，读写较慢，追求速度
内存不足，不能加载太大数据
带宽不够，即存在其他程序或线程存在大量的IO操作，导致带宽不够
?
partition 的数据文件
partition 中的每条 Message 包含三个属性：offset，MessageSize，data，其中 offset 表 示 Message 在这个 partition 中的偏移量，offset 不是该 Message 在 partition 数据文件中的实际存储位置，而是逻辑上一个值，它唯一确定了 partition 中的一条 Message，可以认为 offset 是 partition 中 Message 的 id；MessageSize 表示消息内容 data 的大小；data 为 Message 的具体内容。
数据文件分段 segment
partition 物理上由多个 segment 文件组成，每个 segment 大小相等，顺序读写。每个 segment数据文件以该段中最小的 offset 命名，文件扩展名为.log。这样在查找指定 offset 的 Message 的时候，用二分查找就可以定位到该 Message 在哪个 segment 数据文件中。
数据文件索引
Kafka 为每个分段后的数据文件建立了索引文件，文件名与数据文件的名字是一样的，只是文件扩展名为.index。index 文件中并没有为数据文件中的每条 Message 建立索引，而是采用了稀疏存储的方式，每隔一定字节的数据建立一条索引。这样避免了索引文件占用过多的空间，从而可以将索引文件保留在内存中。
?