同步复制和异步复制
我在学习副本机制的时候，有个疑问，既然领导者副本和跟随者副本是发送 - 等待机制的，这是一种同步的复制方式，那么为什么说跟随者副本同步领导者副本的时候是一种异步操作呢？
我认为是这样的，跟随者副本在同步领导者副本后会把消息保存在本地 log 中，这个时候跟随者会给领导者副本一个响应消息，告诉领导者自己已经保存成功了，同步复制的领导者会等待所有的跟随者副本都写入成功后，再返回给 producer 写入成功的消息。而异步复制是领导者副本不需要关心跟随者副本是否写入成功，只要领导者副本自己把消息保存到本地 log ，就会返回给 producer 写入成功的消息。下面是同步复制和异步复制的过程
同步复制
- producer 通知 ZooKeeper 识别领导者
- producer 向领导者写入消息
- 领导者收到消息后会把消息写入到本地 log
- 跟随者会从领导者那里拉取消息
- 跟随者向本地写入 log
- 跟随者向领导者发送写入成功的消息
- 领导者会收到所有的跟随者发送的消息
- 领导者向 producer 发送写入成功的消息
异步复制
和同步复制的区别在于，领导者在写入本地log之后，直接向客户端发送写入成功消息，不需要等待所有跟随者复制完成。
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 模式是事件驱动架构的一种实现方式，特别适合应用于处理多个客户端并发向服务器端发送请求的场景，如下图所示
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个线程，专门处理客户端发送的请求。
Acceptor 线程会采用轮询的方式将入栈请求公平的发送至网络线程池中，因此，在实际使用过程中，这些线程通常具有相同的机率被分配到待处理请求队列中，然后从响应队列获取响应消息，把它们发送给客户端。Processor 网络线程池中的请求 - 响应的处理还是比较复杂的，下面是网络线程池中的处理流程图
Processor 网络线程池接收到客户和其他 broker 发送来的消息后，网络线程池会把消息放到请求队列中，注意这个是共享请求队列，因为网络线程池是多线程机制的，所以请求队列的消息是多线程共享的区域，然后由 IO 线程池进行处理，根据消息的种类判断做何处理，比如 PRODUCE 请求，就会将消息写入到 log 日志中，如果是FETCH请求，则从磁盘或者页缓存中读取消息。也就是说，IO线程池是真正做判断，处理请求的一个组件。在IO 线程池处理完毕后，就会判断是放入响应队列中还是 Purgatory 中，Purgatory 是什么我们下面再说，现在先说一下响应队列，响应队列是每个线程所独有的，因为响应式模型中不会关心请求发往何处，因此把响应回传的事情就交给每个线程了，所以也就不必共享了。
注意：IO 线程池可以通过 broker 端参数
num.io.threads来配置，默认的线程数是8，表示每台 broker 启动后自动创建 8 个IO 处理线程。
请求类型
下面是几种常见的请求类型
生产请求
我在 真的，关于 Kafka 入门看这一篇就够了 文章中提到过 acks 这个配置项的含义
简单来讲就是不同的配置对写入成功的界定是不同的，如果 acks = 1，那么只要领导者收到消息就表示写入成功，如果acks = 0，表示只要领导者发送消息就表示写入成功，根本不用考虑返回值的影响。如果 acks = all，就表示领导者需要收到所有副本的消息后才表示写入成功。
在消息被写入分区的首领后，如果 acks 配置的值是 all，那么这些请求会被保存在 炼狱(Purgatory)的缓冲区中，直到领导者副本发现跟随者副本都复制了消息，响应才会发送给客户端。
获取请求
broker 获取请求的方式与处理生产请求的方式类似，客户端发送请求，向 broker 请求主题分区中特定偏移量的消息，如果偏移量存在，Kafka 会采用 零复制 技术向客户端发送消息，Kafka 会直接把消息从文件中发送到网络通道中，而不需要经过任何的缓冲区，从而获得更好的性能。
客户端可以设置获取请求数据的上限和下限，上限指的是客户端为接受足够消息分配的内存空间，这个限制比较重要，如果上限太大的话，很有可能直接耗尽客户端内存。下限可以理解为攒足了数据包再发送的意思，这就相当于项目经理给程序员分配了 10 个bug，程序员每次改一个 bug 就会向项目经理汇报一下，有的时候改好了有的时候可能还没改好，这样就增加了沟通成本和时间成本，所以下限值得就是程序员你改完10个 bug 再向我汇报！！！如下图所示
如图你可以看到，在拉取消息 ---> 消息 之间是有一个等待消息积累这么一个过程的，这个消息积累你可以把它想象成超时时间，不过超时会跑出异常，消息积累超时后会响应回执。延迟时间可以通过 replica.lag.time.max.ms 来配置，它指定了副本在复制消息时可被允许的最大延迟时间。
元数据请求
生产请求和响应请求都必须发送给领导者副本，如果 broker 收到一个针对某个特定分区的请求，而该请求的首领在另外一个 broker 中，那么发送请求的客户端会收到非分区首领的错误响应；如果针对某个分区的请求被发送到不含有领导者的 broker 上，也会出现同样的错误。Kafka 客户端需要把请求和响应发送到正确的 broker 上。这不是废话么？我怎么知道要往哪发送？
事实上，客户端会使用一种 元数据请求 ，这种请求会包含客户端感兴趣的主题列表，服务端的响应消息指明了主题的分区，领导者副本和跟随者副本。元数据请求可以发送给任意一个 broker，因为所有的 broker 都会缓存这些信息。
一般情况下，客户端会把这些信息缓存，并直接向目标 broker 发送生产请求和相应请求，这些缓存需要隔一段时间就进行刷新，使用metadata.max.age.ms 参数来配置，从而知道元数据是否发生了变更。比如，新的 broker 加入后，会触发重平衡，部分副本会移动到新的 broker 上。这时候，如果客户端收到 不是首领的错误，客户端在发送请求之前刷新元数据缓存。
Kafka 重平衡流程
我在 真的，关于 Kafka 入门看这一篇就够了 中关于消费者描述的时候大致说了一下消费者组和重平衡之间的关系，实际上，归纳为一点就是让组内所有的消费者实例就消费哪些主题分区达成一致。
我们知道，一个消费者组中是要有一个群组协调者(Coordinator)的，而重平衡的流程就是由 Coordinator 的帮助下来完成的。
这里需要先声明一下重平衡发生的条件
- 消费者订阅的任何主题发生变化
- 消费者数量发生变化
- 分区数量发生变化
- 如果你订阅了一个还尚未创建的主题，那么重平衡在该主题创建时发生。如果你订阅的主题发生删除那么也会发生重平衡
- 消费者被群组协调器认为是
DEAD状态，这可能是由于消费者崩溃或者长时间处于运行状态下发生的，这意味着在配置合理时间的范围内，消费者没有向群组协调器发送任何心跳，这也会导致重平衡的发生。
在了解重平衡之前，你需要知道这两个角色
群组协调器（Coordinator）：群组协调器是一个能够从消费者群组中收到所有消费者发送心跳消息的 broker。在最早期的版本中，元数据信息是保存在 ZooKeeper 中的，但是目前元数据信息存储到了 broker 中。每个消费者组都应该和群组中的群组协调器同步。当所有的决策要在应用程序节点中进行时，群组协调器可以满足 JoinGroup 请求并提供有关消费者组的元数据信息，例如分配和偏移量。群组协调器还有权知道所有消费者的心跳，消费者群组中还有一个角色就是领导者，注意把它和领导者副本和 kafka controller 进行区分。领导者是群组中负责决策的角色，所以如果领导者掉线了，群组协调器有权把所有消费者踢出组。因此，消费者群组的一个很重要的行为是选举领导者，并与协调器读取和写入有关分配和分区的元数据信息。
消费者领导者：每个消费者群组中都有一个领导者。如果消费者停止发送心跳了，协调者会触发重平衡。
在了解重平衡之前，你需要知道状态机是什么
Kafka 设计了一套消费者组状态机(State Machine) ，来帮助协调者完成整个重平衡流程。消费者状态机主要有五种状态它们分别是 Empty、Dead、PreparingRebalance、CompletingRebalance 和 Stable。
了解了这些状态的含义之后，下面我们用几条路径来表示一下消费者状态的轮转
- 消费者组一开始处于
Empty状态，当重平衡开启后，它会被置于PreparingRebalance状态等待新消费者的加入，一旦有新的消费者加入后，消费者群组就会处于CompletingRebalance状态等待分配，只要有新的消费者加入群组或者离开，就会触发重平衡，消费者的状态处于 PreparingRebalance 状态。等待分配机制指定好后完成分配，那么它的流程图是这样的
- 在上图的基础上，当消费者群组都到达
Stable状态后，一旦有新的消费者加入/离开/心跳过期，那么触发重平衡，消费者群组的状态重新处于 PreparingRebalance 状态。那么它的流程图是这样的。
- 在上图的基础上，消费者群组处于 PreparingRebalance 状态后，很不幸，没人玩儿了，所有消费者都离开了，这时候还可能会保留有消费者消费的位移数据，一旦位移数据过期或者被刷新，那么消费者群组就处于
Dead状态了。它的流程图是这样的
- 在上图的基础上，我们分析了消费者的重平衡，在
PreparingRebalance或者CompletingRebalance或者Stable任意一种状态下发生位移主题分区 Leader 发生变更，群组会直接处于 Dead 状态，它的所有路径如下
这里面需要注意两点：
一般出现 Required xx expired offsets in xxx milliseconds 就表明Kafka 很可能就把该组的位移数据删除了
只有 Empty 状态下的组，才会执行过期位移删除的操作。
重平衡流程
上面我们了解到了消费者群组状态的转化过程，下面我们真正开始介绍 Rebalance 的过程。重平衡过程可以从两个方面去看：消费者端和协调者端，首先我们先看一下消费者端
从消费者看重平衡
从消费者看重平衡有两个步骤：分别是 消费者加入组 和 等待领导者分配方案。这两个步骤后分别对应的请求是 JoinGroup 和 SyncGroup。
新的消费者加入群组时，这个消费者会向协调器发送 JoinGroup 请求。在该请求中，每个消费者成员都需要将自己消费的 topic 进行提交，我们上面描述群组协调器中说过，这么做的目的就是为了让协调器收集足够的元数据信息，来选取消费者组的领导者。通常情况下，第一个发送 JoinGroup 请求的消费者会自动称为领导者。领导者的任务是收集所有成员的订阅信息，然后根据这些信息，制定具体的分区消费分配方案。如图
在所有的消费者都加入进来并把元数据信息提交给领导者后，领导者做出分配方案并发送 SyncGroup请求给协调者，协调者负责下发群组中的消费策略。下图描述了 SyncGroup 请求的过程
当所有成员都成功接收到分配方案后，消费者组进入到 Stable 状态，即开始正常的消费工作。
从协调者来看重平衡
从协调者角度来看重平衡主要有下面这几种触发条件，
- 新成员加入组
- 组成员主动离开
- 组成员崩溃离开
- 组成员提交位移
我们分别来描述一下，先从新成员加入组开始
新成员入组
我们讨论的场景消费者集群状态处于Stable 等待分配的过程，这时候如果有新的成员加入组的话，重平衡的过程
从这个角度来看，协调者的过程和消费者类似，只是刚刚从消费者的角度去看，现在从领导者的角度去看
组成员离开
组成员离开消费者群组指的是消费者实例调用 close() 方法主动通知协调者它要退出。这里又会有一个新的请求出现 LeaveGroup()请求 。如下图所示
组成员崩溃
组成员崩溃是指消费者实例出现严重故障，宕机或者一段时间未响应，协调者接收不到消费者的心跳，就会被认为是组成员崩溃，崩溃离组是被动的，协调者通常需要等待一段时间才能感知到，这段时间一般是由消费者端参数 session.timeout.ms 控制的。如下图所示
重平衡时提交位移
这个过程我们就不再用图形来表示了，大致描述一下就是 消费者发送 JoinGroup 请求后，群组中的消费者必须在指定的时间范围内提交各自的位移，然后再开启正常的 JoinGroup/SyncGroup 请求发送。
