kafka原理（这次我只想把Kafka原理讲清楚）

这可能是全网Java学习路线最完整，最详细的版本了，没有之一

前言：

1.由来

为什么使用消息队列？

从系统之间有通信需求开始，就自然产生了消息队列。

在计算机科学中，消息队列（英语：Message queue）是一种进程间通信或同一进程的不同线程间的通信方式，软件的贮列用来处理一系列的输入，通常是来自用户。消息队列提供了异步的通信协议，每一个贮列中的纪录包含详细说明的资料，包含发生的时间，输入设备的种类，以及特定的输入参数，也就是说：消息的发送者和接收者不需要同时与消息队列交互。消息会保存在队列中，直到接收者取回它。

2.特点

解耦：允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。冗余：消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的”插入-获取-删除”范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。扩展性：因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。灵活性 & 峰值处理能力：在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。可恢复性：系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。顺序保证：在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。（Kafka 保证一个 Partition 内的消息的有序性）缓冲：有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。异步通信：很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

3.使用场景

服务解耦：下游系统可能只需要当前系统的一个子集，应对不断增加变化的下游系统，当前系统不停地修改调试与这些下游系统的接口，系统间耦合过于紧密。引入消息队列后，当前系统变化时发送一条消息到消息队列的一个主题中，所有下游系统都订阅主题，这样每个下游系统都可以获得一份实时完整的订单数据。异步处理：以秒杀为例：风险控制->库存锁定->生成订单->短信通知->更新统计数据

限流削峰/流量控制一个设计健壮的程序有自我保护的能力，也就是说，它应该可以在海量的请求下，还能在自身能力范围内尽可能多地处理请求，拒绝处理不了的请求并且保证自身运行正常。使用消息队列隔离网关和后端服务，以达到流量控制和保护后端服务的目的。

消息驱动的系统系统分为消息队列、消息生产者、消息消费者，生产者负责产生消息，消费者(可能有多个)负责对消息进行处理；(分工处理(各自对应相应的队列)，灵活应用(收到就处理/定时处理))

4.两种模式

点对点：系统 A 发送的消息只能被系统 B 接收，其他任何系统都不能读取 A 发送的消息。日常生活的例子比如电话客服就属于这种模型：同一个客户呼入电话只能被一位客服人员处理，第二个客服人员不能为该客户服务。

发布/订阅模型这个模型可能存在多个发布者向相同的主题发送消息，而订阅者也可能存在多个，它们都能接收到相同主题的消息。生活中的报纸订阅就是一种典型的发布 / 订阅模型。

1.名词解释

kafka是一个分布式流处理平台。

类似一个消息系统，读写流式的数据编写可扩展的流处理应用程序，用于实时事件响应的场景安全的将流式的数据存储在一个分布式，有副本备份，容错的集群

2.历史由来

Kafka从何而来?我们为什么要开发Kafka? Kafka到底是什么? Kafka 最初是 LinkedIn 的一个内部基础设施系统。我们发现虽然有很多数据库和系统可以用来存储数据，但在我们的架构里，刚好缺一个可以帮助处理持续数据流的组件。在开发Kafka之前，我们实验了各种现成的解决方案，从消息系统到日志聚合系统，再到ETL工具，它们都无法满足我们的需求。 最后，我们决定从头开发一个系统。我们不想只是开发一个能够存储数据的系统，比如传统的关系型数据库、键值存储引擎、搜索引擎或缓存系统，我们希望能够把数据看成是持续变化和不断增长的流，并基于这样的想法构建出一个数据系统。事实上，是一个数据架构。这个想法实现后比我们最初预想的适用性更广。Kafka 一开始被用在社交网络的实时应用和数据流当中，而现在已经成为下一代数据架构的基础。大型零售商正在基于持续数据流改造他们的基础业务流程，汽车公司正在从互联网汽车那里收集和处理实时数据流，银行也在重新思考基于 Kafka 改造他们的基础。

它可以用于两大类别的应用:

构造实时流数据管道，它可以在系统或应用之间可靠地获取数据。(相当于message queue)构建实时流式应用程序，对这些流数据进行转换或者影响。(就是流处理，通过kafka stream topic和topic之间内部进行变化)

版本号

备注

0.7

上古版本，提供了最基础的消息队列功能

0.8

引入了副本机制，成为了一个真正意义上完备的分布式高可靠消息队列解决方案

0.8.2

新版本 Producer API，即需要指定 Broker 地址的 Producer

0.9

增加了基础的安全认证 / 权限，Java 重写了新版本消费者 API

0.10.0.0

引入了 Kafka Streams

0.11.0.0

提供幂等性 Producer API 以及事务（Transaction） API，对 Kafka 消息格式做了重构。

1.0

Kafka Streams 的各种改进

2.0

Kafka Streams 的各种改进

3.术语

消息：Record。这里的消息就是指 Kafka 处理的主要对象。服务：Broker。一个 Kafka 集群由多个 Broker 组成，Broker 负责接收和处理客户端发送过来的请求，以及对消息进行持久化。主题：Topic。主题是承载消息的逻辑容器，在实际使用中多用来区分具体的业务。分区：Partition。一个有序不变的消息序列。每个主题下可以有多个分区。消息位移：Offset。表示分区中每条消息的位置信息，是一个单调递增且不变的值。副本：Replica。Kafka 中同一条消息能够被拷贝到多个地方以提供数据冗余，这些地方就是所谓的副本。副本还分为领导者副本和追随者副本，各自有不同的角色划分。副本是在分区层级下的，即每个分区可配置多个副本实现高可用。生产者：Producer。向主题发布新消息的应用程序。消费者：Consumer。从主题订阅新消息的应用程序。消费者位移：Consumer Offset。表征消费者消费进度，每个消费者都有自己的消费者位移。消费者组：Consumer Group。多个消费者实例共同组成的一个组，同时消费多个分区以实现高吞吐。重平衡：Rebalance。消费者组内某个消费者实例挂掉后，其他消费者实例自动重新分配订阅主题分区的过程。Rebalance 是 Kafka 消费者端实现高可用的重要手段。

4.Topic

日志

日志可能是一种最简单的不能再简单的存储抽象，只能追加、按照时间完全有序（totally-ordered）的记录序列。日志看起来的样子：

在日志的末尾添加记录，读取日志记录则从左到右。每一条记录都指定了一个唯一的顺序的日志记录编号。

日志记录的次序（ordering）定义了『时间』概念，因为位于左边的日志记录表示比右边的要早。日志记录编号可以看作是这条日志记录的『时间戳』。把次序直接看成是时间概念，刚开始你会觉得有点怪异，但是这样的做法有个便利的性质：解耦了时间和任一特定的物理时钟（physical clock）。引入分布式系统后，这会成为一个必不可少的性质。

日志和文件或数据表（table）并没有什么大的不同。文件是一系列字节，表是由一系列记录组成，而日志实际上只是一种按照时间顺序存储记录的数据表或文件。

对于每一个topic， Kafka集群都会维持一个分区日志，如下所示：

实操

启动zk

cd /usr/local/kara/kafka_2.13-2.6.0/bin zookeeper-server-start.sh ../config/zookeeper.properties

启动kafka服务器

kafka-server-start.sh ../config/server.properties

创建topic，4个分区，一个副本

kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 4 --topic partition_test

发送一些消息

kafka-console-producer.sh --broker-list localhost:9092 --topic partition_test

启动一个consumer

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic partition_test --from-beginning

分区

partition存储分布

一个topic下有多个不同partition，每个partition为一个目录，partiton命名规则为topic名称+有序序号，第一个partiton序号从0开始，序号最大值为partitions数量减1

partition文件存储

1.每个partion(目录)相当于一个巨型文件被平均分配到多个大小相等segment(段)数据文件中。但每个段segment file消息数量不一定相等，这种特性方便old segment file快速被删除。

2.每个partiton只需要支持顺序读写就行了，segment文件生命周期由服务端配置参数决定。

segment文件存储

1.segment file组成：由2大部分组成，分别为index file和data file，此2个文件一一对应，成对出现，分别表示为segment索引文件、数据文件.

2.segment文件命名规则：partion全局的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset值。数值最大为64位long大小，19位数字字符长度，没有数字用0填充。

segment index file采取稀疏索引存储方式，它减少索引文件大小，通过mmap可以直接内存操作，稀疏索引为数据文件的每个对应message设置一个元数据指针,它比稠密索引节省了更多的存储空间，但查找起来需要消耗更多的时间。

3.segment中的消息message物理结构字段说明

关键字

解释说明

8 byte offset

在parition(分区)内的每条消息都有一个有序的id号，这个id号被称为偏移(offset),它可以唯一确定每条消息在parition(分区)内的位置。即offset表示partiion的第多少message

4 byte message size

message大小

4 byte CRC32

用crc32校验message

1 byte “magic”

表示本次发布Kafka服务程序协议版本号

1 byte “attributes”

表示为独立版本、或标识压缩类型、或编码类型。

4 byte key length

表示key的长度,当key为-1时，K byte key字段不填

K byte key

可选

value bytes payload

表示实际消息数据。

文件系统

Kafka 对消息的存储和缓存严重依赖于文件系统。人们对于“磁盘速度慢”具有普遍印象，事实上，磁盘的速度比人们预期的要慢的多，也快得多，这取决于人们使用磁盘的方式。

使用6个7200rpm、SATA接口、RAID-5的磁盘阵列在JBOD配置下的顺序写入的性能约为600MB/秒，但随机写入的性能仅约为100k/秒，相差6000倍以上。

线性的读取和写入是磁盘使用模式中最有规律的，并且由操作系统进行了大量的优化。

read-ahead 是以大的 data block 为单位预先读取数据write-behind 是将多个小型的逻辑写合并成一次大型的物理磁盘写入

关于该问题的进一步讨论可以参考 ACM Queue article，他们发现实际上顺序磁盘访问在某些情况下比随机内存访问还要快！

为了弥补这种性能差异，现代操作系统主动将所有空闲内存用作 disk caching（磁盘高速缓存），所有对磁盘的读写操作都会通过这个统一的 cache（ in-process cache）。

即使进程维护了 in-process cache，该数据也可能会被复制到操作系统的 pagecache 中，事实上所有内容都被存储了两份。

此外，Kafka 建立在 JVM 之上，任何了解 Java 内存使用的人都知道两点：

对象的内存开销非常高，通常是所存储的数据的两倍(甚至更多)。随着堆中数据的增加，Java 的垃圾回收变得越来越复杂和缓慢。

kafka选择了一个非常简单的设计：相比于维护尽可能多的 in-memory cache，并且在空间不足的时候匆忙将数据 flush 到文件系统，我们把这个过程倒过来。所有数据一开始就被写入到文件系统的持久化日志中，而不用在 cache 空间不足的时候 flush 到磁盘。实际上，这表明数据被转移到了内核的 pagecache 中。

Pagecache页面缓存

Page cache（页面缓存）Page cache 也叫页缓冲或文件缓冲，是由好几个磁盘块构成，大小通常为4k，在64位系统上为8k，构成的几个磁盘块在物理磁盘上不一定连续，文件的组织单位为一页，也就是一个page cache大小，文件读取是由外存上不连续的几个磁盘块，到buffer cache，然后组成page cache，然后供给应用程序。Buffer cache（块缓存）Buffer cache 也叫块缓冲，是对物理磁盘上的一个磁盘块进行的缓冲，其大小为通常为1k，磁盘块也是磁盘的组织单位。设立buffer cache的目的是为在程序多次访问同一磁盘块时，减少访问时间。Page cache（页面缓存）与Buffer cache（块缓存）的区别磁盘的操作有逻辑级（文件系统）和物理级（磁盘块），这两种Cache就是分别缓存逻辑和物理级数据的。我们通过文件系统操作文件，那么文件将被缓存到Page Cache，如果需要刷新文件的时候，Page Cache将交给Buffer Cache去完成，因为Buffer Cache就是缓存磁盘块的。简单说来，page cache用来缓存文件数据，buffer cache用来缓存磁盘数据。在有文件系统的情况下，对文件操作，那么数据会缓存到page cache，如果直接采用dd等工具对磁盘进行读写，那么数据会缓存到buffer cache。Buffer(Buffer Cache)以块形式缓冲了块设备的操作，定时或手动的同步到硬盘，它是为了缓冲写操作然后一次性将很多改动写入硬盘，避免频繁写硬盘，提高写入效率。Cache(Page Cache)以页面形式缓存了文件系统的文件，给需要使用的程序读取，它是为了给读操作提供缓冲，避免频繁读硬盘，提高读取效率。

降低时间复杂度

消息系统使用的持久化数据结构通常是和 BTree 相关联的消费者队列或者其他用于存储消息源数据的通用随机访问数据结构。BTree 的操作复杂度是 O(log N)，通常我们认为 O(log N) 基本等同于常数时间，但这条在磁盘操作中不成立。

存储系统将非常快的cache操作和非常慢的物理磁盘操作混合在一起，当数据随着 fixed cache 增加时，可以看到树的性能通常是非线性的——比如数据翻倍时性能下降不只两倍。

kafka选择把持久化队列建立在简单的读取和向文件后追加两种操作之上，这和日志解决方案相同。这种架构的优点在于所有的操作复杂度都是O(1)，而且读操作不会阻塞写操作，读操作之间也不会互相影响。

在不产生任何性能损失的情况下能够访问几乎无限的硬盘空间，Kafka 可以让消息保留相对较长的一段时间(比如一周)，而不是试图在被消费后立即删除。

降低大量小型IO操作的影响

小型的 I/O 操作发生在客户端和服务端之间以及服务端自身的持久化操作中。

为了避免这种情况，kafka的协议是建立在一个 “消息块” 的抽象基础上，合理将消息分组。将多个消息打包成一组，而不是每次发送一条消息，从而使整组消息分担网络中往返的开销。

这个简单的优化对速度有着数量级的提升。批处理允许更大的网络数据包，更大的顺序读写磁盘操作，连续的内存块等等，所有这些都使 KafKa 将随机流消息顺序写入到磁盘，再由 consumers 进行消费。

零拷贝

字节拷贝是低效率的操作，在消息量少的时候没啥问题，但是在高负载的情况下，影响就不容忽视。为了避免这种情况，kafka使用 producer ，broker 和 consumer 都共享的标准化的二进制消息格式，这样数据块不用修改就能在他们之间传递。

保持这种通用格式可以对一些很重要的操作进行优化: 持久化日志块的网络传输。现代的unix 操作系统提供了一个高度优化的编码方式，用于将数据从 pagecache 转移到 socket 网络连接中；在 Linux 中系统调用 sendfile 做到这一点。

传统IO (4次上下文切换4次拷贝)假如将磁盘上的文件读取出来，然后通过网络协议发送给客户端。一般需要两个系统调用，但是一共4次上下文切换，4次拷贝read(file, tmp_buf, len);write(socket, tmp_buf, len);

要想提高文件传输的性能，就需要减少「用户态与内核态的上下文切换」和「内存拷贝」的次数。mmap(4次上下文切换3次拷贝)mmap()系统调用函数会直接把内核缓冲区里的数据「映射」到用户空间，这样，操作系统内核与用户空间就不需要再进行任何的数据拷贝操作，它替换了read()系统调用函数。buf = mmap(file, len);write(sockfd, buf, len);

sendfile（2次上下文切换3次拷贝）Linux 内核版本 2.1 中，提供了一个专门发送文件的系统调用函数 sendfile()首先，它可以替代前面的 read()和 write()这两个系统调用，这样就可以减少一次系统调用，也就减少了 2 次上下文切换的开销。其次，该系统调用，可以直接把内核缓冲区里的数据拷贝到 socket 缓冲区里，不再拷贝到用户态，这样就只有 2 次上下文切换，和 3 次数据拷贝。#include <sys/socket.h>ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count);它的前两个参数分别是目的端和源端的文件描述符，后面两个参数是源端的偏移量和复制数据的长度，返回值是实际复制数据的长度。

零拷贝（2次上下文切换2次拷贝）Linux 内核 2.4 版本开始起，对于支持网卡支持 SG-DMA 技术的情况下， sendfile() 系统调用的过程发生了点变化，具体过程如下：

第一步，通过 DMA 将磁盘上的数据拷贝到内核缓冲区里；第二步，缓冲区描述符和数据长度传到 socket 缓冲区，这样网卡的 SG-DMA 控制器就可以直接将内核缓存中的数据拷贝到网卡的缓冲区里，此过程不需要将数据从操作系统内核缓冲区拷贝到 socket 缓冲区中，这样就减少了一次数据拷贝；

kafka高效文件存储设计特点

Kafka把topic中一个parition大文件分成多个小文件段，通过多个小文件段，就容易定期清除或删除已经消费完文件，减少磁盘占用。通过索引信息可以快速定位message和确定response的最大大小。通过index元数据全部映射到memory，可以避免segment file的IO磁盘操作。通过索引文件稀疏存储，可以大幅降低index文件元数据占用空间大小。

5.Broker

broker端处理请求的流程

请求处理的方案：

顺序处理（同步）while(true){handle(request);}吞吐量查，只适用于请求发送不频繁的系统多线程处理（异步）while(true){() -> {handle(request);}.start();}不会阻塞下一个请求，但是开销很大Reactor模式Reactor 模式是事件驱动架构的一种实现方式，特别适合应用于处理多个客户端并发向服务器端发送请求的场景.

该模式有个请求分发线程 Dispatcher，也就是图中的 Acceptor，它会使用轮询方式将不同的请求下发到多个工作线程中处理。Acceptor 线程只是用于请求分发，不涉及具体的逻辑处理，非常得轻量级，因此有很高的吞吐量表现。kafka使用的就是Reactor模式kafka所有的请求都是通过TCP网络以Socket的方式进行通讯的，它也有对应的 Acceptor 线程和一个工作线程池，在 Kafka 中，这个工作线程池叫网络线程池。

Broker 端参数 num.network.threads，用于调整该网络线程池的线程数。默认3个，专门处理客户端发送的请求。# The number of threads that the server uses for receiving requests from the network and sending responses to the networknum.network.threads=3客户端发来的请求会被 Broker 端的 Acceptor 线程分发到任意一个网络线程中，由它们来进行处理。那么，当网络线程接收到请求后，它是如何处理的？Kafka 在这个环节又做了一层异步线程池的处理。

当网络线程拿到请求后，它不是自己处理，而是将请求放入到一个共享请求队列中。Broker 端还有个 IO 线程池，负责从该队列中取出请求，执行真正的处理。如果是 PRODUCE 生产请求，则将消息写入到底层的磁盘日志中；如果是 FETCH 请求，则从磁盘或页缓存中读取消息。Broker 端参数num.io.threads，用于配置IO线程池中的线程数，默认创建8个IO线程处理请求。# The number of threads that the server uses for processing requests, which may include disk I/Onum.io.threads=8请求队列是所有网络线程共享的，而响应队列则是每个网络线程专属的，因为Dispatcher只用于请求的分发而不负责响应回传，因此只能让每个网络线程自己发送Response给客户端。还有一个Purgatory组件，炼狱组件用于它是用来缓存延时请求（Delayed Request）的。所谓延时请求，就是那些一时未满足条件不能立刻处理的请求。比如设置了 acks=all 的 PRODUCE 请求，一旦设置了 acks=all，那么该请求就必须等待 ISR 中所有副本都接收了消息后才能返回，此时处理该请求的 IO 线程就必须等待其他 Broker 的写入结果。当请求不能立刻处理时，它就会暂存在 Purgatory 中。稍后一旦满足了完成条件，IO 线程会继续处理该请求，并将 Response 放入对应网络线程的响应队列中。

replication副本机制/备份机制

所谓副本，通产指分布式系统中不同机器中保存相同的数据拷贝，本质就是一个只能追加写消息的提交日志提供数据冗余。即使系统部分组件失效，系统依然能够继续运转，因而增加了整体可用性以及数据持久性。提供高伸缩性。支持横向扩展，能够通过增加机器的方式来提升读性能，进而提高读操作吞吐量。改善数据局部性。允许将数据放入与用户地理位置相近的地方，从而降低系统延时。假如kafka一个主题有3个分区3个副本部署在三台Broker上，在实际生产环境中，每台 Broker 都可能保存有各个主题下不同分区的不同副本，因此，单个 Broker 上存有成百上千个副本的现象是非常正常的。

关键性问题：如何保证副本中的数据是一致的？

kafka采用了常见的解决方案：基于领导者（Leader-based）的副本机制

每个分区在创建时都要选举一个副本，称为领导者副本(Leader Replica)，其余的副本自动称为追随者副本(Follower Replica)。所有的请求都必须由领导者副本(Leader Replica)来处理。追随者副本不处理客户端请求，它唯一的任务就是从领导者副本异步拉取消息，并写入到自己的提交日志中，从而实现与领导者副本的同步（如何实现同步？）当领导者副本(Leader Replica)挂了，开启新一轮的领导者选举，依托zookeeper，先到先得

这样做的优缺点：

没能提供读操作横向扩展以及改善局部性。方面实现立即读Read-your-writes，写完就能读到方面实现单调读Monotonic Reads，同一消费者，多次消费消息，不会看到某条消息一会存在一会儿不存在

同步副本集合In-sync Replicas(ISR)

如何定义同步，追随者副本到底在什么条件下才算与 Leader 同步。kafka引入了同步副本ISR集合，ISR 中的副本都是与 Leader 同步的副本，凡是在ISR集合中的副本都认为是同步的。

上图中哪些follower副本是同步的，kafka it depends，视情况而定，不一定是严格的相同Broker参数replica.lag.time.max.ms ，默认10秒钟如果一个follower在这个时间内没有发送fetch请求或消费leader日志到结束的offset，leader将从ISR中移除这个follower，并认为这个follower已经挂了倘若该副本后面慢慢地追上了 Leader 的进度，那么它是能够重新被加回 ISR 的。这也表明，ISR 是一个动态调整的集合，而非静态不变的。

如果ISR为空该如何选举

因为 Leader 副本天然就在 ISR 中，如果 ISR 为空了，就说明 Leader 副本也挂掉了，此时该怎么选举新 Leader 。

Kafka 把所有不在 ISR 中的存活副本都称为非同步副本。通常来说，非同步副本落后 Leader 太多，如果选择这些副本作为新 Leader，就可能出现数据丢失的情况。在 Kafka 中，选举这种副本的过程称为 Unclean 领导者选举。

Broker 端参数 unclean.leader.election.enable 控制是否允许 Unclean 领导者选举。

如果想保证数据的一致性，禁用Unclean。

一个分布式系统通常只能同时满足一致性（Consistency）、可用性（Availability）、分区容错性（Partition tolerance）中的两个。显然，在这个问题上，Kafka 赋予你选择 C 或 A 的权利。

ISR副本同步机制

与 Leader 副本保持同步的条件有两个：

1.该远程 Follower 副本在 ISR 中。

2.该远程 Follower 副本 LEO 值落后于 Leader 副本 LEO 值的时间，不超过 Broker 端参数 replica.lag.time.max.ms 的值。如果使用默认值的话，就是不超过 10 秒。

Committed Message：已提交消息，已经被所有ISR同步的消息。Lagging Message：未提交消息，没有达到所有ISR同步的消息。Hight Watermark：副本高水位值，表示分区中最新一条已提交(Committed)的消息的Offset。位移值等于高水位的消息也属于未提交消息。也就是说，高水位上的消息是不能被消费者消费的。LEO：Log End Offset，Leader中最新消息的Offset，它表示副本写入下一条消息的位移值。

高水位和 LEO 是副本对象的两个重要属性。Kafka 所有副本都有对应的高水位和 LEO 值，而不仅仅是 Leader 副本。只不过 Leader 副本比较特殊，Kafka 使用 Leader 副本的高水位来定义所在分区的高水位。换句话说，分区的高水位就是其 Leader 副本的高水位。

高水位更新机制

Kafka 所有副本都有对应的高水位和 LEO 值，在 Leader 副本所在的 Broker 上，还保存了其他 Follower 副本的 LEO 值。

leader副本更新HW和LEO流程处理生产者请求的逻辑如下：获取 Leader 副本所在 Broker 端保存的所有远程副本 LEO 值{LEO-1，LEO-2，……，LEO-n}。获取 Leader 副本高水位值：currentHW。更新 currentHW = min(currentHW, LEO-1，LEO-2，……，LEO-n)。写入消息到本地磁盘。更新分区高水位值。

处理 Follower 副本拉取消息的逻辑如下：

读取磁盘（或页缓存）中的消息数据。使用 Follower 副本发送请求中的位移值更新远程副本 LEO 值。更新分区高水位值（具体步骤与处理生产者请求的步骤相同）Follower 副本更新HW和LEO流程从 Leader 拉取消息的处理逻辑如下：获取 Leader 发送的高水位值：currentHW。获取步骤 2 中更新过的 LEO 值：currentLEO。更新高水位为 min(currentHW, currentLEO)。写入消息到本地磁盘。更新 LEO 值。更新高水位值。

副本同步机制

当生产者发送一条消息时，Leader 和 Follower 副本对应的高水位是怎么被更新的。

初始初始状态时，所有值都是 0。当生产者给主题分区发送一条消息后，状态变更为：

此时，Leader 副本成功将消息写入了本地磁盘，故 LEO 值被更新为 1。

Follower 再次尝试从 Leader 拉取消息，拉取到消息，Follower 副本也成功地更新 LEO 为 1。

Leader 和 Follower 副本的 LEO 都是 1，但各自的高水位依然是 0，还没有被更新。它们需要在下一轮的拉取中被更新

在新一轮的拉取请求中，Follower 副本这次请求拉取的是位移值 =1 的消息。Leader 副本接收到此请求后，更新远程副本 LEO 为 1，然后更新 Leader 高水位为 1。做完这些之后，它会将当前已更新过的高水位值 1 发送给 Follower 副本。Follower 副本接收到以后，也将自己的高水位值更新成 1。至此，一次完整的消息同步周期就结束了。

Leader Epoch

Leader 副本高水位更新和 Follower 副本高水位更新在时间上是存在错配的。这种错配是很多“数据丢失”或“数据不一致”问题的根源。0.11 版本正式引入了 Leader Epoch，来规避因高水位更新错配导致的各种不一致问题。

Leader Epoch由两部分数据组成:

Epoch。一个单调增加的版本号。每当副本领导权发生变更时，都会增加该版本号。小版本号的 Leader 被认为是过期 Leader，不能再行使 Leader 权力。起始位移（Start Offset）。Leader 副本在该 Epoch 值上写入的首条消息的位移。0 01 120Kafka Broker 会在内存中为每个分区都缓存 Leader Epoch 数据，同时它还会定期地将这些信息持久化到一个 checkpoint 文件中

leader epoch 具体的工作机制如下：

1.当副本成为 leader 时：

这时，如果此时生产者有新消息发送过来，会首先新的 leader epoch 以及 LEO 添加到 leader-epoch-checkpoint 文件中。

2.当副本变成 follower 时：

发送 LeaderEpochRequest 请求给 leader 副本，该请求包括了 follower 中最新的 epoch 版本；leader 返回给 follower 的相应中包含了一个 LastOffset，如果 follower last epoch = leader last epoch，则 LastOffset = leader LEO，否则取大于 follower last epoch 中最小的 leader epoch 的 start offset 值，举个例子：假设 follower last epoch = 1，此时 leader 有 (1, 20) (2, 80) (3, 120)，则 LastOffset = 80；follower 拿到 LastOffset 之后，会对比当前 LEO 值是否大于 LastOffset，如果当前 LEO 大于 LastOffset，则从 LastOffset 截断日志；follower 开始发送 fetch 请求给 leader 保持消息同步。

数据丢失场景

开始时，副本 A 和副本 B 都处于正常状态，A 是 Leader 副本。某个使用了默认 acks 设置的生产者程序向 A 发送了两条消息，A 全部写入成功，此时 Kafka 会通知生产者说两条消息全部发送成功。

现在我们假设 Leader 和 Follower 都写入了这两条消息，而且 Leader 副本的高水位也已经更新了，但 Follower 副本高水位还未更新——这是可能出现的。还记得吧，Follower 端高水位的更新与 Leader 端有时间错配。倘若此时副本 B 所在的 Broker 宕机，当它重启回来后，副本 B 会执行日志截断操作，将 LEO 值调整为之前的高水位值，也就是 1。这就是说，位移值为 1 的那条消息被副本 B 从磁盘中删除，此时副本 B 的底层磁盘文件中只保存有 1 条消息，即位移值为 0 的那条消息。

当执行完截断操作后，副本 B 开始从 A 拉取消息，执行正常的消息同步。如果就在这个节骨眼上，副本 A 所在的 Broker 宕机了，那么 Kafka 就别无选择，只能让副本 B 成为新的 Leader，此时，当 A 回来后，需要执行相同的日志截断操作，即将高水位调整为与 B 相同的值，也就是 1。这样操作之后，位移值为 1 的那条消息就从这两个副本中被永远地抹掉了。这就是这张图要展示的数据丢失场景。

Leader Epoch 如何规避这种数据丢失

引用 Leader Epoch 机制后，Follower 副本 B 重启回来后，需要向 A 发送一个特殊的请求去获取 Leader 的 LEO 值。在这个例子中，该值为 2。当获知到 Leader LEO=2 后，B 发现该 LEO 值不比它自己的 LEO 值小，而且缓存中也没有保存任何起始位移值 > 2 的 Epoch 条目，因此 B 无需执行任何日志截断操作。这是对高水位机制的一个明显改进，即副本是否执行日志截断不再依赖于高水位进行判断。

现在，副本 A 宕机了，B 成为 Leader。同样地，当 A 重启回来后，执行与 B 相同的逻辑判断，发现也不用执行日志截断，至此位移值为 1 的那条消息在两个副本中均得到保留。后面当生产者程序向 B 写入新消息时，副本 B 所在的 Broker 缓存中，会生成新的 Leader Epoch 条目：[Epoch=1, Offset=2]。之后，副本 B 会使用这个条目帮助判断后续是否执行日志截断操作。这样，通过 Leader Epoch 机制，Kafka 完美地规避了这种数据丢失场景。

6.Producer

生产者消息分区机制

分区的作用就是提供负载均衡的能力，或者为了实现系统的高伸缩性（Scalability）

每个节点的机器都能独立地执行各自分区的读写请求处理；还可以通过添加新的节点机器来增加整体系统的吞吐量

分区策略是决定生产者将消息发送到哪个分区的算法

轮询策略（Round-robin）

Kafka Java 生产者 API 默认提供的分区策略，未指定partitioner.class参数则默认。轮询策略有非常优秀的负载均衡表现，它总是能保证消息最大限度地被平均分配到所有分区上，故默认情况下它是最合理的分区策略，也是我们最常用的分区策略之一。随机策略（Randomness）

老版本生产者使用的随机策略，新版的已改为轮询根据Key分区Key 的作用非常大，可以是一个有着明确业务含义的字符串，比如客户代码、部门编号或是业务 ID 等。

7.无消息丢失如何配置

不要使用 producer.send(msg)，而要使用 producer.send(msg, callback)。记住，一定要使用带有回调通知的 send 方法。设置 acks = all。acks 是 Producer 的一个参数，代表了你对“已提交”消息的定义。如果设置成 all，则表明所有副本 Broker 都要接收到消息，该消息才算是“已提交”。这是最高等级的“已提交”定义。设置 retries 为一个较大的值。这里的 retries 同样是 Producer 的参数，对应前面提到的 Producer 自动重试。当出现网络的瞬时抖动时，消息发送可能会失败，此时配置了 retries > 0 的 Producer 能够自动重试消息发送，避免消息丢失。设置 unclean.leader.election.enable = false。这是 Broker 端的参数，它控制的是哪些 Broker 有资格竞选分区的 Leader。如果一个 Broker 落后原先的 Leader 太多，那么它一旦成为新的 Leader，必然会造成消息的丢失。故一般都要将该参数设置成 false，即不允许这种情况的发生。设置 replication.factor >= 3。这也是 Broker 端的参数。其实这里想表述的是，最好将消息多保存几份，毕竟目前防止消息丢失的主要机制就是冗余。设置 min.insync.replicas > 1。这依然是 Broker 端参数，控制的是消息至少要被写入到多少个副本才算是“已提交”。设置成大于 1 可以提升消息持久性。在实际环境中千万不要使用默认值 1。确保 replication.factor > min.insync.replicas。如果两者相等，那么只要有一个副本挂机，整个分区就无法正常工作了。我们不仅要改善消息的持久性，防止数据丢失，还要在不降低可用性的基础上完成。推荐设置成 replication.factor = min.insync.replicas + 1。确保消息消费完成再提交。Consumer 端有个参数 enable.auto.commit，最好把它设置成 false，并采用手动提交位移的方式。就像前面说的，这对于单 Consumer 多线程处理的场景而言是至关重要的。

原文：https://mp.weixin.qq.com/s/LJslx5LhhxeLm_Ga_Xj3Ew

（求关注）如果觉得本文对你有帮助，请三连支持一下谢谢！