揭开Kafka高吞吐量神秘面纱

不念不忘少年蓝@ 2022-11-29 03:04 152阅读 0赞

前面已经对Kafka进行过详细介绍，Kafka是一个分布式发布 - 订阅消息系统和一个强大的队列，可以处理大量的数据，并能够将消息从一个端点传递到另一个端点。 Kafka适合离线和在线消息消费。 Kafka消息保留在磁盘上，并在群集内复制以防止数据丢失。

Kafka专为分布式高吞吐量系统而设计。 Kafka往往工作得很好，作为一个更传统的消息代理的替代品。 与其他消息传递系统相比，Kafka具有更好的吞吐量，内置分区，复制和固有的容错能力，这使得它非常适合大规模消息处理应用程序。

不同于Redis和MemcacheQ等内存消息队列，Kafka的设计是把所有的Message都要写入速度低容量大的硬盘，以此来换取更强的存储能力。实际上，Kafka使用硬盘并没有带来过多的性能损失。

能够存储到几乎无限大的磁盘空间而无需付出性能代价是kafka的一大法宝，意味着它有着从容面对海量数据的能力。

Kafka的消息存储是严重依赖于文件系统的，消息持久化时只是简单的文件读写。这种方式有个优势，操作的复杂度都是O(1),读写操作互不干扰，这样性能就完全同数据大小脱离的关系，数据量大简单加磁盘就行了。

从直觉来看，读写磁盘速度会很慢，实际上由于Kafka采用了顺序读写磁盘的方式，这种I/O的效率并不慢，某些情况下比操作内存更快。

对于人们传统的直觉，磁盘的吞吐性能可快可慢，这取决于磁盘的使用方式，在一个由6个7200rpm的SATA硬盘组成的RAID-5磁盘阵列上，线性写入（linear write）的速度大约是300MB/秒，但随即写入却只有50k/秒，其中的差别接近10000倍。

这是由于随即写入时磁盘需要再次寻道，严重影响了读写的效能，再加之操作系统采用预读和预写的方式对磁盘读写进行优化，进一步提升了顺序读写磁盘的效率。  
kafka主要通过以下几种方式实现了超高的吞吐率：

## 1.顺序读写磁盘 ##

Kafka 的消息是不断追加到文件中的，这个特性使 Kafka 可以充分利用磁盘的顺序读写性能。  
顺序读写不需要硬盘磁头的寻道时间，只需很少的扇区旋转时间，所以速度远快于随机读写。

实现顺序读写磁盘、利用page cache，将文件数据映射到内存，利用sendfile网传时socket通信时直接读取内存区域（减少操作系统上下文切换、零拷贝提速）；  
生产者负责写入数据，Kafka会将消息持久化到磁盘，保证不会丢失数据，Kafka采用了俩个技术提高写入的速度。  
**1.顺序写入**：硬盘需要指针寻址找到存储数据的位置，所以，如果是随机IO，磁盘会进行频繁的寻址，导致写入速度下降。  
Kafka使用了顺序IO提高了磁盘的写入速度，Kafka会将数据顺序插入到文件末尾，消费者端通过控制偏移量来读取消息，这样做会导致数据无法删除，时间一长，磁盘空间会满，kafka提供了2种策略来删除数据：基于时间删除和基于partition文件的大小删除。

**2.Memory Mapped Files**：这个和Java NIO中的内存映射基本相同，mmf直接利用操作系统的Page来实现文件到物理内存的映射，完成之后对物理内存的操作会直接同步到硬盘。  
mmf通过内存映射的方式大大提高了IO速率，省去了用户空间到内核空间的复制。它的缺点显而易见那就是不可靠。  
当发生宕机而数据未同步到硬盘时，数据会丢失，Kafka提供了produce.type参数来控制是否主动的进行刷新，如果kafka写入到mmp后立即flush再返回给生产者则为同步模式，反之为异步模式。

## 2.零拷贝 ##

**零拷贝(直接让操作系统的 Cache 中的数据发送到网卡后传输给下游的消费者)**：平时从服务器读取静态文件时，服务器先将文件从复制到内核空间，再复制到用户空间，最后再复制到内核空间并通过网卡发送出去，而零拷贝则是直接从内核到内核再到网卡，省去了用户空间的复制。

在Linux kernel2.2 之后出现了一种叫做"零拷贝(zero-copy)"系统调用机制，就是跳过“用户缓冲区”的拷贝，建立一个磁盘空间和内存的直接映射，数据不再复制到“用户态缓冲区”，系统上下文切换减少为2次，可以提升一倍的性能。

零拷贝并不是不需要拷贝，而是减少不必要的拷贝次数。通常是说在 IO 读写过程中。“零拷贝技术”只用将磁盘文件的数据复制到页面缓存中一次，然后将数据从页面缓存直接发送到网络中。

#### 传统模式下的四次拷贝与四次上下文切换 ####

传统模式下，一般先将文件数据读入内存，然后通过Socket将内存中的数据发送出去。

这一过程实际上发生了四次数据拷贝。首先通过系统调用将文件数据读入到内核态Buffer（DMA拷贝），然后应用程序将内存态Buffer数据读入到用户态Buffer（CPU拷贝），接着用户程序通过Socket发送数据时将用户态Buffer数据拷贝到内核态Buffer（CPU拷贝），最后通过DMA拷贝将数据拷贝到NIC Buffer。同时，还伴随着四次上下文切换，如下图所示。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center]

#### sendfile和transferTo实现零拷贝 ####

Linux 2.4+内核通过sendfile系统调用，提供了零拷贝。数据通过DMA拷贝到内核态Buffer后，直接通过DMA拷贝到NIC Buffer，无需CPU拷贝。这也是零拷贝这一说法的来源。除了减少数据拷贝外，因为整个读文件-网络发送由一个sendfile调用完成，整个过程只有两次上下文切换，因此大大提高了性能。零拷贝过程如下图所示。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 1]  
从具体实现来看，Kafka的数据传输通过TransportLayer来完成，其子类PlaintextTransportLayer通过Java NIO的 FileChannel 的 transferTo和transferFrom方法实现零拷贝，代码如下：

@Override
    public long transferFrom(FileChannel fileChannel, long position, long count) throws IOException { 
    return fileChannel.transferTo(position, count, socketChannel);
    }

**注意**：  
transferTo和transferFrom并不保证一定能使用零拷贝。实际上是否能使用零拷贝与操作系统相关，如果操作系统提供sendfile这样的零拷贝系统调用，则这两个方法会通过这样的系统调用充分利用零拷贝的优势，否则并不能通过这两个方法本身实现零拷贝。

## 3.批处理 ##

批处理是一种常用的用于提高I/O性能的方式。对Kafka而言，批处理既减少了网络传输的Overhead，又提高了写磁盘的效率。

Kafka 允许进行批量发送消息，先将消息缓存在内存中，然后一次请求批量发送出去比如可  
以指定缓存的消息达到某个量的时候就发出去，或者缓存了固定的时间后就发送出去如 100  
条消息就发送，或者每 5 秒发送一次这种策略将大大减少服务端的 I/O 次数。

Kafka 0.8.1及以前的Producer区分同步Producer和异步Producer。同步Producer的send方法主要分两种形式：  
一种是接受一个KeyedMessage作为参数，一次发送一条消息。  
另一种是接受一批KeyedMessage作为参数，一次性发送多条消息。

而对于异步发送而言，无论是使用哪个send方法，实现上都不会立即将消息发送给Broker，而是先存到内部的队列中，直到消息条数达到阈值或者达到指定的Timeout才真正的将消息发送出去，从而实现了消息的批量发送。

Kafka 0.8.2开始支持新的Producer API，将同步Producer和异步Producer结合。虽然从send接口来看，一次只能发送一个ProducerRecord，而不能像之前版本的send方法一样接受消息列表，但是send方法并非立即将消息发送出去，而是通过batch.size和linger.ms控制实际发送频率，从而实现批量发送。

由于每次网络传输，除了传输消息本身以外，还要传输非常多的网络协议本身的一些内容（称为Overhead），所以将多条消息合并到一起传输，可有效减少网络传输的Overhead，进而提高了传输效率。  
从零拷贝章节的图中可以看到，虽然Broker持续从网络接收数据，但是写磁盘并非每秒都在发生，而是间隔一段时间写一次磁盘，并且每次写磁盘的数据量都非常大（最高达到718MB/S）。

## 4.分区（Partition） ##

Kafka 的队列 topic 被分为了多个区 partition，每个 partition 又分为多个段 segment，所以一个队列中的消息实际上是保存在 N 多个片段文件中通过分段的方式，每次文件操作都是对一个小文件的操作，非常轻便，同时也增加了并行处理能力。

## 5.数据压缩 ##

Kafka 还支持对消息集合进行压缩，Producer 可以通过 GZIP 或 Snappy 格式对消息集合进行  
压缩，压缩的好处就是减少传输的数据量，减轻对网络传输的压力。

Kafka从0.7开始，即支持将数据压缩后再传输给Broker。除了可以将每条消息单独压缩然后传输外，Kafka还支持在批量发送时，将整个Batch的消息一起压缩后传输。数据压缩的一个基本原理是，重复数据越多压缩效果越好。因此将整个Batch的数据一起压缩能更大幅度减小数据量，从而更大程度提高网络传输效率。

Producer压缩之后，在Consumer需进行解压，批处理和数据压缩一起使用,单独做数据压缩的话，效果不明显。

Broker接收消息后，并不直接解压缩，而是直接将消息以压缩后的形式持久化到磁盘。Consumer Fetch到数据后再解压缩。因此Kafka的压缩不仅减少了Producer到Broker的网络传输负载，同时也降低了Broker磁盘操作的负载，也降低了Consumer与Broker间的网络传输量，从而极大得提高了传输效率，提高了吞吐量。

## 6.高效的序列化方式 ##

Kafka消息的Key和Payload（或者说Value）的类型可自定义，只需同时提供相应的序列化器和反序列化器即可。  
因此用户可以通过使用快速且紧凑的序列化-反序列化方式（如Avro，Protocal Buffer）来减少实际网络传输和磁盘存储的数据规模，从而提高吞吐率。  
这里要注意，如果使用的序列化方法太慢，即使压缩比非常高，最终的效率也不一定高。

## 7.Consumer 的负载均衡 ##

当一个 group 中,有 consumer 加入或者离开时,会触发 partitions 均衡.均衡的最终目的,是提升  
topic 的并发消费能力。

#### 负载均衡 ####

在创建一个 Topic 时，Kafka 尽量将 Partition 均分在所有的 Broker 上，并且Replicas 也均分在不同的 Broker 上，这点如前面分区策略、副本策略中所述。

另外关于 Leader 的负载均衡也需要注意，当一个 Broker 停止时，所有本来将它作为 Leader 的分区将会把 Leader 转移到其他 Broker 上去，极端情况下，会导致同一个 Leader 管理多个分区，导致负载不均衡，同时当这个 Broker 重启时，如果这个 Broker 不再是任何分区的 Leader，Kafka 的 Client 也不会从这个 Broker来读取消息，从而导致资源的浪费。

Kafka 中有一个被称为优先副本（preferred replicas）的概念。如果一个分区有 3 个副本，且这 3 个副本的优先级别分别为 0，1，2，根据优先副本的概念，0会作为Leader。  
当0节点的Broker挂掉时，会启动1这个节点Broker当做Leader。  
当0节点的Broker再次启动后，会自动恢复为此 Partition 的Leader。不会导致负 载 不 均 衡 和 资 源 浪 费 ， 这 就 是 Leader 的均衡 机 制 。 可 在 配 置 文 件conf/ server.properties 中配置开启（默认就是开启）。

auto.leader.rebalance.enable=true

例如，某个 Topic 详情如下：

./kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic logdata-es
    
    Topic: logdata-es PartitionCount:2 ReplicationFactor:2 Configs:
    
    Topic: logdata-es Partition: 0 Leader: 0 Replicas: 0,1 Isr: 0,1
    
    Topic: logdata-es Partition: 1 Leader: 1 Replicas: 1,0 Isr: 0,1

Topic logdata-es 中 Partition 0 的 Replicas 为\[0,1\]，则 0 为 Preferred Replica。

### 总结： ###

**Kafka的设计目标是高吞吐量**，**它比其它消息系统快的原因体现在以下几方面**：

1.  Kafka操作的是序列文件I /O（序列文件的特征是按顺序写，按顺序读），为保证顺序，Kafka强制点对点的按顺序传递消息，这意味着，一个consumer在消息流（或分区）中只有一个位置。
2.  Kafka不保存消息的状态，即消息是否被“消费”。一般的消息系统需要保存消息的状态，并且还需要以随机访问的形式更新消息的状态。  
    而Kafka 的做法是保存Consumer在Topic分区中的位置offset，在offset之前的消息是已被“消费”的，在offset之后则为未“消费”的，并且offset是可以任意移动的，这样就消除了大部分的随机IO。
3.  Kafka支持点对点的批量消息传递。
4.  Kafka的消息存储在OS pagecache（页缓存，page cache的大小为一页，通常为4K，在Linux读写文件时，它用于缓存文件的逻辑内容，从而加快对磁盘上映像和数据的访问）。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/ae333fef8b4a4cd694b520d714cff8ab.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221124/716446b79e4846648435477315c4ca5f.png