网易一面：单节点2000Wtps，Kafka怎么做的？

Dear 丶 2024-03-24 20:53 53阅读 0赞

### 说在前面 ###

在40岁老架构师 尼恩的**读者交流群**(50+)中，最近有小伙伴拿到了一线互联网企业如网易、有赞、希音、百度、网易、滴滴的面试资格，遇到一几个很重要的面试题：

> 问题1：单节点2000Wtps，Kafka高性能原理是什么？
> 
> 问题2：做过Kafka 进行性能压测吗？单个节点的极限处理能力是多少？是怎么做到的？

注意，单个节点的极限处理能力接近每秒 2000万 条消息，吞吐量达到每秒 600MB

那 Kafka 为什么这么快？如何做到这个高的性能？

**尼恩提示，Kafka相关的问题，是开发的核心知识，又是线上的重点难题。**

所以，这里尼恩给大家做一下系统化、体系化的线程池梳理，使得大家可以充分展示一下大家雄厚的 “技术肌肉”，**让面试官爱到 “不能自已、口水直流”**。

也一并把这个题目以及参考答案，收入咱们的 《[尼恩Java面试宝典][Java]》V100版本，供后面的小伙伴参考，提升大家的 3高 架构、设计、开发水平。

> 最新《尼恩 架构笔记》《尼恩高并发三部曲》《[尼恩Java面试宝典][Java]》的PDF，请到公号【技术自由圈】获取

**这里，主要从这 3 个角度来分析**：

*  **生产端**
 *  **服务端 Broker**
 *  **消费端**

先来看下生产端发送消息，Kafka 做了哪些优化？

#### 文章目录 ####

*   *  说在前面
     *  一、生产端 Producer
     *   *  生产者发送消息流程
         *  前置知识：队列缓存+批量写入架构
         *  生产者发送消息流程源码
         *   *  1. 将消息封装成ProducerRecord对象：
             *  2. Serializer对消息的key和value做序列化：
             *  3. 经过序列化后，会根据自定义的分区器或者Kafka默认的分区器进行获取消息的所属的分区。
             *  4. 获取到消息所属的分区后，消息会被存放到消息缓冲区中（RecordAccumulator）中，
             *  5. 生产者会有一个send线程，用于不断的获取消息和发送消息。
             *  6. 每个请求都会缓存在一个inFlightRequest缓冲区内，里面为每一个Broker分配了一个队列。
             *  7. selector获取到Request会发往相对应的Broker节点。Broker节点收到Request后会进行ACK确认这个Request
             *  8. 当收到Broker对某个Request的ACK后，会删除inFlightRequest队列中这个Request。然后调用clear方法清除对应的ProducerBatch。
             *  9. 如果发送过程中产生了异常，消息发送会存在重试机制。条件为重试次数小于指定值&&异常为RetriableException
         *  生产端的高并发核心架构设计
     *  二、服务端 Broker
     *   *  1、PageCache 加速消息读写
         *  2、Kafka 的文件布局 以及 磁盘文件顺序写入
         *  3、零拷贝 sendfile：加速消费流程
     *  三、消费端 Consumer
     *  说在最后
     *  参考文献
     *  推荐阅读

### 一、生产端 Producer ###

![b8bc45927c9840adb23f2699b052b5a8.png][]

**先来回顾下 Producer 生产者发送消息的流程**：

Kafka的源码最核心的是由client模块和core模块构成，用一幅图大致介绍一下生产者发送消息的流程。

![d84a046ee770469a9d2f1c1f2a924821.png][]

#### 生产者发送消息流程 ####

1.  将消息封装成ProducerRecord对象
2.  Serializer对消息的key和value做序列化
3.  根据Partitioner将消息分发到不同的分区，需要先获取集群的元数据
4.  RecordAccumulator封装很多分区的消息队列，每个队列代表一个分区，每个分区里面有很多的批次，每个批次里面由多条消息组成
5.  Sender会从RecordAccumulator拉取消息，封装成批次，发送请求
6.  通过网络将请求发送到kafka集群

#### 前置知识：队列缓存+批量写入架构 ####

队列缓存+批量写入架构，是尼恩反复强调的一个高并发写入架构，

kafka、netty都用到这个架构

kafka的生产者，也用了这个架构。 设计了一个核心组件RecordAccumulator

1.  RecordAccumulator：每一个是生产上都会维护一个固定大小的内存空间，主要用于合并单条消息，进行批量发送，提高吞吐量，减少带宽消耗。
2.  RecordAccumulator的大小是可配置的，可以配置buffer.memory来修改缓冲区大小，默认值为：33554432（32M）
3.  RecordAccumulator内存结构分为两部分

*  第一部分为已经使用的内存，这一部分主要存放了很多的队列。每一个主题的每一个分区都会创建一个队列，来存放当前分区下待发送的消息集合。
 *  第二部分为未使用的内存，这一部分分为已经池化后的内存和未池化的整个剩余内存（nonPooledAvailableMemory）。池化的内存的会根据batch.size（默认值为16K）的配置进行池化多个ByteBuffer，放入一个队列中。所有的剩余空间会形成一个未池化的剩余空间。

#### 生产者发送消息流程源码 ####

##### 1. 将消息封装成ProducerRecord对象： #####

生产者生成某个消息后，ProducerRecord首先会经过一个或多个组成的拦截器链。

##### 2. Serializer对消息的key和value做序列化： #####

当消息通过所有的拦截器之后，会进行序列化，会根据key和value的序列化配置进行序列化消息内容，生产者和消费者必须使用相同的key-value序列化方式。

// 消息key序列化
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // 消息value序列化
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

##### 3. 经过序列化后，会根据自定义的分区器或者Kafka默认的分区器进行获取消息的所属的分区。 #####

自定义分区器可以参考下面。

Kafka默认的分区器规则：

*  1）当消息的key存在时，首先获取当前topic下的所有分区数，然后对key进行求hash值，根据hash值和分区总数进行取余，获取所属的的分区。
 *  2）如果key不存在时，会根据topic获取一个递增的数值，然后通过和分区数进行取余，获取所属的分区。

Kafka默认分区器源码：

public class DefaultPartitioner implements Partitioner {
        
    
        private final ConcurrentMap<String, AtomicInteger> topicCounterMap = new ConcurrentHashMap<>();
    
        public void configure(Map<String, ?> configs) {
        }
    
        public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        
            List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
            int numPartitions = partitions.size();
            if (keyBytes == null) {
        
                int nextValue = nextValue(topic);
                List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
                if (availablePartitions.size() > 0) {
        
                    int part = Utils.toPositive(nextValue) % availablePartitions.size();
                    return availablePartitions.get(part).partition();
                } else {
        
                    // no partitions are available, give a non-available partition
                    return Utils.toPositive(nextValue) % numPartitions;
                }
            } else {
        
                // hash the keyBytes to choose a partition
                return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
            }
        }
    
        private int nextValue(String topic) {
        
            AtomicInteger counter = topicCounterMap.get(topic);
            if (null == counter) {
        
                counter = new AtomicInteger(ThreadLocalRandom.current().nextInt());
                AtomicInteger currentCounter = topicCounterMap.putIfAbsent(topic, counter);
                if (currentCounter != null) {
        
                    counter = currentCounter;
                }
            }
            return counter.getAndIncrement();
        }
    
        public void close() {
        }
    
    }

自定义分区器：

public class CustomerPartitions implements Partitioner{
        
        @Override
        public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        
            int partition = 0;
            if(key == null) {
        
            } else {
        
                String keyStr = key.toString();
                if(keyStr.contains("Test")) {
        
                    partition = 1;
                } else {
        
                    partition = 2;
                }
            }
            return partition;
        }
    
        @Override
        public void close() {
        
    
        }
    
        @Override
        public void configure(Map<String, ?> configs) {
        
    
        }
    }

##### 4. 获取到消息所属的分区后，消息会被存放到消息缓冲区中（RecordAccumulator）中， #####

根据topic和分区可以确定一个双端队列(Deque)中，这个队列中每个节点为多个消息的合集（ProducerBatch）,新的消息会被放到队列的最后一个节点上，存放会存在多种情况。

场景一：消息大小不足16K。

首先会根据topic和分区获取所属队列的最后一个ProducerBatch，

*  如果最后一个ProducerBatch+当前消息 <= 16K时，会把当前消息存入这个ProducerBatch中，等待发送。
 *  如果最后一个ProducerBatch+当前消息 > 16K时，此时消息不会放入这个ProducerBatch中，而是会向池化的队列中获取一个ByteBuffer，把这个ByteBuffer放到队列的尾部，然后把消息放入这个新增的ProducerBatch中。
 *  如果最后一个ProducerBatch+当前消息 > 16K时，并且池化的队列中没有可用的ByteBuffer时，池化队列会向剩余的未使用的内存空间（nonPooledAvailableMemory）申请一个大小为16K的内存空间，添加到池化队列尾部。然后把这个新增的ByteBuffer添加到分区下的队列尾部，存储新的消息。

场景二：消息大小超过16K

*  当消息超过16K时，任何一个ProducerBatch都无法存储这个消息。此时会直接向剩余的空间（nonPooledAvailableMemory）的进行分配和当前的消息大小一样的内存空间，加到队列的尾部，然后存储消息，等待发送。
 *  当剩余的空间（nonPooledAvailableMemory） < 消息大小时，nonPooledAvailableMemory会向池化队列获取空间，每次获取一个ByteBuffer(16K)，直到nonPooledAvailableMemory的空间大于或等于消息大小时。获取的ByteBuffer会经过jvm的GC垃圾回收。过程比较慢。当nonPooledAvailableMemory空间大于获取等于消息大小时，会把分配消息大小的空间放入分区队列的尾部，把消息存入这个ProducerBatch内。

##### 5. 生产者会有一个send线程，用于不断的获取消息和发送消息。 #####

sender线程会不断的扫描RecordAccumulator中所有的ProducerBatch，如果ProducerBatch达到batch.size（默认16K）大小或者最早的一个消息已经等待超过linger.ms（默认为0）时，这个ProducerBatch会被sender线程收集到。由于不同的topic和分区会被分到不同的Broker节点上，sender线程会把发送到相同Broker姐节点的ProducerBatch合并在一个Request请求中，一个Request请求不会超过max.request.size（默认1048576B = 1M）

##### 6. 每个请求都会缓存在一个inFlightRequest缓冲区内，里面为每一个Broker分配了一个队列。 #####

新的请求会放在队列尾部，每个队列最多能够容纳max.in.flight.requests.per.connection（默认值为5）个Request，队列满了不会产生新的Request。

##### 7. selector获取到Request会发往相对应的Broker节点。Broker节点收到Request后会进行ACK确认这个Request #####

> acks 有三个配置值:\[-1 , 0 , 1\]  
> acks = -1 表示不需要收到leader节点的ACK回复就会发送下一个Request。高吞吐，低一致性  
> acks = 0 表示只需要接收到leader节点的ACK后就可以发送下一个Request。  
> acks = 1 表示 需要接收到leaer节点和ISR节点的ACK后才会发送下一个Request。一致性较高

##### 8. 当收到Broker对某个Request的ACK后，会删除inFlightRequest队列中这个Request。然后调用clear方法清除对应的ProducerBatch。 #####

RecordAccumulator Clear清理场景：针对2.4.1.1，2.4.1.2，2.4.1.3，ProducerBatch都会标记为删除，然后放入池化队列中，不会进行GC。2.4.1.3中从nonPooledAvailableMemory获取的内存也不会归还給nonPooledAvailableMemory，任然存放在池化队列中。  
针对2.4.2.1，2.4.2.2，超过16K的消息内存空间会被GC进行回收，然后作为nonPooledAvailableMemory的一部分

##### 9. 如果发送过程中产生了异常，消息发送会存在重试机制。条件为重试次数小于指定值&&异常为RetriableException #####

private boolean canRetry(ProducerBatch batch, ProduceResponse.PartitionResponse response) {
        
        return batch.attempts() < this.retries &&
            ((response.error.exception() instanceof RetriableException) ||
             (transactionManager != null && transactionManager.canRetry(response, batch)));
    }

![2bd22882e49c4a0a9af6797f1e6e4a04.png][]

![26c05957c2b742659e41527d1067678d.png][]

#### 生产端的高并发核心架构设计 ####

在消息发送时候，可发现这两个亮点\*\*：批量消息**和**自定义协议格式。\*\*

*  **批量发送**：减少了与服务端 Broker 处理请求的次数，从而提升总体的处理能力。

调用 send() 方法时，不会立刻把消息发送出去，而是缓存起来，选择恰当时机把缓存里的消息划分成一批数据，按批次发送给服务端 Broker。

*  **自定义协议格式**：序列化方式和压缩格式都能减少数据体积，从而节省网络资源消耗。

**各种压缩算法对比**：

*  吞吐量方面：LZ4 > Snappy > zstd 和 GZIP
 *  压缩比方面：zstd > LZ4 > GZIP > Snappy

<table> 
 <thead> 
  <tr> 
   <th>Compressor name</th> 
   <th>Ratio</th> 
   <th>Compression</th> 
   <th>Decompress.</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><strong>zstd 1.3.4-1</strong></td> 
   <td>2.877</td> 
   <td>470 MB/s</td> 
   <td>1380 MB/s</td> 
  </tr> 
  <tr> 
   <td>zlib 1.2.11-1</td> 
   <td>2.743</td> 
   <td>110 MB/s</td> 
   <td>400 MB/s</td> 
  </tr> 
  <tr> 
   <td>brotli 1.0.2-0</td> 
   <td>2.701</td> 
   <td>410 MB/s</td> 
   <td>430 MB/s</td> 
  </tr> 
  <tr> 
   <td>quicklz 1.5.0-1</td> 
   <td>2.238</td> 
   <td>550 MB/s</td> 
   <td>710 MB/s</td> 
  </tr> 
  <tr> 
   <td>lzo1x2.09-1</td> 
   <td>2.108</td> 
   <td>650 MB/s</td> 
   <td>830 MB/s</td> 
  </tr> 
  <tr> 
   <td>lz4 1.8.1</td> 
   <td>2.101</td> 
   <td>750 MB/s</td> 
   <td>3700 MB/s</td> 
  </tr> 
  <tr> 
   <td>snappy 1.1.4</td> 
   <td>2.091</td> 
   <td>530 MB/s</td> 
   <td>1800 MB/s</td> 
  </tr> 
  <tr> 
   <td>lzf 3.6-1</td> 
   <td>2.077</td> 
   <td>400 MB/s</td> 
   <td>860 MB/s</td> 
  </tr> 
 </tbody> 
</table>

### 二、服务端 Broker ###

Broker 的高性能主要从这 3 个方面体现：

*  PageCache 缓存
 *  Kafka 的文件布局 以及 磁盘文件顺序写入
 *  零拷贝 sendfile：加速消费流程

下面展开讲讲。

#### 1、PageCache 加速消息读写 ####

使用 PageCache 主要能带来如下好处：

*  写入文件的时候：操作系统会先把数据写入到内存中的 PageCache，然后再一批一批地写到磁盘上，从而减少磁盘 IO 开销。

![1b40a79485c64e77b7550ff9e78e6d0c.png][]

*  读取文件的时候：也是从 PageCache 中来读取数据。

如果消息刚刚写入到服务端就会被消费，按照 LRU 的“优先清除最近最少使用的页”这种策略，读取的时候，对于这种刚刚写入的 PageCache，命中的几率会非常高。

#### 2、Kafka 的文件布局 以及 磁盘文件顺序写入 ####

文件布局如下图所示：

![ba6cebe62e874fa793a00de44c412ec4.png][]

**主要特征是**：文件的组织方式是“topic + 分区”，每一个 topic 可以创建多个分区，每一个分区包含单独的文件夹。

**Kafka 在分区级别实现文件顺序写：即多个文件同时写入，更能发挥磁盘 IO 的性能。**

*  **相对比 RocketMQ**：RocketMQ 在消息写入时追求极致的顺序写，所有的消息不分主题一律顺序写入 commitlog 文件， topic 和 分区数量的增加不会影响写入顺序。
 *  **弊端**： Kafka 在消息写入时的 IO 性能，会随着 topic 、分区数量的增长先上升，后下降。

所以使用 Kafka 时，要警惕 Topic 和 分区数量。

#### 3、零拷贝 sendfile：加速消费流程 ####

**当不使用零拷贝技术读取数据时**：

![8fe6c653b06f4ba5b29ca9201b38abb2.png][]

**流程如下**：

1）消费端 Consumer：向 Kafka Broker 请求拉取消息

2）Kafka Broker 从 OS Cache 读取消息到 应用程序的内存空间：

*  若 OS Cache 中有消息，则直接读取；
 *  若 OS Cache 中无消息，则从磁盘里读取。

3）再通过网卡，socket 将数据发送给 消费端Consumer

**当使用零拷贝技术读取数据**：

![6abad3a97f7f4686994684a526a0db28.png][]

**Kafka 使用零拷贝技术可以把这个复制次数减少一次，直接从 PageCache 中把数据复制到 Socket 缓冲区中。**

*  这样不用将数据复制到用户内存空间。
 *  DMA 控制器直接完成数据复制，不需要 CPU 参与，速度更快。

### 三、消费端 Consumer ###

消费者只从 Leader分区批量拉取消息。

为了提高消费速度，多个消费者并行消费比不可少。Kafka 允许创建消费组(唯一标识 group.id)，在同一个消费组的消费者共同消费数据。

**举个例子**：

*  有两个 Kafka Broker，即有 2个机子
 *  有一个主题：TOPICA，有 3 个分区(0, 1, 2)

![cbbaaaffc3a04815afb0d75e84ff3d42.png][]

**如上图，举例 4 中情况**：

*  group.id = 1，有一个消费者：这个消费者要处理所有数据，即 3 个分区的数据。
 *  group.id = 2，有两个消费者：consumer 1消费者需处理 2个分区的数据，consumer2 消费者需处理 1个分区的数据。
 *  group.id = 3，有三个消费者\*\*：消费者数量与分区数量相等，\*\*刚好每个消费者处理一个分区。
 *  group.id = 4，有四个消费者\*\*：消费者数量 > 分区数量，\*\*第四个消费者则会处于空闲状态。

### 说在最后 ###

kafka相关面试题，是非常常见的面试题。

以上的内容，如果大家能对答如流，如数家珍，基本上 面试官会被你 震惊到、吸引到。

最终，**让面试官爱到 “不能自已、口水直流”**。offer， 也就来了。

学习过程中，如果有啥问题，大家可以来 找 40岁老架构师尼恩交流。

### 参考文献 ###

https://juejin.cn/post/7134463012563320868

### 推荐阅读 ###

《[炸裂，靠“吹牛”过京东一面，月薪40K][40K]》

《[太猛了，靠“吹牛”过顺丰一面，月薪30K][30K]》

《[炸裂了…京东一面索命40问，过了就50W+][40_50W]》

《[问麻了…阿里一面索命27问，过了就60W+][27_60W]》

《[百度狂问3小时，大厂offer到手，小伙真狠！][3_offer]》

《[饿了么太狠：面个高级Java，抖这多硬活、狠活][Java 1]》

《[字节狂问一小时，小伙offer到手，太狠了！][offer]》

《[收个滴滴Offer：从小伙三面经历，看看需要学点啥？][Offer]》

> 《尼恩 架构笔记》《尼恩高并发三部曲》《尼恩Java面试宝典》PDF，请到下面公号【技术自由圈】取↓↓↓

[Java]: https://blog.csdn.net/crazymakercircle/article/details/124790425
[b8bc45927c9840adb23f2699b052b5a8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/54bbc196b3fe4b64bd722cf77b8fa555.png
[d84a046ee770469a9d2f1c1f2a924821.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/694ccf44ee0746c9bb34ea0a90c1f278.png
[2bd22882e49c4a0a9af6797f1e6e4a04.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/e88a6a8d72d048738100b6f6682f607c.png
[26c05957c2b742659e41527d1067678d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/3769f76f83f84163bd93d4287434a117.png
[1b40a79485c64e77b7550ff9e78e6d0c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/36195f64241448659b7f7d6298d1ea61.png
[ba6cebe62e874fa793a00de44c412ec4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/a1efa45fda5f46028c32dac6603327d0.png
[8fe6c653b06f4ba5b29ca9201b38abb2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/dd56864b1be6426c87513d7767cefb47.png
[6abad3a97f7f4686994684a526a0db28.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/4b4164c7495f456cb8a037119d0aa245.png
[cbbaaaffc3a04815afb0d75e84ff3d42.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/9b3dd33f821844b989a423d7dc9240e2.png
[40K]: https://blog.csdn.net/crazymakercircle/article/details/132150378
[30K]: https://blog.csdn.net/crazymakercircle/article/details/132000420
[40_50W]: https://blog.csdn.net/crazymakercircle/article/details/131615296
[27_60W]: https://blog.csdn.net/crazymakercircle/article/details/131147192
[3_offer]: https://blog.csdn.net/crazymakercircle/article/details/131027012
[Java 1]: https://blog.csdn.net/crazymakercircle/article/details/130989697
[offer]: https://blog.csdn.net/crazymakercircle/article/details/130956376
[Offer]: https://blog.csdn.net/crazymakercircle/article/details/129317821