kafka零拷贝（kafka零拷贝 面试）

本篇文章给大家谈谈kafka零拷贝，以及kafka零拷贝 面试对应的知识点，希望对各位有所帮助，不要忘了收藏本站喔。

本文目录一览：

• 1、架构师经典总结：为什么零拷贝可以提升Kafka性能？
• 2、零拷贝技术
• 3、2、kafka为何能实现每秒几十万的吞吐量
• 4、kafka原理分析
• 5、为什么Kafka这么快
• 6、Kafka零拷贝

架构师经典总结：为什么零拷贝可以提升Kafka性能？

计算机系统是由“硬件”和“软件”两大部分组成，计算机硬件包括一个或多个处理器(CPU)、内存、键盘、显示器、磁盘、I/O接口以及其他一些外围设备比如打印机，绘图仪等等。 总之，计算机硬件部分是一个由多种电子和机械设备组成的硬件系统。

为了让人方便正确使用这些设备，就需要编写若干程序来管理这些设备，正是这些程序组成了计算机的软件系统。软件也可以分为两大类：系统软件和应用软件。人们首先直接在硬件上加载一层程序，用它来管理整个计算机硬件设备以及一些软件信息资源，同时还为用户提供开发应用程序的环境，这就是操作系统软件和实用软件。应用软件是在操作系统支持下，为实现用户要求而编制的各种应用程序。

CPU、内存和I/O接口组成的主设备通常称为主机，把没有加载操作系统的主机叫做裸机。裸机与操作系统软件的接口是由CPU的指令系统和厂商提供的系统BIOS组成。

由于操作系统向用户隐藏了系统使用的硬件设备，因此操作系统要为它上面的应用系统软件提供一组命令或系统调用接口供用户程序使用。比如我们需要使用磁盘，可以通过系统命名或系统调用来间接完成，而不需要亲自手动编写一个磁盘设备驱动程序。因此对于用户来说，当计枯键磨算机加载操作系统后，用户不直接与没斗计算机硬件打交道，而是利用操作系统提供的命令和功能区使用计算机。

由于操作系统处于硬件和软件的中央位置，因此很早就有人把操作系统成为计算机系统软件的核心，简称 核心 或 内核 。

内核态和用户态

从系统安全和保护的角度出发，在进行计算机体系结构设计时，处理机的执行模式一般设定为两种：分别称为内核模式(内核态)和用户模式(用户态)。当处理机处于内核模式执行时，意味着系统除了可以执行一般指令外，还可以执行特权指令，即可以执行访问各种控制寄存器的指令、I/O指令以及程序状态字。

当处理机处于用户模式执行时，只能执行一般指令，而不允许执行特权指令。这样做可以保护核心代码不受用户程序有意和无意的攻击。 显然，处理机在运行期间需要在内核模式和用户模式之前进行切换。

零拷贝

Kafka使用零拷贝(Zero-Copy)技术来提供它的性能，所谓的零拷贝是指将数据直接从磁盘文件复制到网卡设备中，而不需要经由应用程序之手，减少了内核和用户模式之间的上下文切换，零拷贝技术通过DMA技术实现。

直接存储器存取方式(Direct Memory Access, DMA) DMA控制方式是以存储器为中心，在主存和I/O设备之间建立一条直接通路，在DMA控制器的控制下进行设备和主存之间的数据交换。这种方式只在传输开始和传输结束时才需要CPU的干预。它非常适用于高速设备与主存之间的成批数据传输。

我们看下下面这样的这样一亮尺个场景：

客户端在游览器中发起请求获取内容，到看到具体内容经历了什么?

首先，该请求经过解析后，通过系统调用由用户态转为核心态执行，在核心态由操作系统中的TCP/IP协议代码和网卡驱动程序控制网卡把请求发送到相应的网络上， 等待Web服务器相应。当服务器返回时，由网卡接受，并通过内核传送给客户程序。

在服务器端，内核通过网卡从网络上接受Web请求，并通过系统调用传递给Web服务器。Web服务器根据此服务请求执行相应的服务进程，并由内核把结果发送到网络上传送给用户。

从上图中可以看到如果服务器从准备数据到发送数据经历了下面4个过程。

从上面过程可以看出，数据是先从内核模式--用户模式--内核模式，浪费了2次复制过程：第一次是从内核模式复制到用户模式；第二次是从用户模式再复制回内核模式，而且在上面的过程中，内核和用户模式的上下文切换也是4次。

如果采用了零拷贝技术，那么应用程序就可以直接请求内核把磁盘中的数据传输给Socket.

零拷贝技术通过DMA技术将文件内容复制到内核模式下的Read Buffer中。不过没有数据被复制到Socket Buffer，只有包含数据的位置和长度的信息的文件描述符被加到Socket Buffer中。DMA引擎直接将数据从内核模式中传递到网卡设备。这里上下文切换变成了2次,也只经历了2次复制过程就从磁盘中传送出去了。

零拷贝技术

知识补充：

1、DMA是直接内存访问（ Direct Memory Access ） 技术，早期 DMA 只存在在主板上，如今由于 I/O 设备越来越多，数据传输的需求也不尽相同，所以每个 I/O 设备里面都有自己的 DMA 控制器。

2、每次系统调用都得先从用户态切换到内激闹核态，等内核完成任务后，再从内核态切换回用户态。

我们都听过 kafka 很快，其中一个原因是 kafka 使用零拷贝。看看从文件中读取数据并通过网络将数据传输到另一个程序的场景，在内部的复制操作，需要在用户模式和内核明旁罩模式之间进行四次上下文切换，并且进行数据复制四次。

完成所有四个操作后，它将再次切换到用户模式。

回顾以上动作，其实发现实际上第二个和第三个数据拷贝是可以避免的。Java 类库通过 java.nio.channels 中的 transferTo() 方法在 UNIX 系统上执行零副本，使用零拷贝的应用程序请求内核直接将数据从磁盘文件复制到套接字，而不通过应用程序。零拷贝极大提高了应用程序性能，并减启樱少了内核和用户模式之间的上下文切换次数。

我们看看 transferTo() 是如何复制数据的？

这里我们还需要 3 个副本和 2 个上下文切换。

但当前还没有达到零拷贝，如果底层网卡支持收集操作，可以进一步减少内核重复拷贝数据的操作。在 Linux 内核 2.4 及更高版本中，套接字缓冲区描述符支持该场景。

Kafka 和 Nginx 都有实现零拷贝技术，这将大大提高文件传输的性能。拷贝技术，本质上讲就是通过减少非必要的内存拷贝以及上下文切换，来提高文件在通道间复制速度的一种技术。以本文中的transferTo()方法为例，通过该技术，可以将原来 四次内存间拷贝减少成两次，将四次上下文切换减少成两次 ，大大提高复制的速度。但零拷贝技术并非万能的，它有自己的使用场景，对于将大量数据从一个 I/O 通道复制到另一个通道的情况（例如 Web 服务器），都是合适的。

[img]

2、kafka为何能实现每秒几十万的吞吐量

Kafka是高吞吐量低延迟的高并发、高性能的消息中间件，在大数据领域有广泛的应用。那他是如何做到这么高的吞吐量和高性能呢？

生产者通过多batch合并一个request 一次性发送broker提高吞吐量 。

每个Kafka服务端叫做一个broker，负责管理一台机器型哪派上的数据；每个topic拆分成多个partition，这样每个partition存储一部分数据并放在不同的broker上。这时候生产者如果生产一条消息，就建立连接然后发送数据，效率肯定不高。

Kafka会在生产者放一个内存缓冲区，当生产消息后，它会把同一个topic同一个partition的消息打包成一个batch。

这样就结束了吗？当然不是，如果生产者这边有多个topic，不同topic发送到同一个broker的数据是否可以合并呢？当然可以

如果客户端那边有2个topic，每个topic有个3个partition，那每个broker就会有2个partition。这时候我们完全可以将发送到同一个broker的batch一次发送。所以kafka将发送到同一个broker的batch打包成一个request发送，进一步提高了通信的效率。

首先kafka在接收到数据后都会写磁盘，但是我们都知道写磁盘性能会很差，所以它并不是直接写磁盘。操作系统本身有一层缓存叫做页缓存，相当于内存，所卜贺以kafka是把数据写入到页缓存中，接下来由页缓存决定什么时候写入到磁盘。

普通机械磁盘如果随机缓好写的话，性能会非常差，Kafka采用磁盘顺序写的方式，仅仅将数据追加到文件末尾，这种方式性能基本可以和内存媲美。

上面说的是kafka写入数据的优秀设计，那从磁盘读数据分发又是如何设计的呢？

我们可以想想分发数据一般是如何做的。首先肯定是从磁盘中读出数据到页缓存，然后从页缓存中拷贝到kafka中，然后再从kafka中拷贝到socket中，最后再给网卡。

而零拷贝技术，就是去除上面2步拷贝，直接从页缓存中将数据发送到网卡中，socket中仅仅只是拷贝一个描述符，这个过程大大提升了读取的性能。而且从磁盘读数据时候会先看看页缓存中有没有，如果有的话都不用读磁盘了。

kafka原理分析

作为一款典型的消息中间件产品，kafka系统仍然由producer、broker、consumer三部分组成。kafka涉及的几个常用概念和组件简派薯单介绍如下：

当consumer group的状态发生变化（如有consumer故障、增减consumer成员等）或consumer group消费的topic状态发生变化（如增加了partition，消费的topic发生变化），kafka集群会自动调整和重新分配consumer消费的partition，这个过程就叫做rebalance（再平衡）。

__consumer_offsets是kafka集群自己维护的一个特殊的topic，它里面存储的是每个consumer group已经消费了每个topic partition的offset。__consumer_offsets中offset消息的key由group id，topic name，partition id组成，格式为 {topic name}-${partition id}，value值就是consumer提交的已消费的topic partition offset值。__consumer_offsets的分区数和副本数分别由offsets.topic.num.partitions（默认值为50）和offsets.topic.replication.factor（默认值为1）参数配置。我们通过公式 hash(group id) % offsets.topic.num.partitions 就可以计算出指定consumer group的已提交offset存储的partition。由于consumer group提交的offset消息只有最后一条消息有意义，所以__consumer_offsets是一个compact topic，kafka集群会周期性的对__consumer_offsets执行compact操作，只保留最新的一次提交offset。

group coordinator运行在kafka某个broker上，负责consumer group内所有的consumer成员管理、所有的消费的topic的partition的消费关系分配、offset管理、触发rebalance等功能。group coordinator管理partition分配时，会指定consumer group内某个consumer作为group leader执行具体的partition分配任务。存储某个consumer group已提交offset的__consumer_offsets partition leader副本所在的broker就是该consumer group的协调器运行的broker。

跟大多数分布式系统一样，集群有一个master角色管理整个集群，协调集群中各个成员的行为。kafka集群中的controller就相当于其它分布式系统的master，用来负责集群topic的分区分配，分区leader选举以及维护集群的所有partition的ISR等集群协调功能。集群中哪个borker是controller也是通过一致性协议选举产生的，2.8版本之前通腔销过zookeeper进行选主，2.8版本后通过kafka raft协议进行选举。如果controller崩溃，集群会重新选举一个broker作为新的controller，并增加controller epoch值（相当于zookeeper ZAB协议的epoch，raft协议的term值）

当kafka集群新建了topic或为一个topic新增了partition，controller需要为这些新增加的partition分配到具体的broker上，并把分配结果记录下来，供producer和consumer查询获取。

因为只有partition的leader副本才会处理producer和consumer的读写请求，而partition的其他follower副本需要从相应的leader副本同步消息，为了尽量保证集群中所有broker的负载是均衡的，controller在进行集群全局partition副本伍羡游分配时需要使partition的分布情况是如下这样的：

在默认情况下，kafka采用轮询（round-robin）的方式分配partition副本。由于partition leader副本承担的流量比follower副本大，kafka会先分配所有topic的partition leader副本，使所有partition leader副本全局尽量平衡，然后再分配各个partition的follower副本。partition第一个follower副本的位置是相应leader副本的下一个可用broker，后面的副本位置依此类推。

举例来说，假设我们有两个topic，每个topic有两个partition，每个partition有两个副本，这些副本分别标记为1-1-1，1-1-2，1-2-1，1-2-2，2-1-1，2-1-2，2-2-1，2-2-2（编码格式为topic-partition-replia，编号均从1开始，第一个replica是leader replica，其他的是follower replica）。共有四个broker，编号是1-4。我们先对broker按broker id进行排序，然后分配leader副本，最后分配foller副本。

1）没有配置broker.rack的情况

现将副本1-1-1分配到broker 1，然后1-2-1分配到broker 2，依此类推，2-2-1会分配到broker 4。partition 1-1的leader副本分配在broker 1上，那么下一个可用节点是broker 2，所以将副本1-1-2分配到broker 2上。同理，partition 1-2的leader副本分配在broker 2上，那么下一个可用节点是broker 3，所以将副本1-1-2分配到broker 3上。依此类推分配其他的副本分片。最后分配的结果如下图所示：

2）配置了broker.rack的情况

假设配置了两个rack，broker 1和broker 2属于Rack 1，broker 3和broker 4属于Rack 2。我们对rack和rack内的broker分别排序。然后先将副本1-1-1分配到Rack 1的broker 1，然后将副本1-2-1分配到下一个Rack的第一个broker，即Rack 2的broker 3。其他的parttition leader副本依此类推。然后分配follower副本，partition 1-1的leader副本1-1-1分配在Rack 1的broker上，下一个可用的broker是Rack 2的broker 3，所以分配到broker 3上，其他依此类推。最后分配的结果如下图所示：

kafka除了按照集群情况自动分配副本，也提供了reassign工具人工分配和迁移副本到指定broker，这样用户可以根据集群实际的状态和各partition的流量情况分配副本

kafka集群controller的一项功能是在partition的副本中选择一个副本作为leader副本。在topic的partition创建时，controller首先分配的副本就是leader副本，这个副本又叫做preference leader副本。

当leader副本所在broker失效时（宕机或网络分区等），controller需要为在该broker上的有leader副本的所有partition重新选择一个leader，选择方法就是在该partition的ISR中选择第一个副本作为新的leader副本。但是，如果ISR成员只有一个，就是失效的leader自身，其余的副本都落后于leader怎么办？kafka提供了一个unclean.leader.election配置参数，它的默认值为true。当unclean.leader.election值为true时，controller还是会在非ISR副本中选择一个作为leader，但是这时候使用者需要承担数据丢失和数据不一致的风险。当unclean.leader.election值为false时，则不会选择新的leader，该partition处于不可用状态，只能恢复失效的leader使partition重新变为可用。

当preference leader失效后，controller重新选择一个新的leader，但是preference leader又恢复了，而且同步上了新的leader，是ISR的成员，这时候preference leader仍然会成为实际的leader，原先的新leader变为follower。因为在partition leader初始分配时，使按照集群副本均衡规则进行分配的，这样做可以让集群尽量保持平衡。

为了保证topic的高可用，topic的partition往往有多个副本，所有的follower副本像普通的consumer一样不断地从相应的leader副本pull消息。每个partition的leader副本会维护一个ISR列表存储到集群信息库里，follower副本成为ISR成员或者说与leader是同步的，需要满足以下条件：

1）follower副本处于活跃状态，与zookeeper（2.8之前版本）或kafka raft master之间的心跳正常

2）follower副本最近replica.lag.time.max.ms（默认是10秒）时间内从leader同步过最新消息。需要注意的是，一定要拉取到最新消息，如果最近replica.lag.time.max.ms时间内拉取过消息，但不是最新的，比如落后follower在追赶leader过程中，也不会成为ISR。

follower在同步leader过程中，follower和leader都会维护几个参数，来表示他们之间的同步情况。leader和follower都会为自己的消息队列维护LEO（Last End Offset）和HW（High Watermark）。leader还会为每一个follower维护一个LEO。LEO表示leader或follower队列写入的最后一条消息的offset。HW表示的offset对应的消息写入了所有的ISR。当leader发现所有follower的LEO的最小值大于HW时，则会增加HW值到这个最小值LEO。follower拉取leader的消息时，同时能获取到leader维护的HW值，如果follower发现自己维护的HW值小于leader发送过来的HW值，也会增加本地的HW值到leader的HW值。这样我们可以得到一个不等式： follower HW = leader HW = follower LEO = leader LEO 。HW对应的log又叫做committed log，consumer消费partititon的消息时，只能消费到offset值小于或等于HW值的消息的，由于这个原因，kafka系统又称为分布式committed log消息系统。

kafka的消息内容存储在log.dirs参数配置的目录下。kafka每个partition的数据存放在本地磁盘log.dirs目录下的一个单独的目录下，目录命名规范为 ${topicName}-${partitionId} ，每个partition由多个LogSegment组成，每个LogSegment由一个数据文件(命名规范为： {baseOffset}.index）和一个时间戳索引文件（命名规范为：${baseOffset}.timeindex）组成，文件名的baseOffset就是相应LogSegment中第一条消息的offset。.index文件存储的是消息的offset到该消息在相应.log文件中的偏移，便于快速在.log文件中快速找到指定offset的消息。.index是一个稀疏索引，每隔一定间隔大小的offset才会建立相应的索引(比如每间隔10条消息建立一个索引)。.timeindex也是一个稀疏索引文件，这样可以根据消息的时间找到对应的消息。

可以考虑将消息日志存放到多个磁盘中，这样多个磁盘可以并发访问，增加消息读写的吞吐量。这种情况下，log.dirs配置的是一个目录列表，kafka会根据每个目录下partition的数量，将新分配的partition放到partition数最少的目录下。如果我们新增了一个磁盘，你会发现新分配的partition都出现在新增的磁盘上。

kafka提供了两个参数log.segment.bytes和log.segment.ms来控制LogSegment文件的大小。log.segment.bytes默认值是1GB，当LogSegment大小达到log.segment.bytes规定的阈值时，kafka会关闭当前LogSegment，生成一个新的LogSegment供消息写入，当前供消息写入的LogSegment称为活跃（Active）LogSegment。log.segment.ms表示最大多长时间会生成一个新的LogSegment，log.segment.ms没有默认值。当这两个参数都配置了值，kafka看哪个阈值先达到，触发生成新的LogSegment。

kafka还提供了log.retention.ms和log.retention.bytes两个参数来控制消息的保留时间。当消息的时间超过了log.retention.ms配置的阈值（默认是168小时，也就是一周），则会被认为是过期的，会被kafka自动删除。或者是partition的总的消息大小超过了log.retention.bytes配置的阈值时，最老的消息也会被kafka自动删除，使相应partition保留的总消息大小维持在log.retention.bytes阈值以下。这个地方需要注意的是，kafka并不是以消息为粒度进行删除的，而是以LogSegment为粒度删除的。也就是说，只有当一个LogSegment的最后一条消息的时间超过log.retention.ms阈值时，该LogSegment才会被删除。这两个参数都配置了值时，也是只要有一个先达到阈值，就会执行相应的删除策略

当我们使用KafkaProducer向kafka发送消息时非常简单，只要构造一个包含消息key、value、接收topic信息的ProducerRecord对象就可以通过KafkaProducer的send()向kafka发送消息了，而且是线程安全的。KafkaProducer支持通过三种消息发送方式

KafkaProducer客户端虽然使用简单，但是一条消息从客户端到topic partition的日志文件，中间需要经历许多的处理过程。KafkaProducer的内部结构如下所示：

从图中可以看出，消息的发送涉及两类线程，一类是调用KafkaProducer.send()方法的应用程序线程，因为KafkaProducer.send()是多线程安全的，所以这样的线程可以有多个；另一类是与kafka集群通信，实际将消息发送给kafka集群的Sender线程，当我们创建一个KafkaProducer实例时，会创建一个Sender线程，通过该KafkaProducer实例发送的所有消息最终通过该Sender线程发送出去。RecordAccumulator则是一个消息队列，是应用程序线程与Sender线程之间消息传递的桥梁。当我们调用KafkaProducer.send()方法时，消息并没有直接发送出去，只是写入了RecordAccumulator中相应的队列中，最终需要Sender线程在适当的时机将消息从RecordAccumulator队列取出来发送给kafka集群。

消息的发送过程如下：

在使用KafkaConsumer实例消费kafka消息时，有一个特性我们要特别注意，就是KafkaConsumer不是多线程安全的，KafkaConsumer方法都在调用KafkaConsumer的应用程序线程中运行（除了consumer向kafka集群发送的心跳，心跳在一个专门的单独线程中发送），所以我们调用KafkaConsumer的所有方法均需要保证在同一个线程中调用，除了KafkaConsumer.wakeup()方法，它设计用来通过其它线程向consumer线程发送信号，从而终止consumer执行。

跟producer一样，consumer要与kafka集群通信，消费kafka消息，首先需要获取消费的topic partition leader replica所在的broker地址等信息，这些信息可以通过向kafka集群任意broker发送Metadata请求消息获取。

我们知道，一个consumer group有多个consumer，一个topic有多个partition，而且topic的partition在同一时刻只能被consumer group内的一个consumer消费，那么consumer在消费partition消息前需要先确定消费topic的哪个partition。partition的分配通过group coordinator来实现。基本过程如下：

我们可以通过实现接口org.apache.kafka.clients.consumer.internals.PartitionAssignor自定义partition分配策略，但是kafka已经提供了三种分配策略可以直接使用。

partition分配完后，每个consumer知道了自己消费的topic partition，通过metadata请求可以获取相应partition的leader副本所在的broker信息，然后就可以向broker poll消息了。但是consumer从哪个offset开始poll消息？所以consumer在第一次向broker发送FetchRequest poll消息之前需要向Group Coordinator发送OffsetFetchRequest获取消费消息的起始位置。Group Coordinator会通过key {topic}-${partition}查询 __consumer_offsets topic中是否有offset的有效记录，如果存在，则将consumer所属consumer group最近已提交的offset返回给consumer。如果没有（可能是该partition是第一次分配给该consumer group消费，也可能是该partition长时间没有被该consumer group消费），则根据consumer配置参数auto.offset.reset值确定consumer消费的其实offset。如果auto.offset.reset值为latest，表示从partition的末尾开始消费，如果值为earliest，则从partition的起始位置开始消费。当然，consumer也可以随时通过KafkaConsumer.seek()方法人工设置消费的起始offset。

kafka broker在收到FetchRequest请求后，会使用请求中topic partition的offset查一个skiplist表（该表的节点key值是该partition每个LogSegment中第一条消息的offset值）确定消息所属的LogSegment，然后继续查LogSegment的稀疏索引表（存储在.index文件中），确定offset对应的消息在LogSegment文件中的位置。为了提升消息消费的效率，consumer通过参数fetch.min.bytes和max.partition.fetch.bytes告诉broker每次拉取的消息总的最小值和每个partition的最大值（consumer一次会拉取多个partition的消息）。当kafka中消息较少时，为了让broker及时将消息返回给consumer，consumer通过参数fetch.max.wait.ms告诉broker即使消息大小没有达到fetch.min.bytes值，在收到请求后最多等待fetch.max.wait.ms时间后，也将当前消息返回给consumer。fetch.min.bytes默认值为1MB，待fetch.max.wait.ms默认值为500ms。

为了提升消息的传输效率，kafka采用零拷贝技术让内核通过DMA把磁盘中的消息读出来直接发送到网络上。因为kafka写入消息时将消息写入内存中就返回了，如果consumer跟上了producer的写入速度，拉取消息时不需要读磁盘，直接从内存获取消息发送出去就可以了。

为了避免发生再平衡后，consumer重复拉取消息，consumer需要将已经消费完的消息的offset提交给group coordinator。这样发生再平衡后，consumer可以从上次已提交offset出继续拉取消息。

kafka提供了多种offset提交方式

partition offset提交和管理对kafka消息系统效率来说非常关键，它直接影响了再平衡后consumer是否会重复拉取消息以及重复拉取消息的数量。如果offset提交的比较频繁，会增加consumer和kafka broker的消息处理负载，降低消息处理效率；如果offset提交的间隔比较大，再平衡后重复拉取的消息就会比较多。还有比较重要的一点是，kafka只是简单的记录每次提交的offset值，把最后一次提交的offset值作为最新的已提交offset值，作为再平衡后消息的起始offset，而什么时候提交offset，每次提交的offset值具体是多少，kafka几乎不关心（这个offset对应的消息应该存储在kafka中，否则是无效的offset），所以应用程序可以先提交3000，然后提交2000，再平衡后从2000处开始消费，决定权完全在consumer这边。

kafka中的topic partition与consumer group中的consumer的消费关系其实是一种配对关系，当配对双方发生了变化时，kafka会进行再平衡，也就是重新确定这种配对关系，以提升系统效率、高可用性和伸缩性。当然，再平衡也会带来一些负面效果，比如在再平衡期间，consumer不能消费kafka消息，相当于这段时间内系统是不可用的。再平衡后，往往会出现消息的重复拉取和消费的现象。

触发再平衡的条件包括：

需要注意的是，kafka集群broker的增减或者topic partition leader重新选主这类集群状态的变化并不会触发在平衡

有两种情况与日常应用开发比较关系比较密切：

consumer在调用subscribe()方法时，支持传入一个ConsumerRebalanceListener监听器，ConsumerRebalanceListener提供了两个方法，onPartitionRevoked()方法在consumer停止消费之后，再平衡开始之前被执行。可以发现，这个地方是提交offset的好时机。onPartitonAssigned()方法则会在重新进行partition分配好了之后，但是新的consumer还未消费之前被执行。

我们在提到kafka时，首先想到的是它的吞吐量非常大，这也是很多人选择kafka作为消息传输组件的重要原因。

以下是保证kafka吞吐量大的一些设计考虑：

但是kafka是不是总是这么快？我们同时需要看到kafka为了追求快舍弃了一些特性：

所以，kafka在消息独立、允许少量消息丢失或重复、不关心消息顺序的场景下可以保证非常高的吞吐量，但是在需要考虑消息事务、严格保证消息顺序等场景下producer和consumer端需要进行复杂的考虑和处理，可能会比较大的降低kafka的吞吐量，例如对可靠性和保序要求比较高的控制类消息需要非常谨慎的权衡是否适合使用kafka。

我们通过producer向kafka集群发送消息，总是期望消息能被consumer成功消费到。最不能忍的是producer收到了kafka集群消息写入的正常响应，但是consumer仍然没有消费到消息。

kafka提供了一些机制来保证消息的可靠传递，但是有一些因素需要仔细权衡考虑，这些因素往往会影响kafka的吞吐量，需要在可靠性与吞吐量之间求得平衡：

kafka只保证partition消息顺序，不保证topic级别的顺序，而且保证的是partition写入顺序与读取顺序一致，不是业务端到端的保序。

如果对保序要求比较高，topic需要只设置一个partition。这时可以把参数max.in.flight.requests.per.connection设置为1，而retries设置为大于1的数。这样即使发生了可恢复型错误，仍然能保证消息顺序，但是如果发生不可恢复错误，应用层进行重试的话，就无法保序了。也可以采用同步发送的方式，但是这样也极大的降低了吞吐量。如果消息携带了表示顺序的字段，可以在接收端对消息进行重新排序以保证最终的有序。

为什么Kafka这么快

内存映射文件将磁盘上的文件内容与内存映射起来，我们往内存里写入数据，操作系统会在稍后把数据冲刷到磁盘上。所以，在写入数据时几乎就是写入内存的速度，这是Kafka快到飞起的另一个原因。

当Kafka客户端从服务器读取数据时，如果不使用零拷贝技术，那么大致需裂带要经历这样的一个过程：

从图中可以看到，数据在内核空间和用户空间之间穿梭了两次，那么能否避免这个多余的过程呢？当然可以，Kafka使用了零拷贝技术，也就是直接将数据从内核空间的读缓冲区直接拷贝到内核空间的socket缓冲区，然后再写入到NIC缓冲区，避免了在内核空间和用户空间之间穿行渣梭

除了利用底层的技术外，Kafka还在应用程序层面提供了一些手段来提升性能。最明显的就是使用批次。在向Kafka写入数据时，可以启用批次写入，这样可以避免在网络上频繁传输单个消息带来的延迟和带宽开销。假设网络带宽为10MB/S，一次性传输10MB的消息比传输1KB的消息10000万次显然要档源悄快得多。

Kafka零拷贝

用kafka做存储层，为什么呢？一大堆可以做数据厅返码存储的 MySQL、MongoDB、HDFS……

因为kafka数据是持久化磁盘的，还速度快；还可靠、支持分布式……

啥！用了磁盘，还速度快！！！

没错，kafka就是速度无敌，本文将探究kafka无敌性能背后的秘密。

首先要有个概念，kafka高性能的背后，是多方面协同后、最终的结果，kafka从宏观架构、分布式partition存储、ISR数据同步、以及“无孔不入”的高效利用磁盘/操作系统特性，这些多方面的协同，是kafka成为性能之王的必然结果。

本文将从kafka零拷贝，探究其是如何“无孔不入”的高效利用磁盘/操作系统特性的。

零拷贝并不是不需要拷贝，而是减少不必要的拷贝次数。通常是说在IO读写过程中。

实际上，零拷贝是有广义和狭义之分，目前我们通常听到的零拷贝，包括上面这个定义 减少不必要的拷贝次数 都是广义上的零拷贝。其实了解到这点就足够了。

我们知道，减少不必要的拷贝次数，就是为了提高效率。那零拷贝之前，是怎样的呢？

比如：读取文件，再用socket发送出去

传统方式实现：

先读取、再发送，实际经过1~4四次copy。

1、第一次：将磁盘文件，读取到操作系统内核缓冲区；

2、第二次：将内核缓冲区的数据，copy到application应用程序的buffer；

3、第三步：将application应用程序buffer中的数据，copy到socket网络发送缓冲区(属于操作系统内核的缓冲区)；

4、第四次：将socket buffer的数据，copy到网卡，由网卡进行网络传输。

传统方式，读取磁盘文件并进行网络发送，经过的四次数据copy是非常繁琐的。实际IO读写，需要进行IO中断，需要CPU响应中断(带来上下文切换)，尽管后来引入DMA来接管CPU的中断请求，但四次copy是存在“不必要的拷贝”的。

重新思考传统IO方式，会注意到实际上并不需要第二个和第三个数据副本。应用程序除了缓存数据并将其传输回套接字缓冲区之外什么都不做。相反，数据可以直接从读缓冲区传输到套接字缓冲区。

显然，第二次和第三次数据copy 其实在这种场景下没有什么帮助反而带来开销，这也正是零拷贝出现的意义。

这种场景 ：是指读取磁盘文件后，不需要做其他处理，直接用网络发送出去。试想，如果读取磁盘的数据需要用程序进一步处理的话，必须要经过第二次和第三次数据copy，让应用程序在内存缓冲区处理。

kafka作为扮哪MQ也好，作为存储层也好，无非是两个重要功能，一是Producer生产的数据存到broker，二是 Consumer从broker读取数据；我们把它简化成如下两个过程：

1、网络数据持久化到磁盘 (Producer 到 Broker)

2、磁盘文件通过网络发送（Broker 到 Consumer）

下面，先给出“kafka用了磁盘，还速度快”的结论

1、顺序读写

磁盘顺序读或写的速度400M/s，能够发挥磁盘最大的速度。

随机读写，磁盘速度慢的时候十几到几百K/s。这就看出了差距。

kafka将来自Producer的数据，顺序追加在partition，partition就是一个文件，以此实现顺序写入。

Consumer从broker读取数据时，因为自带了偏移量，接着上次读取的位置继续读，以此实现顺序读。

顺序读写，是kafka利用磁盘特性的一个重要体现。

2、零拷贝 sendfile(in,out)

数据直接在内核完成输入和输出，不需世旦要拷贝到用户空间再写出去。

kafka数据写入磁盘前，数据先写到进程的内存空间。

3、mmap文件映射

虚拟映射只支持文件；

在进程 的非堆内存开辟一块内存空间，和OS内核空间的一块内存进行映射，

kafka数据写入、是写入这块内存空间，但实际这块内存和OS内核内存有映射，也就是相当于写在内核内存空间了，且这块内核空间、内核直接能够访问到，直接落入磁盘。

这里，我们需要清楚的是：内核缓冲区的数据，flush就能完成落盘。

我们来重点探究 kafka两个重要过程、以及是如何利用两个零拷贝技术sendfile和mmap的。

传统方式实现：

先接收生产者发来的消息，再落入磁盘。

实际会经过四次copy，如下图的四个箭头。

数据落盘通常都是非实时的，kafka生产者数据持久化也是如此。Kafka的数据 并不是实时的写入硬盘 ，它充分利用了现代操作系统分页存储来利用内存提高I/O效率。

对于kafka来说，Producer生产的数据存到broker，这个过程读取到socket buffer的网络数据，其实可以直接在OS内核缓冲区，完成落盘。并没有必要将socket buffer的网络数据，读取到应用进程缓冲区；在这里应用进程缓冲区其实就是broker，broker收到生产者的数据，就是为了持久化。

在此 特殊场景 下：接收来自socket buffer的网络数据，应用进程不需要中间处理、直接进行持久化时。——可以使用mmap内存文件映射。

简称mmap，简单描述其作用就是：将磁盘文件映射到内存, 用户通过修改内存就能修改磁盘文件。

它的工作原理是直接利用操作系统的Page来实现文件到物理内存的直接映射。完成映射之后你对物理内存的操作会被同步到硬盘上（操作系统在适当的时候）。

通过mmap，进程像读写硬盘一样读写内存（当然是虚拟机内存），也不必关心内存的大小有虚拟内存为我们兜底。

使用这种方式可以获取很大的I/O提升，省去了用户空间到内核空间复制的开销。

mmap也有一个很明显的缺陷——不可靠，写到mmap中的数据并没有被真正的写到硬盘，操作系统会在程序主动调用flush的时候才把数据真正的写到硬盘。Kafka提供了一个参数——producer.type来控制是不是主动flush；如果Kafka写入到mmap之后就立即flush然后再返回Producer叫同步(sync)；写入mmap之后立即返回Producer不调用flush叫异步(async)。

Java NIO，提供了一个 MappedByteBuffer 类可以用来实现内存映射。

MappedByteBuffer只能通过调用FileChannel的map()取得，再没有其他方式。

FileChannel.map()是抽象方法，具体实现是在 FileChannelImpl.c 可自行查看JDK源码，其map0()方法就是调用了Linux内核的mmap的API。

使用 MappedByteBuffer类要注意的是：mmap的文件映射，在full gc时才会进行释放。当close时，需要手动清除内存映射文件，可以反射调用sun.misc.Cleaner方法。

传统方式实现：

先读取磁盘、再用socket发送，实际也是进过四次copy。

而 Linux 2.4+ 内核通过 sendfile 系统调用，提供了零拷贝。磁盘数据通过 DMA 拷贝到内核态 Buffer 后，直接通过 DMA 拷贝到 NIC Buffer(socket buffer)，无需 CPU 拷贝。这也是零拷贝这一说法的来源。除了减少数据拷贝外，因为整个读文件 - 网络发送由一个 sendfile 调用完成，整个过程只有两次上下文切换，因此大大提高了性能。零拷贝过程如下图所示。

相比于文章开始，对传统IO 4步拷贝的分析，sendfile将第二次、第三次拷贝，一步完成。

其实这项零拷贝技术，直接从内核空间（DMA的）到内核空间（Socket的)、然后发送网卡。

应用的场景非常多，如Tomcat、Nginx、Apache等web服务器返回静态资源等，将数据用网络发送出去，都运用了sendfile。

简单理解 sendfile(in,out)就是，磁盘文件读取到操作系统内核缓冲区后、直接扔给网卡，发送网络数据。

Java NIO对sendfile的支持 就是FileChannel.transferTo()/transferFrom()。

fileChannel.transferTo( position, count, socketChannel);

把磁盘文件读取OS内核缓冲区后的fileChannel，直接转给socketChannel发送；底层就是sendfile。消费者从broker读取数据，就是由此实现。

具体来看，Kafka 的数据传输通过 TransportLayer 来完成，其子类 PlaintextTransportLayer 通过Java NIO 的 FileChannel 的 transferTo 和 transferFrom 方法实现零拷贝。

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

总的来说Kafka快的原因：

1、partition顺序读写，充分利用磁盘特性，这是基础；

2、Producer生产的数据持久化到broker，采用mmap文件映射，实现顺序的快速写入；

3、Customer从broker读取数据，采用sendfile，将磁盘文件读到OS内核缓冲区后，直接转到socket buffer进行网络发送。

1、都是Linux内核提供、实现零拷贝的API；

2、sendfile 是将读到内核空间的数据，转到socket buffer，进行网络发送；

3、mmap将磁盘文件映射到内存，支持读和写，对内存的操作会反映在磁盘文件上。

RocketMQ 在消费消息时，使用了 mmap。kafka 使用了 sendFile。

关于kafka零拷贝和kafka零拷贝 面试的介绍到此就结束了，不知道你从中找到你需要的信息了吗 ？如果你还想了解更多这方面的信息，记得收藏关注本站。