概要

本文主要介绍一些kafka内部原理概念，包括controller，Coordinator，patition，storage，produce和consume。

Patitions

Patition是kafka实际的存储单元，topic只是Patition的一个逻辑集合。Patition分为leader和followers，它们是由controller进行的分配。Patition又分为isr（in-sync replica）和osr（out-sync replica）；isr是表示leader patition认定该broker能实时同步数据（由replica.lag.time.max.ms参数控制lag的范围），osr表示改broker已经滞后了，需要加油赶上才能成为isr。

最开始分配的patition leader叫做preferred leader，由于controller在分配的时候考虑到了整个集群的均衡性，因此使用preferred leader可以较好的分担集群的压力。Kafka支持当leader迁移后通过选举再重新回到原先broker的功能。

Produce

生产数据是由client主动连接上leader patition并推送数据。

在api设计时，为了考虑效率，提供了同步和异步接口，同样也提供了单条推送和批量推送的接口。参数acks是生产数据的一个重要指标，=0时表示只要发送出去就认为生产成功（异步），=1时表示只要有leader patition接收到该消息就认为是成功，=all时表示所有isr patition都回复leader接受到该消息时表示成功（其实是各follower主动拉取的，leader只是进行被动的统计而已）。Patition在写接受到的数据时，也只是通过写入文件在内存中的镜像后就返回，并不保证一定会同步到磁盘上。

在多patition的情况下，如果不做处理，发送的messages会分散到不同的patition上，因此无从保证数据消费的顺序。Kafka每条消息有key，并且消息被分配到哪个patition上是由该key做hash后计算得到的。因此虽然kafka做全局消息的一致性比较难（除非是单patition，单线程生产，单线程消费），但是可以做到针对每个key的局部一致性。

Consume

同生产消息一致，消费时也是客户端主动连接到leader patition所在的broker上，并请求获取数据。除了topic和patition外，client还需要提供offset和最小与最大消费的数据量。设定最大的消费量是因为过大时可能引起client的有效内存不足，过小时又会导致client频繁请求broker，当已有的数据不足最小申请的数据量，但超过设定响应的最大时限时，也会把数据返回给客户端。可见kafka对于性能的考量无处不在。

Broker在处理client的消费请求时，会通过其提供的offset定位到文件的特定位置（后文会介绍定位的方式），然后使用zero-copy的方式直接从文件拷贝到网络管道（network channel）。

Client能接收到的数据并非是leader patition接收到的所有数据，而是isr都成功ack的数据，这也被称为high water mark。这样做的目的是保证在极端情况下，不同consumer所消费到的数据是一致的。

Controller

Kafka broker都含有controller（控制器）的功能，但同一时刻有且只有一个broker的该功能生效（通过向zookeeper抢占式注册来实现）。

Controller的主要功能是对于元数据进行管理。包括1、topic管理（创建、新增、删除），partition管理（选主，指定broker，异常迁移）；2、分区重分配；3、preferred leader选举；4、broker成员管理；5、向其它broker提供给数据服务；

虽然元数据是由controller所在的broker进行管理的，但是其它broker会定期拉取它的数据。因此client可以从任何一台broker中获取到元数据信息。

当controller所在的broker与zookeeper失联时，会由一个新的broker去主导controller，这时会产生一个new epoch number。如果原先的controller继续同步它的配置信息，通过这个参数broker知道这份配置已经失效了，可以不予理睬。

Coordinator

Coordinator主要是为consumer group服务的。对于任意一个consumer group，有唯一的broker负责协调者的工作，所有的consumer都需要通过该broker获取分配的patition和offset。

• 确定 consumer group 位移信息写入consumers_offsets 的哪个分区。具体计算公式： groupId.hashCode() % PartitionCount
• 该分区 leader 所在的 broker 就是被选定的 coordinator

触发consumer group的一次再分配过程称之为rebalance，触发时机如下：

1. 组成员发生变更 (新 consumer 加入组、已有 consumer 主动离开组或已有 consumer 崩溃了)
2. 订阅主题数发生变更——这当然是可能的，如果你使用了正则表达式的方式进行订阅，那么新建匹配正则表达式的 topic 就会触发 rebalance
3. 订阅主题的分区数发生变更

当触发rebalance后，该consumer group会生成一个新的generation，这样老的generation client提交的消费和commit都会被忽略。Rebalance分为两步进行：

• 所有的consumer client都向Coordinator发送join的请求，Coordinator指定其中一个client成为leader，并提供本generation的所有client信息和上次generation分配的patiton等信息。
• 所有的consumer client都向Coordinator发送sync请求。被指定为leader的client会发布它的决定（哪个client消费哪些patition），其它的client同步leader的决定。

storage

Kafka的物理存储是基于patition完成的，topic只是一个逻辑捆绑的概念。

Patition分布

假设我们有6个brokers，一个topic有10个patition，每个patition有3个replica（1个leader和2个follower）。我们的目标有（由controller进行实际的分配）

• 将所有replica均匀分布在所有的brokers上。因为一共有30个replicas，因此每个broker上会存储5个replicas；
• 对同一个patition，leader和followers需要分布在不同的brokers上；
• 如果broker有机架信息，同一个patition的不同replica尽可能分布在不同的机架上；

对于一个brokers，假设它的patitions可以存储在不同的物理磁盘上（文件夹），新的patition被放到哪一个文件夹，取决于哪个文件夹中已有的patition数量较少。

Segment

虽然kafka是基于patition进行存储的，但是segment才是最小的存储单元。一个segment包含不超过1G的数据（或者一段设置时间的数据）。当达到数据存储的最大时间后（或者达到最大容量后），最老的segment会被删除。这么设计的好处是对于一个文件，我们只会添加数据，不会删除部分数据，因此性能是非常优异的。

Dat文件存储格式

上图是一条正常的压缩message和一条带压缩的batch message

Message value的格式和producer发送以及consumer接收的完全一致，因此可以使用zero-copy来提升传输性能

Kafka自带的工具kafka-run-class.sh可以打印dat文件的内容

Idx文件

Idx和dat文件是成对出现的，它支持通过offset和timestamp分别进行索引，详细可参考附录文献。

Compation

Kafka也支持保存全量数据的，方式就是通过compation。它将相同的key进行过滤，只存储最新的value。Kafka对compation的算法是主要考虑内存的使用量。假设对于1G的dat文件，每条message平均长度是1K，那么总共有1M的数据量。Kafka在compation的过程中，内存会有一个map，map key是原始message key的hash，占16-byte，map value并非存储原始message value，而是message offset，只占8-byte。这样当内存的最大使用量也就是24M。

参考文献

Controller -- Kafka实战宝典：Kafka的控制器controller详解 - WindyQin - 博客园

Coordinator -- kafka系列之Coordinator(14) - 掘金

1. kafka Coordinator · 大专栏

消息保留机制 -- Kafka 消息保留机制 - gao88 - 博客园

Storage -- A Practical Introduction to Kafka Storage Internals · FreBlogg

Book -- Introduction · The Internals of Apache Kafka

数据一致性 -- https://cloud.tencent.com/developer/article/1839597