Kafka-Broker_kafka broker

1. 集群成员

Kafka使用Zookeeper来管理集群成员（Broker），Broker启动时通过创建临时节点将自己注册到在Zookeeper，Zookeeper会将Broker的变动信息通知给控制器。

1.1 控制器

控制器是一个Kafka集群中的一个Broker，其作用是在Zookeeper的帮助下管理和协调整个Kafka集群。

Zookeeper概述
Zookeeper是一个高可靠性的分布式协调框架，核心功能是提供了：文件系统+通知机制。
ZK在内存中以树状目录结构存储数据，保证了高吞吐量和低延迟，但也限制了数据量的大小。
ZK赋予了客户端监听节点znode变更的能力，客户端通过在节点上注册Watcher，当事件发生时ZK会显示通知客户端。
常用作集群成员管理、Leader选举、注册中心、分布式锁等等。

选举
Broker启动时会尝试创建/controller临时节点，第一个成功创建的Broker会被指定为控制器。其它节点在该节点上注册Watcher以接收节点变更通知，并在控制器离线时，竞选控制器。
控制器使用epoch来避免“脑裂”，防止两个Broker同时认为自己是当前唯一的控制器。Broker在成为控制器之后，会自增epoch，旧控制器的epoch值较小，其消息会被其他Broker忽略。

功能

1. 集群成员管理（Broker上下线）
   利用ZK的Watch机制和临时节点对集群Broker进行管理。比如：

   • 当有新Broker启动后，其会在/broker/ids下创建一个临时节点，ZK通过Watch机制将该事件通知给控制器，控制器进而开始新增Broker作业。
   • 当有Broker离线时，其与ZK之间的session结束，临时节点会被自动删除，ZK同样将该变更通知给控制器，控制器进而进行一些善后操作。比如首领副本离线，控制器需要为该分区重新选举分区首领（副本列表下一个副本），并将结果通知给该分区所有副本Broker。消费者需要重新发送Metadata请求获取新的分区首领信息。

2. 主题、分区管理（增删主题/分区、分区分配、首领选举）

3. 数据存储
   控制器上保存了最全的集群元数据信息，其它Broker会定期接收控制器发来的元数据更新请求，从而更新其缓存数据，所以客户端元数据请求可发送给任意Broker。比较重要的数据有：所有主题信息、所有Broker信息。这些信息在ZK上也存储了一份，控制器初始化时会从ZK拉取。

故障转移Failover
任意时刻，Kafka集群中只有一个控制器，为了防止出现单点故障，Kafka为控制器提供了故障转移机制。
当控制器离线时，与ZK之间的session中断，ZK会删除/controller临时节点，集群中的其它Broker由于在/controller节点上注册了Watcher，会接收到ZK发来的节点变更通知，进而开始竞选控制器：抢先创建/controller节点。最终，第一个创建者成为控制器，从ZK拉取集群元数据，开始履行工作职责。

设计

• 单线程 + 事件队列
• 异步Zookeeper操作

1.2 群组协调器

群组协调器是为消费者群组服务的Broker，提供组成员管理及位移管理。其位置是保存该群组位移主题分区首领副本所在的Broker，由groupID.hashCode % 位移主题分区数 即可确定群组位移主题分区。

2. 副本

复制和分区多副本机制是Kafka Broker高可用、高持久性的实现方案。

副本通常是指分布式系统在多台机器上保存相同的数据拷贝，其作用有：

• 提供数据冗余，用于故障转移；
• 提高伸缩性，通过横向扩展提高读性能；
• 改善数据局部性，将数据放在里用户更近的地方，从而降低数据延迟。

Kafka将副本分为两类：领导者副本（Leader Replica）和跟随者副本（Follower Replica）。每个分区都仅有一个领导者副本，所在broker称为分区首领，负责处理所有生产和消费请求；跟随者副本只是异步地与领导者副本进行同步，不对外提供服务，并在领导者故障时参与首领选举。

Kafka的跟随者副本为什么不对外提供服务？
在一些数据存储系统中，例如MySQL、Redis做读写分离，从库是可以处理读操作的。Kafka跟随者副本不对外提供服务的原因在于：

1. 使用场景：Kafka常用作消息引擎而不是数据存储，其使用场景并不属于读多写少的。
2. Kafka的多分区机制，分区（副本、领导者副本）一般均匀地分布在不同的Broker上，已经起到了负载均衡的作用。
3. 更容易保证数据一致性，实现单调读。不会出现从多个副本读取消息而出现不一致的现象。

2.1 同步副本 In-Sync Replicas(ISR)

跟随者副本不对外提供服务，只是异步地从领导者副本拉取消息，既然是异步的，就存在与领导者不实时同步的风险。为了精准的定义同步的含义，Kafka提出了同步副本的概念，并利用ISR集合来管理同步副本。ISR是一个动态调整的集合。

跟随者成为同步副本需要满足的条件：

1. 跟随者副本落后于领导者副本之间的时间延迟小于replica.lag.time.max.ms，与落后的消息数量无关；
2. 跟随者与ZK维持着正常的心跳。

滞后的同步副本会拖慢生产者和消费者，因为：

1. 发送确认acks=all时，生产者会等待消息写入所有同步副本；
2. 消费者只能消费已提交消息，即消息被写入所有同步副本后。

非同步副本同样滞后，但不会对性能产生影响。但更少的同步副本意味着更低的有效复制系数，发生数据丢失的风险更高。

Broker下线时，分区会丢失一个副本，但是Kafka并不会为其创建新的副本。

3. 位移主题

"__customer_offset"是kafka的一个内部主题，用于管理消费者的消费位移。老版本消费者的消费位移是利用Zookeeper进行管理的，这减少了Broker端需要维持的状态，有利于实现高伸缩性。但ZK并不适用于高频写操作场景，而Kakfa的主题设计天然就满足高频写操作和高持久性。

消息格式
位移主题的消息格式是KV对，key为消息的键，value为消息体。

• key的内容有：GroupID，主题名，分区名
• value的内容有：偏移量、用户自定义数据等元数据

除了常见的位移消息，位移主题中还有墓碑消息tombstone，其消息体是null。当消费者群组下的所有消费者实例都停止后，Kafka会往位移主题的对应分区发送墓碑消息，用于删除Group。

分区数及复制系数
位移主题的默认分区数为50，副本数为3。
消费者群组所属位移主题的分区可由 groupID.hashCode % 位移主题分区数 确定。

整理、清理：Compact策略
对于位移数据这种状态信息，只需要保存最新的一条，之前的消息都是可以删除的。
Kafka利用Compact策略来整理这种保存状态信息的主题：只保存同一个键的最新消息，删除其过期消息。

4. 请求处理

Kafka Broler使用Reactor模式来处理请求。

4.1 Reactor模式

Reactor模式是事件驱动架构的一种实现方式，特别适用于客户端高并发请求服务端的应用场景。在该模型中，客户端请求发送给了Reactor，请求分发线程Acceptor将请求分发给工作线程。工作线程池负责实际的业务处理，可根据负载进行横向伸缩。Acceptor只负责请求分发，不实际具体业务处理，显得非常轻量级，能达到很高的吞吐量。

4.2 Reactor在Kafka中的实现

在Kafka中，Broker端的Acceptor线程以轮询的方式将请求分发给工作线程——网络线程。网络线程池并不对请求进行处理，而是将请求放入到共享的请求队列中。IO线程池从队列中取出请求，执行真正的处理：对于生产请求，将消息写入磁盘；对于获取请求，则从磁盘或是页缓存读取数据。IO线程处理完请求后，将响应放入对应的响应队列中，由相应的网络线程返回给客户端。Acceptor线程只负责请求分发，不负责响应回传。

在响应队列前还有一个炼狱组件（Purgatory），用来缓存延迟请求。所谓的延迟请求，就是指那些未满足条件不能立即处理的请求。比如，设置了发送确认acks = all的生产请求，那么该请求必须等所有同步副本收到消息后，才能返回。因此，处理该请求的IO线程不能立即处理该请求，而必须等待其它Broker的写入结果，那么它就会将请求暂存在炼狱组件中。稍后满足条件时再将请求取出，继续处理完成后放入响应队列中。

在2.3版本之前，没对控制类请求和数据类请求进行区分，而实际情况是控制类请求应该具有更高的优先级才对。在新版中，社区引入了两套相同的组件，分别处理这两类请求。

4.3 请求请求

数据类请求
生产和获取请求都必须发送给分区的首领副本，否则会收到“非分区首领”的错误响应。客户端通过元数据请求来获取这些集群信息。

首领副本收到生产请求时，会根据发送确认来决定将请求写入多少个同步副本后再将响应返回，如果同步副本数不足，Kafka可以配置拒绝处理新消息。消息被写入文件系统的缓存中，但并不保证何时会被刷盘。Kafka利用复制和分区多副本机制来保证持久性。PS：Broker端在写入消息前，会对消息批次进行解压，对消息执行各种验证，没有利用零拷贝技术。

Kafka利用零拷贝来处理获取请求，无需在内核空间和用户空间进行冗余的内存拷贝。如果混用新老消费者程序，Broker会对消息进行格式转换，这丧失了零拷贝特性。只有已提交消息对消费者可见，Kafka利用高水位和Leader Epoch来实现。

元数据请求
可以发送给任意Broker，以获取集群信息，包括主题分区、分区副本、首领副本等等。

5. 物理存储

文件管理
Kafka使用只能追加写（Append-Only）的消息日志来保存数据，避免了缓慢的随机IO。每个消息日志又被分成若干个日志片段，当消息只能被追加到最新的片段中，也就是活跃片段。当数据达到片段上限时，Kafka会关闭当前文件，重新打开一个新的文件。

Kafka不会一直保存消息，也不会等所有所有消费者读取消息后才删除消息。而是通过配置的保留期限或者保留大小来进行清理。

文件格式
保存在Broker上的数据格式与生产者发来的数据格式，以及发往消费者的数据格式是一致的。使用相同的数据格式来存储和网络传输，这是Kafka可以使用零拷贝和页缓存来加速获取请求的前提。Kafka不会操作具体的一条消息，而是在批次层面进行写入操作。但如果混用新老消费者程序，Broker会对消息进行格式转换，同时也丧失了零拷贝特性。

索引
消费者可以从任意可用偏移量处开始处理消息，为了快速定位消息，Kafka利用跳表和稀疏索引为每个分区设计了一个索引。用跳表可定位到消息所在的日志片段，在片段内部使用稀疏索引大致定位。在删除消息片段时，对应的索引也会被删除，跳表这种索引结构更容易进行区间删除，而树需要考虑平衡问题。PS：B+树是为磁盘优化的，在千万级数据量的MySQL中，读取数据时只需2-3次IO即可。

整理/清理 Compact
对于保存状态信息的主题，只需要保存最新的一条，之前的消息都是可以删除的。
Kafka利用Compact策略来整理这种保存状态信息的主题：只保存同一个键的最新消息，删除其过期消息。

分区分配
在创建主题时，Kafka会在Broker上均匀的分配分区及其副本。首先随机从一台Broker开始，顺序分配首领副本。随后，在从每个分区的首领副本所在Broker开始，顺序分配跟随者副本。

6. 高水位（HW）和Laeder Epoch

Kafka利用高水位（HW）来定义消息的可见性，即用来表示分区下哪些消息是可以被消费者消费的。同时也用来帮助Kafka完成副本同步。
在高水位以下的消息为已提交消息，在不考虑事务的情况下，消费者可以消费这部分消息。

6.1 高水位机制的实现

Kafka在每个副本对象中都保存两个属性：高水位和LEO值。LEO值表示日志末端位移（Log End Offset），即下一条消息的位移值。在同一个副本对象中，高水位值永远小于LEO值。

首领副本的高水位代表着分区的高水位，为了确定这个值，Kafka在首领副本上还额外保存了所有跟随者副本的LEO值。当跟随者副本从首领副本拉取消息时，首领副本会用其请求的偏移量来更新该副本的LEO值。由于在LEO值之前的消息已经被跟随者副本保存，那么首领副本的高水位 = 所有同步副本LEO值的最小值。利用高水位实现了消费者只能获取到已提交消息的限制。

6.2 Leader Epoch

由于首领副本和跟随者副本的高水位更新存在延迟，在首领副本和跟随者副本接连发生宕机时，单纯依靠高水位会造成数据丢失、数据不一致（相同位移不同消息），无法实现对已提交消息做有限度持久化的保证！

日志截断造成消息丢失
在宕机副本重启后，利用高水位进行不必要的日志截断会造成数据丢失。

比如，在一个有两个副本的分区上，生产者先后发送了两条消息，发送确认acks=1，当首领副本成功写入后，会给生产者返回成功。假设虽然跟随者副本也写入了两条消息，但由于高水位的更新延迟，导致首领副本的高水位已经更新了，但是跟随者副本还没有更新。此时，跟随者副本宕机了。当它重启后，会根据自身的高水位对消息日志进行日志截断（？？），这样最后一条消息就被删除了。随后，跟随者副本开始从首领副本同步消息。这时，首领副本宕机了…跟随者副本接替了首领副本的位置。当原首领副本重启后，由于跟随者副本的高水位不能高于首领副本，其同样执行了日志截断。这样，第二条消息就丢失了。

利用Leader Epoch进行日志截断
0.11版本，日志截断不再依赖高水位，宕机副本重启后，如果Leader Epoch没有变化，那么日志截断可能就是不必要的。

Leader Epoch可以理解为Leader的版本，保存在所有分区副本上，有两部分组成：版本号 +　起始位移。当首领副本变更时，会生成一个新的Leader Epoch条目。当宕机的副本重启后，会根据Leader Epoch来判断宕机期间是否发生过首领变更，如果没有，并且当前首领副本的LEO大于自身LEO，就不必进行日志截断。

7. Broker的配置参数

复制系数 replication.factor
分区的副本数量，默认值为3。更大的复制系数意味着更高的可用性，但会占据更多的空间。这需要在可用性和存储空间作出权衡。

最少同步副本 min.insync.replicas
最小同步副本保证了已提交消息至少被写入多少个副本，当同步副本数量不足时，Broker会停止接收新的写请求，此时该分区变成了只读模式。该参数需要在可用性A和数据一致性之间做出权衡。

Kafka只对已提交消息作有限度的持久化保证。消息被写入所有同步副本即被认为是已提交的，有限度是指至少一个保存该消息的Broker存活，Kafka才能保证消息不丢失。如果当前只有一个同步副本，那么当该副本不可用时，数据就可能会丢失，因为数据可能没有落盘。

配合生产者的发送确认acks = all，就可以确定在返回确认前，至少有多少个副本能够收到消息。这种确认参数可靠性最高。

不完全的首领选举 unclean.leader.election
分区首领选举时是否允许非同步副本成为首领。当首领副本宕机时，一个同步副本会被选举为新的首领，这样已提交的消息不会丢失。但若在选举时，副本都是不同步的，这个时候就需要在可用性A和数据一致性C之间做出选择。若允许不同步的副本成为首领，就有数据丢失和不一致的风险；若不允许，那么分区在原首领副本恢复前将处于不可用状态。

建议禁用不完全的首领选举，保证数据一致性，并通过增加副本数来提高可用性。

数据留存 retention.xxx
可按时间、容量对消息保存进行限制。

JVM配置
6G堆（或将堆设置成Full GC后占用内存的1.5 - 2 倍），G1较CMS有着更少的FUll GC。

文件系统参数
Kafka持久化消息并不是等消息落盘后才认为是成功的，数据被写入操作系统的页缓存就可以了，随后由flush线程定期对脏页进行刷盘。默认提交时间是5秒，可稍微拉大提交间隔以换取性能提升。

如果页缓存中的数据没有刷盘前，Broker宕机了，那么数据将会丢失。但是，Kafka是在软件架构层面，通过分区多副本机制来保证数据的高持久性。

参考

《Kafka权威指南》
《Kafka核心技术与实战》
《大数据中台之Kafka，到底好在哪里？》　微信公众号　架构师社区