杨四正的kafka源码剖析课——1、kafka基本概念原创

你好，我是杨思正，今天开始我们的卡夫卡源码剖析课。阿帕奇卡夫卡是一个分布式消息系统，它是由卡拉语言编写而成的。我们下面来看一下卡夫卡的基本概念。首先是message message是卡不卡中最基本的数据元素。它是由K和VALUE2部分构成的，KV这两部分都是字节数组。卡普卡会按照一定的策略将message按照K路由到指定的partition中，从而保证相同K的message全部写入到同一个partition中。Value是卡夫卡真正传递的业务数据。好，接下来我们再来看卡夫卡中topic partition和broker这三个核心概念。Topic是卡夫卡中的一个逻辑概念，卡夫卡集群中可以定义多个topic。在使用的时候，Producer和consumer会约定好一个topic的名称。Producer产生的message会写入到指定的topic中，Consumer会从这个指定的topic中读取message。

Topic可以看作是一个message的通道，连接多个producer和consumer。也就是下面这张图展示的这样一个topic，同时连接了多个producer和多个。其中，Producer通过push message的方式将消息写入topic。Consumer则通过拉取消息的方式将消息拉取到本地，然后进行处理。在topic这个逻辑概念中，卡普卡做了进一步的拆分。将一个topic分为了一个或者多个partition。也就是说，一个topic至少有一个position。下面这张图就展示了拥有三个partition的一个topic。在producer写入的时候，会按照他的K将message写入到不同的partition中。从上面这张图我们可以看到，当message在被producer push到一个partition的时候。

都会被分配一个offset的编号，这个offset就是这条message在这个partition中的唯一编号。通过offset编号，卡夫卡就可以保证一个partition内的message是有序的。我们可以认为partition是一个用来记录有序message的存储。在生产环境中，不同topic需要支持的流量大小是有所不同的，也就是说topic需要横向扩展的能力，卡卡就是通过来实现这种横向扩展能力的。同一topic的不同partition是可以分布在不同物理机器上的。partition是topic横向扩展的最小单位。也就是说，一个part中存储的这一组有序的message没办法存储到多台机器上。常见的卡夫卡架构如下图所示，我们可以看到这里有两个topic，一个是TOPIC1，还有一个TOPIC2。TOPIC1和TOPIC2的PARTITION1都分布到了BROOK1这个物理机上。然后TOPIC1和TOPIC2他们的PARTITION2分别分配到了BROOK2这台机器上。

从这张图中我们可以看到，多个partition位于同一个broker上，Broker其实就是卡普卡的server服务，卡普卡集群是由多个broker构成的。我们可以通过增加partition以及broker的方式提高topic的吞吐量。这里的broker是卡夫卡的核心概念之一。broker主要负责下面三个核心的功能，第一个是接收producer发来的message信息，并保存到磁盘中，第二个核心功能是处理consumer的请求，也就是拉取消息的请求，最后一个核心功能就是处理集群中其他broke的请求。根据请求类型进行相应的处理并返回响应。了解了partition topic broker3个宏观的概念之后，我们向微观进一步分析。Partition在逻辑上对应一个log。

当producer向partition推送message的时候，卡普卡实际上是将写入到了producer对应的log中。Log由多个segment构成，Log和segment都是逻辑上的一个概念，实际上log对应磁盘上的一个文件夹，Segment对应的是文件夹下的一个segment文件和index文件。在写入partition时，我们只会写入最新的segment文件和index索引文件。当这个segment文件膨胀到一定大小之后，卡夫卡会创建新的segment文件继续写入，旧的segment文件就不再写入了。之所以这么设计，就是为了避免出现单个超大的sig文件。这也是为了采取顺序IO的方式写入，所以我们只向最新的segment追加数据。再来看index文件，它是segment文件的一个稀疏索引，在卡夫卡运行的过程中，会将index文件映射到内存中，从而提高索引的速度。

接下来我们再来看replica这个概念，也就是我们常说的副本。为了提高分布式系统的可用性，保证数据的完整性和安全性，我们一般会对数据进行备份，卡不卡也是这个套路。在卡夫卡中，Partition一般会有多个repl。每一个partition至少要有一个副本，同一个partition的不同副本中的message是一模一样的。虽然partition中的replica数据是一样的，但是他们还是有角色上的区分。一个partition中有多个副本，有一个副本是leader副本，其他的都是follow副本。所有的读写请求都是由leader副本进行处理的，其他的further副本仅仅是定期的从leader中拉取最新的message写入到本地。并同步更新到自己的logo中。下面的结构图展示了主副本和follow副本之间的交互producer push的时候会直接写入到副本中。

接下来。由fo副本发起破message请求，将最新写入的message同步到本地的logo中。同步完成之后，Consumer就可以拉取消息到本地进行消费了。正如上图所示，同一个partition中的不同副本会被分配到不同的broke上。这样，当leader副本所在的broke档期之后，卡夫卡会从剩余的follow副本中选举出新的leader副本，然后由这个新的leader副本继续对外提供读写服务。下面我们继续深入了解一下副本这个概念，在卡夫卡的副本中，有一个叫做isr概念。全称为insinc replplca，它表示的是一个副本集合，在这个集合中的副本必须满足下面两个条件，第一个条件是副本所在的broker必须与zookeper保持连接。

第二个条件是副本中最后一条message offset。与主副本中最后一条message的offset之间的差值不能超过我们指定的一个阈值。卡夫卡集群会为每个partition维护一个isr集合。卡夫卡写入message的功能与isr集合有着非常紧密的关系。在leader副本接收到写请求的时候，首先是由leader副本进行处理，并持久化到leader副本本地的log文件中，之后for副本定期从leader副本中拉取最新的message。并同步到follow副本自己本地的logo中。当然，Follow副本的定期同步请求相对于leader副本的写入操作来说会有一定的延迟，这就会造成follow副本中存储的message会略微落后于leader副本，但是只要未超出指定的阈值，都是可以容忍的。

这些follow副本，也就是处于isr集合中的这些副本。如果一个follow副本出现宕机、长时间的GC或者是网络故障等异常情况，导致其长时间没有与later副本进行通信，这也就导致了follow副本不再满足上述的条件，就会被踢出isr集合。在followler副本从上述故障恢复之后，会重新请求leader副本进行同步，当follow副本追上leader副本的时候，这个follow副本也就会重新加入isr集合。接下来我们再来看high water mark和log end of。这两个标记与上面的isr集合有着非常紧密的关联。HW记录的是一个特殊的offset值，在consumer拉取的时候，只能拉取到HW标记之前的message。

也就是说，HW之后的对于consumer来说是不可见的。与二集合类似，HW也是由leader副本管理的，当isr集合中全部的follow副本都同步了HW对应的message之后。副本会将HW标记进行更新。正是因为HW之前的message同时存在于多个副本中，即使leader副本出现宕机或者磁盘损坏等这些故障。这些message也不会出现数据丢失的问题。因为副本已经存储了HW之前的这些消息。所以卡夫卡认为，HW之前的message都是处于已提交状态。接下来我们再来看Leo这个标记，它是所有副本都拥有的一个off标记。它用来记录追加到当前副本的最后一个message的offset。

当leader副本接收到producer发送的message时，Leader副本的Lu标记就会增加。当fo副本成功从leader副本中拉取message并更新到本地log之后，Fo副本的Leo也会增加。了解了HW和Leo的基本概念之后，我们这里用一个动态的方式来说明一下H和Leo这两个标记是如何一起协同工作的。我们来看下面这张图。首先。Producer。向我们关心的一个topic其中的一个partition发送了一条message，该message写入到leader副本时，被分配的offset值为五。同时，我们leader副本会将它自身的Leo标记从四修改到五。此时。Leader副本的HW标记依然是四。接下来follow副本，从leader副本拉取message，同步到本地。

我们可以看到这里有两个follow副本，他们都会拉取offset为五这个message同步到本地的log。同步的过程中，Follow副本会将自己的Lu标记从四更新到五。当IS2集合中所有的副本都完成了对offset等于五这条message的同步之后。Leader副本会将HW标记从四修改招五。完成上述操作之后，Consumer再次来拉取message的时候，就可以看到offset为五的这条message了。介绍完卡夫卡副本的基础知识之后，我们再来深入介绍一下卡夫卡副本设计的一些思想。就如前文介绍的那样，冗余备份是分布式存储中最常见的方案之一。备份的常用方案有同步复制和异步复制两种。同步复制的含义是，所有的follow副本都要复制完一条message之后，该条message才处于提交状态。异步复制的含义是leader副本收到producer发送的message之后，会立即更新HW这个标记。

也就是认为这个message已经处于了已提交的状态，之后会用异步的方式从leader副本这里同步这条message。下面来看这两种复制方案的一些问题。对于同步方案来说，一旦有一个follow副本出现故障，或者长期C无法同步message。就会导致HW这个标记无法更新，消息始终无法提交，加游的也就获取不到这条消息了。此时，这个故障的follow副本就会导致整个分布式系统不可用。再来看异步复制方案。虽然异步方案避免了同步复制方案一个故障点拖垮整个集群的问题，但是异步方案存在一些数据丢失的风险，例如在集群中所有的follow副本。同步的速度都远远慢于leader副本，并且follow副本中存储的message量都远远落后于leader副本。

如果此时leader副本发生了宕机。则会重新选择新的leader副本，而重新选出的leader副本并没有包含原有leader副本的全部信息，这就会造成数据丢失。另外。一些consumer可能已经消费了原leader副本中的一些消息，但是这些消息在新中没有，也没有人知道这些丢失的message是什么，这样的话，整个系统的状态就变得不可控了。卡夫卡的副本设计和isr集合设计权衡了同步复制和异步复制这两个方案。当follow副本落后于雷副本的时候，卡夫卡将其判定为一个可能出现故障的副本，就会将他踢出IS2集合。由于message的提交只关注isr集合中的副本，所以这个慢速的follow副本并不会影响整个系统的性能。当leader副本出现宕机等异常的时候，卡夫卡会优先从IS2集合中选举出新的leader副本。

而新选举出的这个leader副本中，包含了HW标记之前的全部message。因为HW标记之后的messenger处于未提交状态。从卡夫卡集群的外部来看，是感知不到leader的切换的，同时也没有数据的丢失。接下来我们来看卡夫卡中的保留策略和日志压缩。对卡夫卡有一定了解的同学可能知道，无论message是否已经被consumer消费了，卡夫卡都会长时间保留message的信息，这种设计是为了让consumer可以轻松的回退到某个offset并进行消费。但是卡夫卡并不是数据库。不应该一直保存历史的message，尤其是那些已经确定不会再使用的历史message。我们可以通过修改卡夫卡的retention policy。

来实现周期性的清理历史message的效果。卡夫卡默认提供两种保留策略，第一种是根据message的保留时间进行清理，具体的含义是，当一条message在卡夫卡集群中保留的时间超过了指定的阈值，就可以被后台的线程清理掉了。第二种策略是根据topic占用的磁盘大小进行清理。具体的含义是，当topic的log大于一个指定的阈值之后，就可以开始由后台线程清除最旧的message了。卡夫卡的policy可以针对全部的topic进行配置，也可以针对某个特殊的topic进行特殊的配置。除了保留策略之外，卡夫卡还提供了日志压缩来减少磁盘的占用量。我们知道message是由KV2部分构成的，如果一个K对应的value会不断更新，且consumer只关心最新的VALUE6值，那我们就可以开启日志压缩的功能。其核心的原理是卡夫卡会启动一个后台线程，定期将K相同的message进行合并，合并后的结果只保留最新的Y6值。

下图展示了一次日志压缩的工作过程，这里我们关注K为K1的message，我们可以看到图中offset为一百一百零二一百零五以及107的四个message。他们的K都是K1。经过压缩之后，我们只保留offset为107的这个message。它的VALUE6值，也就是最新的VALUE6值，就是Y68。正如前面介绍的那样，卡夫卡集群是由多个broker构成的，卡夫卡集群会选出其中一个broker来担任controller。Controller主要负责管理partition的状态，管理partition下的副本状态以及监听组keepper数据的变化。controller是一主多从的实现方式，所有broker都会监听controller leader的状态，当leader出现故障的时候，会重新选举出新的R成为controller。

Controller的具体实现，我们将在后面的剖析过程中进行详细的介绍。了解了卡夫卡server的基本概念之后，我们再来看卡夫卡consumer的一些内容。consumer的主要工作是从topic拉取message并消费这些message来完成自身的业务逻辑。consumer中维护了一个offset信息，用来记录当前consumer消费到了partition的哪个位置。这个具体的位置就是这个offset的值。由consumer来维护offset是为了减少kafka broker维护consumer状态的压力，尤其是在broker出现故障或者延迟的时候。如果是由broker来管理consumer的消费状态。这种情况下就会导致消费状态的丢失，或者是影响consumer消费的速度。另外，这种设计也是让consumer可以按照自己的需求指定offset。例如我们可以通过指定offset跳过一部分message，或者是重复消费一些message。接下来我们看consumer group，卡卡中的多个consumer可以组成一个consumer group，其实consumer group才是消费卡夫卡message的基本单位，一个consumer只能属于一个consumer group。

在一个consumer group内部，会保证其消费的topic的每个partition只会分配给该consumer group中的一个consumer进行消费。当然，不同consumer group之间是不会相互影响的，一个topic可以有多个consumer group进行消费。根据consumer group的这个特性，我们可以将每个consumer独立成为一个consumer group。这样我们就可以实现消息的广播效果，也就是说，一个message被多个consumer同时消费。如果要实现独占消费的效果，我们可以将目标topic的全部consumer放入到一个consumer group中，这样的话就可以保证每个只有一个consumer进行消费。

自然，这个partition中的每一个message也只能被一个consumer进行消费。下图展示了一个consumer group中consumer与partition之间的对应关系，我们可以看到其中的CONSUMER0和CONSUMER1分别负责消费PART0和PART1这两个partition。CONSUMER2负责消费PART2和PART3这两个part consumer group除了提供独占和广播两种消费模式之外，Consumer group还提供了水平扩展和故障转移的能力。在上图CONSUMER2处理能力不足以消费两个part的时候。我们可以通过向consumer group中添加新的consumer的方式重新分配partition与consumer的映射关系，如下图所示，在添加CONSUMER3之后，CONSUMER2只消费二中的MESSAGE3则会重新分配给CONSUMER3来进行消费。接下来我们再来看consumer group故障转移的场景。

在CONSUMER3发生故障的时候，Consumer group也会进行重新分配。这个时候，就像下图展示的这样，CONSUMER2会重新接手三处理其中的。根据上述consumer与的对应原则。Consumer group中的consumer数量并不是越多越好。当一个consumer group中的consumer数量超过topic中的partition的数量，就会导致有的consumer分配不到，从而造成这些consumer空闲。浪费一些机器资源。在这一课时，我们重点介绍了卡夫卡的一些基本概念。我们介绍了message的构成，卡夫卡中topic partition broker这些核心的概念和他们之间的关联。然后深入到了partition的细节，介绍了log这个逻辑概念以及对应的物理实现。其中还介绍了segment文件、index稀疏索引文件等等。

接下来介绍了副本与partition之间的对应关系以及副本的主从设计，然后由副本引出了IR的概念，以及HW和Leo这两个标记。最后我们还讨论了isr这种复制方案的好处，除了这些之外，我们还介绍了卡普卡broker中。保留策略和日志压缩的相关内容。最后我们还介绍了卡不卡consumer以及consumer group的一些内容，感谢同学们观看本课程的相关文章和视频，还会放到微信公众号、B站和抖音。感谢大家的关注，我们下一节课再见。