进击消息中间件系列（四）：Kafka 服务器 Broker

简介

这篇文章介绍Kafka的Broker工作流程，包括其中控制器的选举过程；kafka副本的leader选举以及leader和follower故障流程；简单讲述了生产环境中如何调整分区副本；kafka的文件存储机制以及日志文件的删除策略；最后了解下kafka中使用的页缓冲和零拷贝的原理。更多关于消息中间件 Kafka 系列的学习文章，请参阅：消息中间件 Kafka，本系列持续更新中。

kafka Broker工作流程

• 第一步：broker启动后在zk中注册，争抢controller
• 第二步：谁先注册Controller，谁说了算
• 第三步：由选举出来的Controller，监听brokers节点变化
• 第四步：Controller决定Leader选举
• 第五步：Controller将节点信息上传到ZK
• 第六步：其他contorller从zk同步相关信息，万一挂了他们好上位
• 第七步：假设Broker1中Leader挂了
• 第八步：Controller监听到节点变化
• 第九步：获取ISR
• 第十步：选举新的Leader（在 isr中存活为前提，按照 AR中排在前面的优先）
• 第十一步：更新Leader及ISR

zookeeper作用

zookeeper 中存储的kafka信息

zookeeper在kafka中扮演了重要的角色，kafka使用zookeeper进行元数据管理，保存broker注册信息，包括主题（Topic）、分区（Partition）信息等，选择分区leader，在低版本kafka消费者的offset信息也会保存在zookeeper中。如图使用zookeeper客户端PrettyZoo查看内容如下：

Broker节点

kafka在zookeeper上存储的brokers节点信息如下：

其中/brokers/ids/[0...n]：是使用临时节点存储在线的是各个服务节点的信息，当下线后自动删除；

{
  "listener_security_protocol_map" : {
    "PLAINTEXT" : "PLAINTEXT"
  },
  "endpoints" : [ "PLAINTEXT://192.168.31.32:9091" ],
  "jmx_port" : -1,
  "features" : { },
  "host" : "192.168.31.32",
  "timestamp" : "1646557610504",
  "port" : 9091,
  "version" : 5
}

/brokers/seqid：辅助生成的brokerId，当用户没有配置broker.id时，ZK会自动生成一个全局唯一的id。

/brokers/topics/{topicName}：持久化数据节点存储topic的分区副本分配信息。在/brokers/topics/{topicName}//partitions/0/state中记录了leader和isr队列的内容。

{
  "controller_epoch" : 5,
  "leader" : 3,
  "version" : 1,
  "leader_epoch" : 4,
  "isr" : [ 2, 3 ]
}

其中_transaction_offsets，是事务存储的节点。

Consumers节点

0.9版本之前用于保存offset信息，0.9版本之后offset存储在kafka主题中。

Controller和/Controller_epoch节点

/controller：保存控制器（broker的leader， 这里的leader要和副本的leader区分开，这里的leader是kafka集群中所有broker的leader）对应的brokerId信息等

/controller_epoch：这里用来解决脑裂问题，存放的是一个整形值（纪元编号，也称为隔离令牌）

其他的节点

/config/topics：存储动态修改主题级别的配置信息。

/config/clients：存储动态修改客户端级别的配置信息。

/config/changes：动态修改配置时存储相应的信息。

/admin/delete_topics：在对主题进行删除操作时保存待删除主题的信息。

/isr_change_notification：保存Kafka副本ISR列表发生变化时通知的相应路径。

kafka broker的leader选举

leader选举的种类

这里需要先明确一个概念leader选举，因为kafka中涉及多处选举机制，容易搞混，kafka由三个方面会涉及到选举：

• broker（控制器）选leader
• 分区多副本选leader
• 消费者选leader

本文会讲述Broker选leader和分区选leader的过程，后面将消费者的时候在说消费者选leader的过程。

Broker选举过程

在kafka集群中由很多的broker（也叫做控制器），但是他们之间需要选举出一个leader，其他的都是follower。broker的leader有很重要的作用，诸如：创建、删除主题、增加分区并分配leader分区；集群broker管理，包括新增、关闭和故障处理；分区重分配（auto.leader.rebalance.enable=true，后面会介绍）；分区leader选举。

下面看一下Broker的leader选举过程和故障处理

如图可以大概描述如下：

• 1.集群中第一个启动的broker会在zookeeper中创建临时节点/controller来让自己成为控制器，当其他的broker启动也会在zookeeper中创建临时节点/controller，但是会发现节点已经存在，此时会受到一个异常，此时就会在zookeeper中创建一个watch对象，方便这些broker接受leader的变更消息；
• 2.如果主leader因为网络问题与zookeeper断开连接或者发生异常退出了，其他的broker就可以通watch接收到控制器的变更通知，开始尝试去创建临时节点/controller，如果有一个broker创建成功了，就和上面说的一样，其他的broker也会收到异常通知，此时就说明集群中broker的leader已经确定，其他的broker只能创建watch对象了
• 3.集群中broker的leader发生异常退出，在选举出新的leader之后，会检测这个异常的broker上面是否有分区副本的leader，如果有就是发起分区的leader选举，选出新额分区leader，然后更新ISR队列数据
• 脑裂问题

什么是脑裂

说起脑裂，那么什么是脑裂，出字面意思上理解，就是一分为二有了多个脑袋。在分布式系统的高可用情况下很容易出现这种问题，简单来说就是因为网络或者其他的原因导致leader出现假死状态，此时会触发leader选举，这样就会出现两个leader进而产生一系列问题。

kafka broker的leader相当于整个kafka集群的master，负责很多重要的工作（上文有，此处不再累述），broker是通过抢占的方式在zookeeper中注册临时节点/controller来时实现的，先到先得。由于zookeeper的临时节点的有效性是通过session来判断的，若在session timeout时间内，controller所在的broker断开，就会触发重新选举。

那么发生脑裂又会什么影响？

从上文可以知道，broker的leader主要是用于管理主题的，那些发生脑裂之后创建主题、增加分区的操作都会报错；但是现有的主题的读写是不影响的，这是因为读写是获取分区的元数据在任意一个broker中有可以拿到。

发生脑裂的情况？

broker的leader进行GC的时间超过zookeeper session timeout；broker的leader发生网络故障。

kafka的解决方案

为了解决Controller脑裂问题，zookeeper中有一个持久节点/controller_epoch，存放的是一个整形值的epoch number（纪元编号，也称为隔离令牌），集群中每选举一次控制器，就会通过Zookeeper创建一个数值更大的epoch number，如果有broker收到比这个epoch数值小的数据，就会忽略消息。

羊群效应

在早期的kafka版本中，如果宕机的那个Broker上的Partition比较多， 会造成多个Watch被触发，造成集群内大量的调整，导致大量网络阻塞，这种羊群效应会导致zookeeper过载的隐患。之后kafka有一个controller的概念（也就是broker的leader）来对分区副本的状态进行管理，当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本。当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。

触发leader选举

现阶段的kakfa集群中，只需要broker的leader在zookeeper去注册相应的监听器，其他的broker很少去监听zookeeper的数据变化，但是每个broker还是需要对/controller进行监听；当/controller节点发生数据变化的时候，每个broker都会更新自身内存中保存的activeControllerId。

当/controller节点被删除时，集群中的broker会进行选举，如果broker在节点被删除前是控制器的话，在选举前还需要有一个下线的操作（关闭相应的资源，比如关闭状态机、注销相应的监听器等）。如果有特殊需要，可以手动删除/controller节点来触发新一轮的选举。当然关闭控制器所对应的broker以及手动向/controller节点写入新的brokerid的所对应的数据同样可以触发新一轮的选举。

leader和follower

kafka的副本

kafka副本的作用就是提高数据的可靠性，系统默认副本数量是1，生产环境一般配置数量是2个，保证数据可靠性；否则副本太多会增加磁盘的存储空间，增加网络上的数据传输，降低效率。

kafka的副本分为leader和follower，其中leader数据读写，follower只负责数据同步。关于副本有下面三个概念：

• ISR：表示和leader保持同步的follower集合
• OSR：表示follower与leader同步延时过多的副本
• AR：分区中所有副本统称为AR（Assigned Repllicas），AR = ISR + OSR，一个分区的AR集合在分配的时候就被指定，并且只要不发生重分配的情况，集合内部副本的顺序是保持不变的，而分区的ISR集合中副本的顺序可能会改变。

这里ISR在上一篇文章中也介绍了，如果follower长时间没有向leader发送通信请求或者同步数据，这个follower将会被提出ISR队列，这个时间阈值是由replica.lag.time.max.ms参数设置的，默认是30s。

如果leader发送故障，就会从ISR中选举出新的leader。

leader选举流程

分区leader的选举由kafka的broker leader（后面文章会以controller代替broker leader的描述）负责具体实施。

当创建分区（创建主题或增加分区都有创建分区的动作）或分区上线（比如分区中原先的leader副本下线，此时分区需要选举一个新的leader上线来对外提供服务）的时候都需要leader选举。选举的时候将会从AR集合中副本的顺序查找第一个存活的副本，并且要保证这个副本在ISR队列中。

另外当分区发生重分配的情况（下面会讲）也是需要执行leader选举，此时从重分配的AR列表中找到第一个存活的副本，且这个副本在目前的ISR队列中。

再有就是当某一个borker节点关闭的时候，位于这个节点上的leader副本都会下线，所以与此对应的分区需要执行leader的选举。此时将会从AR列表中找到第一个存活的副本，且这个副本在目前的ISR列表中，另外还要确保这个副本不处于正在被关闭的节点上。

Unclean leader选举

kafka还提供了一个参数配置：unclean.leader.election.enable，默认是true，参数规定是否允许非ISR的副本成为leader，如果设置为true，当ISR队列是空，ISR为空说明leader和follower都挂掉了，此时将选择那些不在ISR队列中的副本选择为新的leader，这写副本的消息可能远远落后于leader，所以可能会造成丢失数据的风险。生产环境中建议关闭这个参数，设置为false。

leader和follower故障流程

LEO和HW

在生产环境中可能会出现follower和leader出现故障，那么Kafka是如何处理这些故障的呢？下面简单介绍一下流程，在讲流程之前，先了解一下LEO和HW这两个概念。

• LEO（log end offset）：每个副本的最后一个offset，LEO就是最新的offset+1
• HW（high watermark）：所有副本中最小的LEO；

LEO和HW的概念产生其实是因为，数据先写入leader，然后follower拉取数据进行同步，但是同步速度不一致，会出现先后问题，那个这是后副本的offset是不一样的，此时kafka会使用所有副本中最小的offset+1，也是HW。

follower故障流程

此时假如Broker1上的follower发生故障会出现什么情况呢？首先Broker1上的follower会被踢出ISR队列中，但是leader和其他的follower都还是会继续接受数据，并不会收到影响，对应的LEO和HW都会往后移动；如果此时发生故障的Broker1上的follower恢复后，此时Broker1上的follower会读取本地磁盘记录的上次HW位置，并将log文件中高于HW的部分截取掉，从HW开始向Leader进行同步；直到Broker1上的follower的LEO大于等于该分区的HW，此时说明这个follower追上了leader，就会将其重新加入ISR队列中。

leader故障流程

上面了解了follower故障的情况，那么如果leader发生故障呢？接着上面的图片来看，首先如果Broker0上的leader发生故障之后，也是一样会先从ISR队列中被踢出，然后从ISR中选出一个新的Leader来；此时为了保证多个副本之间的数据一致性，其他的follower会先将各自的log文件中高于HW的部分截取掉，然后从新的leader同步数据（由此可知这只能保证副本之间数据一致性，并不能保证数据不丢失或者不重复）。

分区副本的调整

从上一篇文章知道在kafka集群中分区的副本分布是做到尽量的均匀的分配的到各个节点中，以此来保证每台机器的读写吞吐量是均匀的，但是出现某些broker宕机，会导致leader都集中在几台broker中，造成读写压力过大，并且就算恢复了宕机的broker，原来的leader也会变成follower并无法分担压力，造成集群负载不均衡。

Leader Partition自动平衡

为了解决上述问题kafka出现了自动平衡的机制。kafka提供了下面几个参数进行控制：

• auto.leader.rebalance.enable：自动leader parition平衡，默认是true
• leader.imbalance.per.broker.percentage：每个broker允许的不平衡的leader的比率，默认是10%，如果超过这个值，控制器将会触发leader的平衡
• leader.imbalance.check.interval.seconds：检查leader负载是否平衡的时间间隔，默认是300秒

但是在生产环境中是不开启这个自动平衡，因为触发leader partition的自动平衡会损耗性能，或者可以将触发自动平和的参数leader.imbalance.per.broker.percentage的值调大点。

手动调整副本分配

会导致服务器的性能不一样，服务器磁盘不足或者其他的原因需要将性能好，磁盘空间大的服务器节点多存放副本，那么在生产环境中如何去手动调整分区副本的分布比例呢？

下面先创建一个测试的主题：

下面演示一下如何更新分区间的副本配比，首先创建一个assign-replicas.json的文件，内容如下：

{
    "version": 1,
    "partitions": [
        {"topic": "test-assign", "partition": 0, "replicas": [1, 2]},
        {"topic": "test-assign", "partition": 1, "replicas": [1, 2]},
        {"topic": "test-assign", "partition": 2, "replicas": [1, 2]}
    ]
}

接着执行命令：

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file assign-replicas.json --execute

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file assign-replicas.json --verify

最后看一个这个主题的副本分布情况：

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test-assign

增加副本因子

生产环境中由于某个主题的重要等级需要提升，考虑增加副本。下面演示下如何增加副本。

创建一个Json文件：add-relication-factor.json

{
    "version": 1,
    "partitions": [
        {"topic": "test-assign", "partition": 0, "replicas": [3, 2, 1]},
        {"topic": "test-assign", "partition": 1, "replicas": [1, 3, 2]},
        {"topic": "test-assign", "partition": 2, "replicas": [2, 1, 2]}
    ]
}

执行副本存储计划：

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file add-relicati
on-factor.json --execute

文件存储机制

存储结构

在kafka中主题（Topic）是一个逻辑上的概念，分区（partition）是物理上的存在的。每个partition对应一个log文件，该log文件中存储的就是Producer生产的数据。Producer生产的数据会被不断追加到该log文件末端。为防止log文件过大导致数据定位效率低下，kafka采用了分片和索引机制，将每个partition分为多个segment，每个segment包括.index文件、.log文件和.timeindex等文件。这些文件位于文件夹下，该文件命名规则为：topic名称+分区号。

上图对应的是某一个主题的文件结构图，一个主题是对应多个分区，一个分区对应一个日志（Log），如果只通过一个log文件记录的话，这就会导致日志过大，导致数据定位效率低下，所以kafka采用了分片和索引机制。

kafka引入了日志分段（LogSegment），将日志分为多个较小的文件；Log会存储在配置的log.dirs文件夹内，而每个LogSegment由三个文件组成：偏移量索引文件（.index后缀）、时间戳的索引文件（.timeindex后缀）和消息数据文件（.log后缀）；注意这里面还有一个leader-epoch-checkpoint文件，保存的是Leader Epoch的值（解决副本数据一致性需要）。

目录结构如下

├── long-topic-0
    ├── 00000000000000000000.index
    ├── 00000000000000000000.log
    ├── 00000000000000000000.timeindex
    ├── 00000000000000000113.index
    ├── 00000000000000000113.log
    ├── 00000000000000000113.timeindex
    └── leader-epoch-checkpoint

分段文件名规则：分区的第一个segment是从0开始的，后续每个segment文件名为上一个segment文件最后一条消息的offset，ofsset的数值最大为64位（long类型），20位数字字符长度，没有数字用0填充。

log文件默认写满1G后，会进行log rolling形成一个新的分段（segment）来记录消息，这里面的分段大小取决于：log.segment.bytes参数决定。

index和timeindex文件在刚使用时会分配10M的大小，当进行log rolling后，它会修剪为实际的大小，所以看到前几个索引文件的大小，只有几百K。

查看log文件内容的方法：

./bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files 00000000000000000000.log  --print-data-log

log、timeindex和index文件的对应关系

当log文件写入4k（这里可以通过log.index.interval.bytes设置）数据，就会写入一条索引信息到index文件中，这样的index索引文件就是一个稀疏索引，它并不会每条日志都建立索引信息。

log日志文件是顺序写入，大体上由message+实际offset+position组成，而索引文件的数据结构则是由相对offset（4byte）+position（4byte）组成。

当kafka查询一条offset对应实际消息时，可以通过index进行二分查找，获取最近的低位offset，然后从低位offset对应的position开始，从实际的log文件中开始往后查找对应的消息。

时间戳索引文件，它的作用是可以查询某一个时间段内的消息，它的数据结构是：时间戳（8byte）+ 相对offset（4byte），如果要使用这个索引文件，先要通过时间范围找到对应的offset，然后再去找对应的index文件找到position信息，最后在遍历log文件，这个过程也是需要用到index索引文件的。

kafka高效读写原因

kafka之所以可以快速读写的原因如下：

• kafka是分布式集群，采用分区方式，并行操作
• 读取数据采用稀疏索引，可以快速定位消费数据
• 顺序写磁盘
• 页缓冲和零拷贝

页缓冲和零拷贝

页缓冲

在 Kafka 中，大量使用了 PageCache， 这也是 Kafka 能实现高吞吐的重要因素之一。

首先看一下读操作，当一个进程要去读取磁盘上的文件内容时，操作系统会先查看要读取的数据页是否缓冲在PageCache 中，如果存在则直接返回要读取的数据，这就减少了对于磁盘 I/O的 操作；但是如果没有查到，操作系统会向磁盘发起读取请求并将读取的数据页存入 PageCache 中，之后再将数据返回给进程，就和使用redis缓冲是一个道理。

接着写操作和读操作是一样的，如果一个进程需要将数据写入磁盘，操作系统会检查数据页是否在PageCache 中已经存在，如果不存在就在 PageCache中添加相应的数据页，接着将数据写入对应的数据页。另外被修改过后的数据页也就变成了脏页，操作系统会在适当时间将脏页中的数据写入磁盘，以保持数据的一致性。

零拷贝

零拷贝并不是不需要拷贝，而是减少不必要的拷贝次数，通常使用在IO读写过程中。常规应用程序IO过程如下图，会经过四次拷贝：

• 数据从磁盘经过DMA到内核的Read Buffer；
• 内核态的Read Buffer到用户态应用层的Buffer
• 用户态的Buffer到内核态的Socket Buffer
• Socket Buffer到网卡的NIC Buffer

从上面的流程可以知道内核态和用户态之间的拷贝相当于执行两次无用的操作，之间切换也会花费很多资源；当数据从磁盘经过DMA 拷贝到内核缓存（页缓存）后，为了减少CPU拷贝的性能损耗，操作系统会将该内核缓存与用户层进行共享，减少一次CPU copy过程，同时用户层的读写也会直接访问该共享存储，本身由用户层到Socket缓存的数据拷贝过程也变成了从 内核到内核的CPU拷贝过程，更加的快速，这就是零拷贝，IO流程如下图。

Java的JDK NIO中方法transferTo()方法就能够实现零拷贝操作，这个实现依赖于操作系统底层的sendFile()实现的：

public void transferTo(long position, long count, WritableByteChannel target);

底层调用的是：

#include <sys/socket.h>
ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count);

但是需要注意零拷贝和系统底层有很大的关系，所以是否可以进行零拷贝的系统调用的看具体的操作系统是否实现。

下面看一下Java nio的零拷贝例子：

import java.io.*;
import java.nio.channels.FileChannel;

public class ZeroCopy {

    public static void main(String[] args) {
        File source = new File("G:/source.zip");
        File target = new File("G:/target.zip");
        NioZeroCopy(source, target);
    }

    public static void NioZeroCopy(File source, File target) {
        try (
                FileChannel sourceChannel = new FileInputStream(source).getChannel();
                FileChannel targetChannel = new FileOutputStream(target).getChannel();
        ) {
            for(long count = sourceChannel.size(); count > 0;) {
                long transfer = sourceChannel.transferTo(sourceChannel.position(), count, targetChannel);
                sourceChannel.position(sourceChannel.position() + transfer);
                count -= transfer;
            }
        } catch (IOException e) {
            System.out.println("异常：" + e.getMessage());
        }
    }

}

Broker核心参数配置

replica.lag.time.max.ms 
#ISR中，如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。时间阈值默认为30s。

auto.leader.rebalance.enable 
#默认true。自动Leader Partition平衡。建议关闭。

leader.imbalance.per.broker.percentage 
#默认为10%。每个broker允许的不平衡的leader的比率。如果每个broker超过了这个值，控制器会触发leader的平衡。

leader.imbalance.check.interval.seconds 
#默认值 300 秒 。检查 leader 负载是否平衡的间隔时间。

log.segment.bytes 
#kafka中log日志是分成一块块存储的。log日志划分成块的大小，默认值1G。

log.index.interval.bytes 
#默认 4kb ，kafka 里面每当写入了 4kb 大小的日志（.log），然后就往 index 文件里面记录一个索引。

log.retention.hours 
#kafka中数据保存的时间，默认7天。

log.retention.mintues 
#kafka中数据保存的时间，分钟级别，默认关闭。

log.retention.ms 
#kafka中数据保存的时间，毫秒级别，默认关闭。

log.retention.check.interval.ms 
#检查数据是否保存超时的间隔，默认是5min。

log.retention.bytes 
#默认等于-1，表示无穷大。超过设置的所有日志总大小，删除最早的segment。

log.cleanup.policy 
#默认是delete，表示所有数据启用删除策略。如果设置值为compact，表示所有数据启用压缩策略。

num.io.threads 
#默认是8。负责写磁盘的线程数。整个参数值要占总核数的50%

num.replica.fetchers 
#默认是1。副本拉去线程数。该参数占总核数的50%的1/3

num.network.threads 
#默认是3。数据传输线程数。该参数占总核数的50%的1/3

log.flush.interval.messages 
#强制页缓存刷写到磁盘的条数，默认是long的最大值，9223372036854775807。一般不建议修改，交给系统自己管理。

log.flush.interval.ms 
#每隔多久刷数据到磁盘，默认是null。一般不建议修改。

参考文章：https://blog.csdn.net/yhflyl/article/ details/123582735