集群成员间的关系
我们知道，Kafka 是运行在 ZooKeeper 之上的，因为 ZooKeeper 是以集群形式出现的，所以 Kafka
也可以以集群形式出现。这也就涉及到多个生产者和多个消费者如何协调的问题，这个维护集群间的关系也是由 ZooKeeper 来完成的. Kafka
集群间会有多个 主机(broker)，每个 broker 都会有一个 broker.id，每个 broker.id
都有一个唯一的标识符用来区分，这个标识符可以在配置文件里手动指定，也可以自动生成。
Kafka 可以通过 broker.id.generation.enable 和 reserved.broker.max.id 来配合生成新的
broker.id。
broker.id.generation.enable参数是用来配置是否开启自动生成 broker.id
的功能，默认情况下为true，即开启此功能。自动生成的broker.id有一个默认值，默认值为1000，也就是说默认情况下自动生成的 broker.id
从1001开始。
Kafka 在启动时会在 ZooKeeper 中 /brokers/ids 路径下注册一个与当前 broker 的 id 相同的临时节点。Kafka
的健康状态检查就依赖于此节点。当有 broker 加入集群或者退出集群时，这些组件就会获得通知。
- 如果你要启动另外一个具有相同 ID 的 broker，那么就会得到一个错误 —— 新的 broker 会试着进行注册，但不会成功，因为 ZooKeeper 里面已经有一个相同 ID 的 broker。
- 在 broker 停机、出现分区或者长时间垃圾回收停顿时，broker 会从 ZooKeeper 上断开连接，此时 broker 在启动时创建的临时节点会从 ZooKeeper 中移除。监听 broker 列表的 Kafka 组件会被告知该 broker 已移除。
- 在关闭 broker 时，它对应的节点也会消失，不过它的 ID 会继续存在其他数据结构中，例如主题的副本列表中，副本列表复制我们下面再说。在完全关闭一个 broker 之后，如果使用相同的 ID 启动另一个全新的 broker，它会立刻加入集群，并拥有一个与旧 broker 相同的分区和主题。
Broker Controller 的作用
我们之前在讲 Kafka Rebalance 重平衡的时候，提过一个群组协调器，负责协调群组间的关系，那么 broker
之间也有一个控制器组件（Controller），它是 Kafka 的核心组件。它的主要作用是在 ZooKeeper 的帮助下管理和协调整个 Kafka
集群，集群中的每个 broker 都可以称为 controller，但是在 Kafka 集群启动后，只有一个 broker 会成为 Controller
。既然 Kafka 集群是依赖于 ZooKeeper 集群的，所以有必要先介绍一下 ZooKeeper 是什么，
这里就简单提一下 znode 节点的问题。
ZooKeeper 的数据是保存在节点上的，每个节点也被称为znode，znode 节点是一种树形的文件结构，它很像 Linux
操作系统的文件路径，ZooKeeper 的根节点是 /。
image.png
znode 根据数据的持久化方式可分为临时节点和持久性节点。持久性节点不会因为 ZooKeeper 状态的变化而消失，但是临时节点会随着 ZooKeeper
的重启而自动消失。
node 节点有一个 Watcher 机制：当数据发生变化的时候， ZooKeeper 会产生一个 Watcher 事件，并且会发送到客户端。Watcher
监听机制是 Zookeeper 中非常重要的特性，我们基于 Zookeeper
上创建的节点，可以对这些节点绑定监听事件，比如可以监听节点数据变更、节点删除、子节点状态变更等事件，通过这个事件机制，可以基于 ZooKeeper
实现分布式锁、集群管理等功能。
控制器的选举
Kafka 当前选举控制器的规则是：Kafka 集群中第一个启动的 broker 通过在 ZooKeeper 里创建一个临时节点 /controller
让自己成为 controller 控制器。其他 broker 在启动时也会尝试创建这个节点，但是由于这个节点已存在，所以后面想要创建 /controller
节点时就会收到一个 节点已存在 的异常。然后其他 broker 会在这个控制器上注册一个 ZooKeeper 的 watch 对象，/controller
节点发生变化时，其他 broker 就会收到节点变更通知。这种方式可以确保只有一个控制器存在。那么只有单独的节点一定是有个问题的，那就是单点问题。
image.png
如果控制器关闭或者与 ZooKeeper 断开链接，ZooKeeper 上的临时节点就会消失。集群中的其他节点收到 watch
对象发送控制器下线的消息后，其他 broker 节点都会尝试让自己去成为新的控制器。其他节点的创建规则和第一个节点的创建原则一致，都是第一个在
ZooKeeper 里成功创建控制器节点的 broker 会成为新的控制器，那么其他节点就会收到节点已存在的异常，然后在新的控制器节点上再次创建 watch
对象进行监听。
image.png
控制器的作用
那么说了这么多，控制是什么呢？控制器的作用是什么呢？或者说控制器的这么一个组件被设计用来干什么？别着急，接下来我们就要说一说。
Kafka 被设计为一种模拟状态机的多线程控制器，它可以作用有下面这几点
- 控制器相当于部门（集群）中的部门经理（broker controller），用于管理部门中的部门成员（broker）
- 控制器是所有 broker 的一个监视器，用于监控 broker 的上线和下线
- 在 broker 宕机后，控制器能够选举新的分区 Leader
- 控制器能够和 broker 新选取的 Leader 发送消息
再细分一下可以具体分为如下 5 点
- 主题管理 : Kafka Controller 可以帮助我们完成对 Kafka 主题创建、删除和增加分区的操作，简而言之就是对分区拥有最高行使权。
换句话说，当我们执行kafka-topics 脚本时，大部分的后台工作都是控制器来完成的。
- 分区重分配: 分区重分配主要是指，kafka-reassign-partitions 脚本提供的对已有主题分区进行细粒度的分配功能。这部分功能也是控制器实现的。
- Prefered 领导者选举 : Preferred 领导者选举主要是 Kafka 为了避免部分 Broker 负载过重而提供的一种换 Leader 的方案。
- 集群成员管理: 主要管理 新增 broker、broker 关闭、broker 宕机
- 数据服务: 控制器的最后一大类工作，就是向其他 broker 提供数据服务。控制器上保存了最全的集群元数据信息，其他所有 broker 会定期接收控制器发来的元数据更新请求，从而更新其内存中的缓存数据。这些数据我们会在下面讨论
当控制器发现一个 broker 离开集群（通过观察相关 ZooKeeper 路径），控制器会收到消息：这个 broker 所管理的那些分区需要一个新的
Leader。控制器会依次遍历每个分区，确定谁能够作为新的 Leader，然后向所有包含新 Leader 或现有 Follower
的分区发送消息，该请求消息包含谁是新的 Leader 以及谁是 Follower 的信息。随后，新的 Leader
开始处理来自生产者和消费者的请求，Follower 用于从新的 Leader 那里进行复制。
这就很像外包公司的一个部门，这个部门就是专门出差的，每个人在不同的地方办公，但是中央总部有一个部门经理，现在部门经理突然离职了。公司不打算外聘人员，决定从部门内部选一个能力强的人当领导，然后当上领导的人需要向自己的组员发送消息，这条消息就是任命消息和明确他管理了哪些人，大家都知道了，然后再各自给部门干活。
当控制器发现一个 broker 加入集群时，它会使用 broker ID 来检查新加入的 broker
是否包含现有分区的副本。如果有控制器就会把消息发送给新加入的 broker 和 现有的 broker。
broker controller 数据存储
上面我们介绍到 broker controller 会提供数据服务，用于保存大量的 Kafka 集群数据。如下图
image.png
可以对上面保存信息归类，主要分为三类
- broker 上的所有信息，包括 broker 中的所有分区，broker 所有分区副本，当前都有哪些运行中的 broker，哪些正在关闭中的 broker 。
- 所有主题信息，包括具体的分区信息，比如领导者副本是谁，ISR 集合中有哪些副本等。
- 所有涉及运维任务的分区。包括当前正在进行 Preferred 领导者选举以及分区重分配的分区列表。
Kafka 是离不开 ZooKeeper的，所以这些数据信息在 ZooKeeper 中也保存了一份。每当控制器初始化时，它都会从 ZooKeeper
上读取对应的元数据并填充到自己的缓存中。
broker controller 故障转移
我们在前面说过，第一个在 ZooKeeper 中的 /brokers/ids下创建节点的 broker 作为 broker controller，也就是说
broker controller 只有一个，那么必然会存在单点失效问题。kafka 为考虑到这种情况提供了故障转移功能，也就是 Fail Over。如下图
image.png
最一开始，broker1 会抢先注册成功成为 controller，然后由于网络抖动或者其他原因致使 broker1 掉线，ZooKeeper 通过
Watch 机制觉察到 broker1 的掉线，之后所有存活的 brokers 开始竞争成为 controller，这时 broker3 抢先注册成功，此时
ZooKeeper 存储的 controller 信息由 broker1 -> broker3，之后，broker3 会从 ZooKeeper
中读取元数据信息，并初始化到自己的缓存中。
注意：ZooKeeper 中存储的不是缓存信息，broker 中存储的才是缓存信息。
broker controller 存在的问题
在 Kafka 0.11 版本之前，控制器的设计是相当繁琐的。我们上面提到过一句话：Kafka controller
被设计为一种模拟状态机的多线程控制器，这种设计其实是存在一些问题的
- controller 状态的更改由不同的监听器并罚执行，因此需要进行很复杂的同步，并且容易出错而且难以调试。
- 状态传播不同步，broker 可能在时间不确定的情况下出现多种状态，这会导致不必要的额外的数据丢失
- controller 控制器还会为主题删除创建额外的 I/O 线程，导致性能损耗
- controller 的多线程设计还会访问共享数据，我们知道，多线程访问共享数据是线程同步最麻烦的地方，为了保护数据安全性，控制器不得不在代码中大量使用ReentrantLock 同步机制，这就进一步拖慢了整个控制器的处理速度。
broker controller 内部设计原理
在 Kafka 0.11 之后，Kafka controller 采用了新的设计，把多线程的方案改成了单线程加事件队列的方案。如下图所示
image.png
主要所做的改变有下面这几点
第一个改进是增加了一个 Event Executor Thread，事件执行线程，从图中可以看出，不管是 Event Queue 事件队列还是
Controller context
控制器上下文都会交给事件执行线程进行处理。将原来执行的操作全部建模成一个个独立的事件，发送到专属的事件队列中，供此线程消费
第二个改进是将之前同步的 ZooKeeper 全部改为异步操作。ZooKeeper API 提供了两种读写的方式：同步和异步。之前控制器操作
ZooKeeper 都是采用的同步方式，这次把同步方式改为异步，据测试，效率提升了10倍。
第三个改进是根据优先级处理请求，之前的设计是 broker 会公平性的处理所有 controller
发送的请求。什么意思呢？公平性难道还不好吗？在某些情况下是的，比如 broker 在排队处理 produce 请求，这时候 controller 发出了一个
StopReplica 的请求，你会怎么办？还在继续处理 produce 请求吗？这个 produce 请求还有用吗？此时最合理的处理顺序应该是，赋予
StopReplica 请求更高的优先级，使它能够得到抢占式的处理。
副本机制
复制功能是 Kafka 架构的核心功能，在 Kafka 文档里面 Kafka 把自己描述为
一个分布式的、可分区的、可复制的提交日志服务。复制之所以这么关键，是因为消息的持久存储非常重要，这能够保证在主节点宕机后依旧能够保证 Kafka
高可用。副本机制也可以称为备份机制(Replication)，通常指分布式系统在多台网络交互的机器上保存有相同的数据备份/拷贝。
Kafka 使用主题来组织数据，每个主题又被分为若干个分区，分区会部署在一到多个 broker 上，每个分区都会有多个副本，所以副本也会被保存在
broker 上，每个 broker 可能会保存成千上万个副本。下图是一个副本复制示意图
image.png
如上图所示，为了简单我只画出了两个 broker ,每个 broker 指保存了一个 Topic 的消息，在 broker1 中分区0
是Leader，它负责进行分区的复制工作，把 broker1 中的分区0复制一个副本到 broker2 的主题 A 的分区0。同理，主题 A
的分区1也是一样的道理。
副本类型分为两种：一种是 Leader(领导者) 副本，一种是Follower(跟随者)副本。
Leader 副本
Kafka 在创建分区的时候都要选举一个副本，这个选举出来的副本就是 Leader 领导者副本。
Follower 副本
除了 Leader 副本以外的副本统称为 Follower 副本，Follower 不对外提供服务。下面是 Leader 副本的工作方式
image.png
- Kafka 中，Follower 副本也就是追随者副本是不对外提供服务的。这就是说，任何一个追随者副本都不能响应消费者和生产者的请求。所有的请求都是由领导者副本来处理。或者说，所有的请求都必须发送到 Leader 副本所在的 broker 中，Follower 副本只是用做数据拉取，采用异步拉取的方式，并写入到自己的提交日志中，从而实现与 Leader 的同步
- 当 Leader 副本所在的 broker 宕机后，Kafka 依托于 ZooKeeper 提供的监控功能能够实时感知到，并开启新一轮的选举，从追随者副本中选一个作为 Leader。如果宕机的 broker 重启完成后，该分区的副本会作为 Follower 重新加入。
首领的另一个任务是搞清楚哪个跟随者的状态与自己是一致的。跟随者为了保证与领导者的状态一致，在有新消息到达之前先尝试从领导者那里复制消息。为了与领导者保持一致，跟随者向领导者发起获取数据的请求，这种请求与消费者为了读取消息而发送的信息是一样的。
跟随者向领导者发送消息的过程是这样的，先请求消息1，然后再接收到消息1，在时候到请求1之后，发送请求2，在收到领导者给发送给跟随者之前，跟随者是不会继续发送消息的。这个过程如下
image.png
跟随者副本在收到响应消息前，是不会继续发送消息，这一点很重要。通过查看每个跟随者请求的最新偏移量，首领就会知道每个跟随者复制的进度。如果跟随者在10s
内没有请求任何消息，或者虽然跟随者已经发送请求，但是在10s
内没有收到消息，就会被认为是不同步的。如果一个副本没有与领导者同步，那么在领导者掉线后，这个副本将不会称为领导者，因为这个副本的消息不是全部的。
与之相反的，如果跟随者同步的消息和领导者副本的消息一致，那么这个跟随者副本又被称为同步的副本。也就是说，如果领导者掉线，那么只有同步的副本能够称为领导者。
同步复制和异步复制
我在学习副本机制的时候，有个疑问，既然领导者副本和跟随者副本是发送 -
等待机制的，这是一种同步的复制方式，那么为什么说跟随者副本同步领导者副本的时候是一种异步操作呢？
我认为是这样的，跟随者副本在同步领导者副本后会把消息保存在本地 log
中，这个时候跟随者会给领导者副本一个响应消息，告诉领导者自己已经保存成功了，同步复制的领导者会等待所有的跟随者副本都写入成功后，再返回给 producer
写入成功的消息。而异步复制是领导者副本不需要关心跟随者副本是否写入成功，只要领导者副本自己把消息保存到本地 log ，就会返回给 producer
写入成功的消息。下面是同步复制和异步复制的过程
ISR
Kafka动态维护了一个同步状态的副本的集合（a set of In-Sync Replicas），简称ISR，ISR
也是一个很重要的概念，我们之前说过，追随者副本不提供服务，只是定期的异步拉取领导者副本的数据而已，拉取这个操作就相当于是复制，ctrl-c ctrl-
v大家肯定用的熟。那么是不是说 ISR 集合中的副本消息的数量都会与领导者副本消息数量一样呢？那也不一定，判断的依据是 broker 中参数
replica.lag.time.max.ms 的值，这个参数的含义就是跟随者副本能够落后领导者副本最长的时间间隔。
replica.lag.time.max.ms 参数默认的时间是 10秒，如果跟随者副本落后领导者副本的时间不超过 10秒，那么 Kafka
就认为领导者和跟随者是同步的。即使此时跟随者副本中存储的消息要小于领导者副本。如果跟随者副本要落后于领导者副本 10秒以上的话，跟随者副本就会从 ISR
被剔除。倘若该副本后面慢慢地追上了领导者的进度，那么它是能够重新被加回 ISR 的。这也表明，ISR 是一个动态调整的集合，而非静态不变的。
Unclean 领导者选举
既然 ISR 是可以动态调整的，那么必然会出现 ISR 集合中为空的情况，由于领导者副本是一定出现在 ISR 集合中的，那么 ISR
集合为空必然说明领导者副本也挂了，所以此时 Kafka 需要重新选举一个新的领导者，那么该如何选举呢？现在你需要转变一下思路，我们上面说 ISR
集合中一定是与领导者同步的副本，那么不再 ISR 集合中的副本一定是不与领导者同步的副本了，也就是不再 ISR 列表中的跟随者副本会丢失一些消息。如果你开启
broker 端参数
unclean.leader.election.enable的话，下一个领导者就会在这些非同步的副本中选举。这种选举也叫做Unclean 领导者选举。
如果你接触过分布式项目的话你一定知道 CAP 理论，那么这种 Unclean 领导者选举其实是牺牲了数据一致性，保证了 Kafka 的高可用性。
你可以根据你的实际业务场景决定是否开启 Unclean 领导者选举，一般不建议开启这个参数，因为数据的一致性要比可用性重要的多。
Kafka 请求处理流程
broker 的大部分工作是处理客户端、分区副本和控制器发送给分区领导者的请求。这种请求一般都是请求/响应式的，我猜测你接触最早的请求/响应的方式应该就是
HTTP 请求了。事实上，HTTP 请求可以是同步可以是异步的。一般正常的 HTTP
请求都是同步的，同步方式最大的一个特点是提交请求->等待服务器处理->处理完毕返回 这个期间客户端浏览器不能做任何事。而异步方式最大的特点是
请求通过事件触发->服务器处理（这是浏览器仍然可以作其他事情）-> 处理完毕。
这里需要注意一点，我们只是使用 HTTP 请求来举例子，而 Kafka 采用的是 TCP 基于 Socket 的方式进行通讯
那么这两种方式有什么缺点呢？
我相信聪明的你应该能马上想到，同步的方式最大的缺点就是吞吐量太差，资源利用率极低，由于只能顺序处理请求，因此，每个请求都必须等待前一个请求处理完毕才能得到处理。这种方式只适用于请求发送非常不频繁的系统。
异步的方式的缺点就是为每个请求都创建线程的做法开销极大，在某些场景下甚至会压垮整个服务。
响应式模型
说了这么半天，Kafka 采用同步还是异步的呢？都不是，Kafka 采用的是一种
响应式(Reactor)模型，那么什么是响应式模型呢？简单的说，Reactor
模式是事件驱动架构的一种实现方式，特别适合应用于处理多个客户端并发向服务器端发送请求的场景，如下图所示
image.png
Kafka 的 broker 端有个 SocketServer组件，类似于处理器，SocketServer 是基于 TCP 的 Socket
连接的，它用于接受客户端请求，所有的请求消息都包含一个消息头，消息头中都包含如下信息
- Request type （也就是 API Key）
- Request version（broker 可以处理不同版本的客户端请求，并根据客户版本做出不同的响应）
- Correlation ID --- 一个具有唯一性的数字，用于标示请求消息，同时也会出现在响应消息和错误日志中（用于诊断问题）
- Client ID --- 用于标示发送请求的客户端
broker 会在它所监听的每一个端口上运行一个 Acceptor 线程，这个线程会创建一个连接，并把它交给 Processor(网络线程池)，
Processor 的数量可以使用 num.network.threads 进行配置，其默认值是3，表示每台 broker
启动时会创建3个线程，专门处理客户端发送的请求。
broker 会在它所监听的每一个端口上运行一个 Acceptor 线程，这个线程会创建一个连接，并把它交给 Processor(网络线程池)，
Processor 的数量可以使用 num.network.threads 进行配置，其默认值是3，表示每台 broker
启动时会创建3个线程，专门处理客户端发送的请求。
Acceptor
线程会采用轮询的方式将入栈请求公平的发送至网络线程池中，因此，在实际使用过程中，这些线程通常具有相同的机率被分配到待处理请求队列中，然后从响应队列获取响应消息，把它们发送给客户端。Processor
网络线程池中的请求 - 响应的处理还是比较复杂的，下面是网络线程池中的处理流程图
image.png
Processor 网络线程池接收到客户和其他 broker
发送来的消息后，网络线程池会把消息放到请求队列中，注意这个是共享请求队列，因为网络线程池是多线程机制的，所以请求队列的消息是多线程共享的区域，然后由 IO
线程池进行处理，根据消息的种类判断做何处理，比如 PRODUCE 请求，就会将消息写入到 log
日志中，如果是FETCH请求，则从磁盘或者页缓存中读取消息。也就是说，IO线程池是真正做判断，处理请求的一个组件。在IO
线程池处理完毕后，就会判断是放入响应队列中还是 Purgatory 中，Purgatory
是什么我们下面再说，现在先说一下响应队列，响应队列是每个线程所独有的，因为响应式模型中不会关心请求发往何处，因此把响应回传的事情就交给每个线程了，所以也就不必共享了。
注意：IO 线程池可以通过 broker 端参数 num.io.threads 来配置，默认的线程数是8，表示每台 broker 启动后自动创建 8 个IO
处理线程。
请求类型
- 生产请求undefinedacks 这个配置项的含义
简单来讲就是不同的配置对写入成功的界定是不同的，如果 acks = 1，那么只要领导者收到消息就表示写入成功，如果acks =
0，表示只要producer发送消息就表示写入成功，根本不用考虑返回值的影响。如果 acks =
all，就表示领导者需要收到所有副本的消息后才表示写入成功。
在消息被写入分区的首领后，如果 acks 配置的值是 all，那么这些请求会被保存在
炼狱(Purgatory)的缓冲区中，直到领导者副本发现跟随者副本都复制了消息，响应才会发送给客户端。
- 获取请求undefinedbroker 获取请求的方式与处理生产请求的方式类似，客户端发送请求，向 broker 请求主题分区中特定偏移量的消息，如果偏移量存在，Kafka 会采用 零复制 技术向客户端发送消息，Kafka 会直接把消息从文件中发送到网络通道中，而不需要经过任何的缓冲区，从而获得更好的性能。
客户端可以设置获取请求数据的上限和下限，上限指的是客户端为接受足够消息分配的内存空间，这个限制比较重要，如果上限太大的话，很有可能直接耗尽客户端内存。下限可以理解为攒足了数据包再发送的意思，这就相当于项目经理给程序员分配了
10 个bug，程序员每次改一个 bug
就会向项目经理汇报一下，有的时候改好了有的时候可能还没改好，这样就增加了沟通成本和时间成本，所以下限值得就是程序员你改完10个 bug
再向我汇报！！！如下图所示
image.png
如图你可以看到，在拉取消息 ---> 消息
之间是有一个等待消息积累这么一个过程的，这个消息积累你可以把它想象成超时时间，不过超时会跑出异常，消息积累超时后会响应回执。延迟时间可以通过
replica.lag.time.max.ms 来配置，它指定了副本在复制消息时可被允许的最大延迟时间。
元数据请求
生产请求和响应请求都必须发送给领导者副本，如果 broker 收到一个针对某个特定分区的请求，而该请求的首领在另外一个 broker
中，那么发送请求的客户端会收到非分区首领的错误响应；如果针对某个分区的请求被发送到不含有领导者的 broker 上，也会出现同样的错误。Kafka
客户端需要把请求和响应发送到正确的 broker 上。这不是废话么？我怎么知道要往哪发送？
事实上，客户端会使用一种 元数据请求
，这种请求会包含客户端感兴趣的主题列表，服务端的响应消息指明了主题的分区，领导者副本和跟随者副本。元数据请求可以发送给任意一个 broker，因为所有的
broker 都会缓存这些信息。
一般情况下，客户端会把这些信息缓存，并直接向目标 broker
发送生产请求和相应请求，这些缓存需要隔一段时间就进行刷新，使用metadata.max.age.ms 参数来配置，从而知道元数据是否发生了变更。比如，新的
broker 加入后，会触发重平衡，部分副本会移动到新的 broker 上。这时候，如果客户端收到 不是首领的错误，客户端在发送请求之前刷新元数据缓存。
