Kafka系列教程 - Kafka 集群 -4

1. Kafka 和 ZooKeeper

Kafka 使用 Zookeeper 来维护集群成员的信息。每个 Broker 都有一个唯一标识符，这个标识符可以在配置文件里指定，也可以自动生成。在 Broker 启动的时候，它通过创建临时节点把自己的 ID 注册到 Zookeeper。Kafka 组件订阅 Zookeeper 的 /broker/ids 路径，当有 Broker 加入集群或退出集群时，这些组件就可以获得通知。

如果要启动另一个具有相同 ID 的 Broker，会得到一个错误——新 Broker 会试着进行注册，但不会成功，因为 ZooKeeper 中已经有一个具有相同 ID 的 Broker。

在 Broker 停机、出现网络分区或长时间垃圾回收停顿时，Broker 会与 ZooKeeper 断开连接，此时 Broker 在启动时创建的临时节点会自动被 ZooKeeper 移除。监听 Broker 列表的 Kafka 组件会被告知 Broker 已移除。

Kafka 在 ZooKeeper 的关键存储信息：

• admin：存储管理信息。主要为删除主题事件，分区迁移事件，优先副本选举，信息 (一般为临时节点)
• brokers：存储 Broker 相关信息。broker 节点以及节点上的主题相关信息
• cluster：存储 kafka 集群信息
• config：存储 broker，client，topic，user 以及 changer 相关的配置信息
• consumers：存储消费者相关信息
• controller：存储控制器节点信息
• controller_epoch：存储控制器节点当前的年龄（说明控制器节点变更次数）

ZooKeeper 两个重要特性：

• 客户端会话结束时，ZooKeeper 就会删除临时节点。
• 客户端注册监听它关心的节点，当节点状态发生变化（数据变化、子节点增减变化）时，ZooKeeper 服务会通知客户端。

2. 控制器

控制器（Controller），是 Apache Kafka 的核心组件。它的主要作用是在 ZooKeeper 的帮助下管理和协调整个 Kafka 集群。控制器其实就是一个 Broker，只不过它除了具有一般 Broker 的功能以外，还负责 Leader 的选举。

2.1. 如何选举控制器

集群中任意一台 Broker 都能充当控制器的角色，但是，在运行过程中，只能有一个 Broker 成为控制器，行使其管理和协调的职责。实际上，Broker 在启动时，会尝试去 ZooKeeper 中创建 /controller 节点。Kafka 当前选举控制器的规则是：第一个在 ZooKeeper 成功创建 /controller 临时节点的 Broker 会被指定为控制器。

选举控制器的详细流程：

 基于 ZooKeeper 的控制器选举（传统模式）

1. 第一个在 ZooKeeper 中成功创建 /controller 临时节点的 Broker 会被指定为控制器。

2. 其他 Broker 在控制器节点上创建 Zookeeper watch 对象。

3. 如果控制器被关闭或者与 Zookeeper 断开连接，Zookeeper 临时节点就会消失。集群中的其他 Broker 通过 watch 对象得到状态变化的通知，它们会尝试让自己成为新的控制器。

4. 第一个在 Zookeeper 里创建一个临时节点 /controller 的 Broker 成为新控制器。其他 Broker 在新控制器节点上创建 Zookeeper watch 对象。

5. 每个新选出的控制器通过 Zookeeper 的条件递增操作获得一个全新的、数值更大的 controller epoch。其他节点会忽略旧的 epoch 的消息。

6. 当控制器发现一个 Broker 已离开集群，并且这个 Broker 是某些 Partition 的 Leader。此时，控制器会遍历这些 Partition，并用轮询方式确定谁应该成为新 Leader，随后，新 Leader 开始处理生产者和消费者的请求，而 Follower 开始从 Leader 那里复制消息。

简而言之，Kafka 使用 Zookeeper 的临时节点来选举控制器，并在节点加入集群或退出集群时通知控制器。控制器负责在节点加入或离开集群时进行 Partition Leader 选举。控制器使用 epoch 来避免“脑裂”，“脑裂”是指两个节点同时被认为自己是当前的控制器。

基于 KRaft 的控制器选举（无 ZooKeeper 模式）

Kafka 的 KRaft 模式（Kafka Raft）移除了对 ZooKeeper 的依赖，改用内部的 Raft 协议进行控制器选举。

1. Raft 集群：

   • Kafka 的元数据管理由一个内部的 Raft 集群负责。
   • Raft 集群中的成员可以是 Broker，但不是所有 Broker 都是元数据节点。

2. 选举机制：

   • 在 Raft 协议中，控制器是 Raft 的 Leader 节点。
   • 当集群启动或 Leader 失效时，所有 Follower 节点会发起选举，最终通过大多数（quorum）投票选出新的 Leader。

3. 控制器职责：

   • 控制器直接从 Raft 日志中读取或写入元数据，而不需要 ZooKeeper。
   • 它与传统模式相比，减少了外部依赖，提升了系统的一致性和可用性。

控制器选举的优先级

Kafka 的控制器选举并没有特别的优先级规则，而是基于节点尝试创建 /controller 节点的顺序（ZooKeeper 模式）或 Raft 的内部逻辑（KRaft 模式）。但是可以通过以下方式间接影响控制器选举：

1. 优先选择性能较好的 Broker：

   • 将性能较好的 Broker 启动得更早，增加其成为控制器的概率。

2. 避免频繁控制器切换：

   • 设置 zookeeper.session.timeout.ms 足够长，避免因短暂网络问题导致控制器切换。

3. 监控和告警：

   • 通过 Kafka 的监控工具检测控制器变更（如 kafka.controller:type=KafkaController,name=ActiveControllerCount），及时发现异常情况。

2.2. 控制器的作用

Topic 管理（创建、删除、增加分区）

这里的 Topic 管理，就是指控制器帮助我们完成对 Kafka Topic 的创建、删除以及分区增加的操作。换句话说，当我们执行 kafka-topics 脚本时，大部分的后台工作都是控制器来完成的。

#分区重分配

分区重分配主要是指，kafka-reassign-partitions 脚本（关于这个脚本，后面我也会介绍）提供的对已有 Topic 分区进行细粒度的分配功能。这部分功能也是控制器实现的。

#选举 Leader

Preferred 领导者选举主要是 Kafka 为了避免部分 Broker 负载过重而提供的一种换 Leader 的方案。在专栏后面说到工具的时候，我们再详谈 Preferred 领导者选举，这里你只需要了解这也是控制器的职责范围就可以了。

#集群成员管理

集群成员管理，包括自动检测新增 Broker、Broker 主动关闭及被动宕机。这种自动检测是依赖于前面提到的 Watch 功能和 ZooKeeper 临时节点组合实现的。

比如，控制器组件会利用Watch 机制检查 ZooKeeper 的 /brokers/ids 节点下的子节点数量变更。目前，当有新 Broker 启动后，它会在 /brokers 下创建专属的 znode 节点。一旦创建完毕，ZooKeeper 会通过 Watch 机制将消息通知推送给控制器，这样，控制器就能自动地感知到这个变化，进而开启后续的新增 Broker 作业。

侦测 Broker 存活性则是依赖于刚刚提到的另一个机制：临时节点。每个 Broker 启动后，会在 /brokers/ids 下创建一个临时 znode。当 Broker 宕机或主动关闭后，该 Broker 与 ZooKeeper 的会话结束，这个 znode 会被自动删除。同理，ZooKeeper 的 Watch 机制将这一变更推送给控制器，这样控制器就能知道有 Broker 关闭或宕机了，从而进行“善后”。

#数据服务

控制器的最后一大类工作，就是向其他 Broker 提供数据服务。控制器上保存了最全的集群元数据信息，其他所有 Broker 会定期接收控制器发来的元数据更新请求，从而更新其内存中的缓存数据。

控制器中保存了多种数据，比较重要的的数据有：

• 所有 Topic 信息。包括具体的分区信息，比如领导者副本是谁，ISR 集合中有哪些副本等。
• 所有 Broker 信息。包括当前都有哪些运行中的 Broker，哪些正在关闭中的 Broker 等。
• 所有涉及运维任务的分区。包括当前正在进行 Preferred 领导者选举以及分区重分配的分区列表。

值得注意的是，这些数据其实在 ZooKeeper 中也保存了一份。每当控制器初始化时，它都会从 ZooKeeper 上读取对应的元数据并填充到自己的缓存中。有了这些数据，控制器就能对外提供数据服务了。这里的对外主要是指对其他 Broker 而言，控制器通过向这些 Broker 发送请求的方式将这些数据同步到其他 Broker 上。

2.3 控制器选举的常见问题

问题 1：频繁的控制器切换

• 原因：
  • 控制器所在的 Broker 经常发生网络抖动或宕机。
  • zookeeper.session.timeout.ms 设置过低。
• 解决方案：
  • 提高 zookeeper.session.timeout.ms，例如从默认的 6 秒增加到 20 秒。
  • 增强控制器所在节点的稳定性，使用高性能的硬件或更可靠的网络连接。

问题 2：控制器选举耗时过长

• 原因：
  • ZooKeeper 性能瓶颈或 Broker 与 ZooKeeper 的连接延迟较高。
  • Broker 启动速度较慢。
• 解决方案：
  • 优化 ZooKeeper 的配置和性能（如增加 ZooKeeper 节点数量或升级硬件）。
  • 确保 Broker 能快速启动并与 ZooKeeper 建立连接。

问题 3：控制器无法选出

• 原因：
  • 所有 Broker 都无法与 ZooKeeper 建立连接。
  • Raft 集群无法达到多数投票（KRaft 模式）。
• 解决方案：
  • 检查 ZooKeeper 集群是否正常运行。
  • 在 KRaft 模式下，确保元数据节点数量为奇数，且大多数节点在线。

监控控制器选举

Kafka 提供了多种方式监控控制器状态和选举过程：

ZooKeeper 模式

1. 检查 ZooKeeper 的 /controller 节点：

   • 使用 ZooKeeper CLI：

     bash

     Copy code

     echo stat /controller | zkCli.sh

   • 查看当前控制器所在的 Broker ID。

2. Kafka 指标：

   • ActiveControllerCount：当前活动控制器的数量（正常值为 1）。
   • ControllerChangeRateAndTimeMs：控制器切换的频率和时间。

KRaft 模式

1. 查看 Raft 集群状态：

   • 使用 Kafka 提供的 Admin 工具查询元数据节点的状态。

2. 监控指标：

   • kafka.controller:type=KafkaController,name=ActiveControllerCount。
   • kafka.controller:type=KafkaController,name=ControllerElectionRateAndTimeMs。

3. 副本机制

• 副本（Replica） 是某个分区的数据副本。一个 Kafka 分区可以有多个副本，副本的数量由分区的 Replication Factor 决定。
• 每个副本存储在不同的 Broker 上，确保副本的分布能够抵抗单点故障。

例如，某个分区的 Replication Factor 为 3，则该分区有三个副本分布在三个不同的 Broker 上。

Kafka 副本类型

Leader 副本

• 每个分区都有一个 Leader 副本，负责处理所有读写请求。
• Leader 副本是分区中当前活跃的副本。
• 生产者和消费者只能与 Leader 副本交互。

Follower 副本

• 其他非 Leader 的副本称为 Follower 副本。
• Follower 副本从 Leader 副本同步数据（称为 Replication）。
• Follower 副本不直接处理客户端请求，只作为备份。

副本机制是分布式系统实现高可用的不二法门，Kafka 也不例外。

副本机制有哪些好处？

1. 提供可用性：有句俗语叫：鸡蛋不要放在一个篮子里。副本机制也是一个道理——当部分节点宕机时，系统仍然可以依靠其他正常运转的节点，从整体上对外继续提供服务。
2. 提供伸缩性：通过增加、减少机器可以控制系统整体的吞吐量。
3. 改善数据局部性：允许将数据放入与用户地理位置相近的地方，从而降低系统延时。

但是，Kafka 只实现了第一个好处，原因后面会阐述。

3.1. Kafka 副本角色

Kafka 使用 Topic 来组织数据，每个 Topic 被分为若干个 Partition，每个 Partition 有多个副本。每个 Broker 可以保存成百上千个属于不同 Topic 和 Partition 的副本。Kafka 副本的本质是一个只能追加写入的提交日志。

Kafka 副本有两种角色：

• Leader 副本（主）：每个 Partition 都有且仅有一个 Leader 副本。为了保证数据一致性，Leader 处理一切对 Partition （分区）的读写请求；
• Follower 副本（从）：Leader 副本以外的副本都是 Follower 副本。Follower 唯一的任务就是从 Leader 那里复制消息，保持与 Leader 一致的状态。
• 如果 Leader 宕机，其中一个 Follower 会被选举为新的 Leader。
• 

为了与 Leader 保持同步，Follower 向 Leader 发起获取数据的请求，这种请求与消费者为了读取消息而发送的请求是一样的。请求消息里包含了 Follower 想要获取消息的偏移量，而这些偏移量总是有序的。

Leader 另一个任务是搞清楚哪个 Follower 的状态与自己是一致的。通过查看每个 Follower 请求的最新偏移量，Leader 就会知道每个 Follower 复制的进度。如果跟随者在 10s 内没有请求任何消息，或者虽然在请求消息，但是在 10s 内没有请求最新的数据，那么它就会被认为是不同步的。如果一个副本是不同步的，在 Leader 失效时，就不可能成为新的 Leader——毕竟它没有包含全部的消息。

除了当前首领之外，每个分区都有一个首选首领——创建 Topic 时选定的首领就是分区的首选首领。之所以叫首选 Leader，是因为在创建分区时，需要在 Broker 之间均衡 Leader。

副本配置与选项

1 副本数量

副本数量由分区的 Replication Factor 配置。示例：

• 如果 Replication Factor = 1，则只有一个副本，无法容错。
• 如果 Replication Factor = 3，则可以容忍最多 2 个 Broker 故障。

示例配置：

--replication-factor 3

2 最小同步副本

• min.insync.replicas：配置生产者在成功写入消息时，至少需要同步到的副本数量。
• 如果可用副本数量少于 min.insync.replicas，生产者会收到错误。

示例配置：

min.insync.replicas=2

3 ACK 确认机制

生产者通过 acks 参数指定消息写入的确认级别：

• acks=0：生产者不等待 Broker 确认。
• acks=1：生产者等待 Leader 副本确认写入。
• acks=all：生产者等待所有 ISR 副本确认写入。

3.2. ISR

ISR 即 In-sync Replicas，表示同步副本。Follower 副本不提供服务，只是定期地异步拉取领导者副本中的数据而已。既然是异步的，说明和 Leader 并非数据强一致性的。

判断 Follower 是否与 Leader 同步的标准：

Kafka Broker 端参数 replica.lag.time.max.ms 参数，指定了 Follower 副本能够落后 Leader 副本的最长时间间隔，默认为 10s。这意味着：只要一个 Follower 副本落后 Leader 副本的时间不连续超过 10 秒，那么 Kafka 就认为该 Follower 副本与 Leader 是同步的，即使此时 Follower 副本中保存的消息明显少于 Leader 副本中的消息。

ISR 是一个动态调整的集合，会不断将同步副本加入集合，将不同步副本移除集合。Leader 副本天然就在 ISR 中。

 副本机制的常见问题

问题 1：ISR 列表收缩

• 现象：某些副本从 ISR 中移除。
• 原因：
  • Follower 副本未能及时同步数据，滞后于 Leader。
  • 网络延迟或复制速度慢。
• 解决方案：
  • 检查网络或磁盘性能，优化 Broker 的硬件配置。
  • 调整复制相关配置参数，如：

    Copy code

    replica.lag.time.max.ms=10000 # 最大滞后时间 replica.fetch.max.bytes=1048576 # 每次拉取数据的大小

问题 2：非同步副本选举

• 现象：数据丢失。
• 原因：允许非同步副本选举导致选出的 Leader 没有最新数据。
• 解决方案：
  • 设置 unclean.leader.election.enable=false。
  • 增加副本数量和 min.insync.replicas。

4. 选举 Leader

4.1. Unclean 领导者选举

因为 Leader 副本天然就在 ISR 中，如果 ISR 为空了，就说明 Leader 副本也“挂掉”了，Kafka 需要重新选举一个新的 Leader。

Kafka 把所有不在 ISR 中的存活副本都称为非同步副本。通常来说，非同步副本落后 Leader 太多，因此，如果选择这些副本作为新 Leader，就可能出现数据的丢失。毕竟，这些副本中保存的消息远远落后于老 Leader 中的消息。在 Kafka 中，选举这种副本的过程称为 Unclean 领导者选举。Broker 端参数 unclean.leader.election.enable 控制是否允许 Unclean 领导者选举。

开启 Unclean 领导者选举可能会造成数据丢失，但好处是：它使得 Partition Leader 副本一直存在，不至于停止对外提供服务，因此提升了高可用性。反之，禁止 Unclean 领导者选举的好处在于维护了数据的一致性，避免了消息丢失，但牺牲了高可用性。

5. 处理请求

Broker 的大部分工作是处理客户端、Partition 副本和控制器发送给 Partition Leader 的请求。Kafka 提供了一个二进制协议（基于 TCP），指定了请求消息的格式以及 Broker 如何对请求作出响应。

broker 会在它所监听的每一个端口上运行一个 Acceptor 线程，这个线程会创建一个连接，并把它交给 Processor 线程去处理。Processor 线程的数量是可配置的。Processor 线程负责从客户端获取请求消息，把它们放进请求队列，然后从响应队列获取响应消息，把它们发送给客户端。

当请求放进请求队列后，IO 线程负责进行处理。

生产请求和获取请求都需要发送给 Partition 的 Leader 副本处理。如果 Broker 收到一个针对特定分区的请求，而该分区的 Leader 在另一个 Broker 上，那么发送请求的客户端会收到一个“非分区 Leader”的错误响应。Kafka 客户端要自己负责把生成请求和获取请求发送到正确的 Broker 上。

5.1. 元数据请求

客户端怎么知道哪个是 Leader 呢？客户端通过使用另一种类型的请求来实现，那就是元数据请求（metadata request）。这种请求包含了客户端感兴趣的 Topic 列表。broker 的响应消息指明了这些 Topic 所包含的 Partition、Partition 有哪些副本，以及哪个副本是 Leader。元数据请求可以发给任意一个 broker，因为所有 Broker 都缓存了这些信息。

客户端会把这些信息缓存起来，并直接往目标 Broker 上发送生产请求和获取请求。它们需要时不时地通过发送元数据请求来刷新这些信息（刷新的时间间隔通过 metadata.max.age.ms 来配置），从而知道元数据是否发生了变化。

5.2. 生产请求

acks 参数控制多少个副本确认写入成功后生产者才认为消息生产成功。这个参数的取值可以为：

• acks=0 - 消息发送完毕，生产者认为消息写入成功；
• acks=1 - Leader 写入成功，生产者认为消息写入成功；
• acks=all - 所有同步副本写入成功，生产者才认为消息写入成功。

如果 Leader 收到生产消息，它会执行一些检查逻辑，包含：

• 发送的用户是否有权限写入 Topic？
• 请求的 acks 参数取值是否合法（只允许 0，1，all）？
• 如果 acks 设置为 all，是否有足够的同步副本已经安全写入消息？（我们可以配置如果同步副本数量不足，Leader 拒绝处理新消息）

之后，消息被写入到本地磁盘。一旦消息本地持久化后，如果 acks 被设为 0 或 1，那么会返回结果给客户端；如果 acks 被设为 all 那么会将请求放置在一个称为 purgatory 的缓冲区中等待其他的副本写入完成。

1. acks 参数的取值

1.1 acks=0

• 定义：
  • 生产者发送消息后不等待 Broker 的任何确认。
• 特点：
  • 极快的吞吐量。
  • 可能导致数据丢失：Broker 可能尚未接收到消息，生产者也认为消息已发送成功。
• 适用场景：
  • 数据可靠性要求较低。
  • 临时日志或无关紧要的数据。

1.2 acks=1

• 定义：
  • 生产者等待 Leader 副本写入日志后返回成功确认。
• 特点：
  • 性能较好。
  • 可能导致数据丢失：如果消息写入 Leader 后，尚未复制到 Follower 副本，Leader 崩溃会导致数据丢失。
• 适用场景：
  • 性能与可靠性之间的折中场景。
  • 中等重要的数据。

1.3 acks=all（或 acks=-1）

• 定义：
  • 生产者等待所有 ISR（In-Sync Replica） 副本成功写入后返回确认。
• 特点：
  • 最高的可靠性：数据在多个副本中被成功复制后，生产者才会收到成功响应。
  • 性能相对较低：需要等待多个副本完成同步。
  • 如果 ISR 副本数量不足，生产者会收到错误，防止数据丢失。
• 适用场景：
  • 数据丢失无法接受的场景，如金融交易、订单处理。

2. acks 参数设置的影响

2.1 数据可靠性

• acks=0：无可靠性保障，可能丢失数据。
• acks=1：中等可靠性，适合对数据丢失容忍度中等的应用。
• acks=all：最高可靠性，确保数据被复制到多个副本后才返回成功。

2.2 延迟

• acks=0：最短延迟，无需等待确认。
• acks=1：延迟取决于 Leader 副本写入的速度。
• acks=all：延迟最高，因为需要等待所有 ISR 副本同步完成。

2.3 吞吐量

• acks=0：最高吞吐量，生产者无阻塞。
• acks=1：吞吐量适中。
• acks=all：吞吐量较低，受副本同步时间的限制。

3. acks 的配置示例

生产者配置

通过 Kafka 生产者 API 配置 acks 参数：

props.put("acks", "all"); // 设置为 "0", "1" 或 "all"

4. 配合其他参数优化

4.1 min.insync.replicas

• 作用：定义一个分区中至少需要与 Leader 同步的副本数量。
• 与 acks 的关系：
  • acks=all 时，min.insync.replicas 生效。
  • 如果同步副本数量小于 min.insync.replicas，生产者会收到错误，防止数据丢失。
• 配置示例：

  min.insync.replicas=2

4.2 retries

• 作用：生产者写入失败时的重试次数。
• 问题：
  • 设置过低可能导致短暂的网络或副本故障影响消息发送。
  • 设置过高可能造成大量重复数据写入。
• 推荐配置：

  retries=3

4.3 linger.ms

• 作用：控制生产者批量发送消息的等待时间。
• 优化方向：
  • 较大的值可以提高吞吐量，但会增加延迟。

  linger.ms=5

5. 常见问题及解决方案

问题 1：数据丢失

• 原因：
  • 使用 acks=0 或 acks=1。
  • 副本数量少，且 Leader 副本发生故障。
• 解决方案：
  • 使用 acks=all，确保数据复制到多个副本。
  • 增加副本数量（Replication Factor）。

问题 2：生产者发送失败

• 原因：
  • ISR 副本数量不足，min.insync.replicas 未满足要求。
• 解决方案：
  • 检查 Broker 状态，确保 ISR 副本在线。
  • 减少 min.insync.replicas（权衡数据可靠性）。

问题 3：延迟过高

• 原因：
  • 使用 acks=all，等待副本同步。
  • 网络或磁盘性能瓶颈。
• 解决方案：
  • 优化网络和磁盘性能。
  • 适当减少 acks 要求或副本数量（如果可以接受数据丢失风险）。

6. 推荐配置

高性能优先

适用于对数据可靠性要求较低的场景：

acks=1 min.insync.replicas=1 retries=3 linger.ms=1

高可靠性优先

适用于关键任务，确保数据不丢失：

acks=all min.insync.replicas=2 retries=5 linger.ms=5

平衡性能与可靠性

适合大多数应用场景：

acks=1 min.insync.replicas=1 retries=3 linger.ms=2

总结

Kafka 的 acks 参数在性能和可靠性之间提供了灵活的选择。对于高可靠性需求的应用，推荐 acks=all 并结合 min.insync.replicas。对于性能要求较高的场景，可以使用 acks=1 或 acks=0。具体设置需根据业务需求、集群规模以及容错能力进行权衡。

5.3. 消费请求

Leader 处理拉取请求和处理生产请求的方式很相似：

1. 请求需要先到达指定的 Partition Leader 上，然后客户端通过查询元数据来确保请求的路由是正确的。
2. Leader 在收到请求时，会先检查请求是否有效。
3. 如果请求的偏移量存在，Broker 将按照客户端指定的数量上限从 Partition 里读取消息，再把消息返回给客户端。Kafka 使用零拷贝技术向客户端发送消息——也就是说，Kafka 直接把消息从文件（更准确的说，是文件系统缓存）里发送到网络通道，而不需要经过任何中间缓冲区。这避免了内存的字节拷贝和缓冲区维护，极大地提高了性能。

客户端可以指定 Broker 返回数据量的上限和下限，防止数据量过大造成客户端内存溢出。同时，客户端也可以指定返回的最小数据量，当消息数据量没有达到最小数据量时，请求会一直阻塞直到有足够的数据返回。指定最小的数据量在负载不高的情况下非常有用，通过这种方式可以减轻网络往返的额外开销。当然请求也不能永远的阻塞，客户端可以指定最大的阻塞时间，如果到达指定的阻塞时间，即便没有足够的数据也会返回。

不是所有 Leader 的数据都能够被读取。消费者只能读取已提交的消息。只有当消息被写入分区的若干同步副本时，才被认为是已提交的。为什么是若干个 Broker 呢？这取决于你对“已提交”的定义。你可以选择只要 Leader 成功保存该消息就算是已提交，也可以是令所有 Broker 都成功保存该消息才算是已提交。

因为还没有被足够的副本持久化的消息，被认为是不安全的——如果 Leader 发生故障，另一个副本成为新的 Leader，这些消息就丢失了。如果允许读取这些消息，就可能会破坏数据一致性。

这也意味着，如果 Broker 间的消息复制因为某些原因变慢了，那么消息到达消费者的时间也会随之变长。延迟时间可以通过 replica.lag.time.max.ms 来配置，它指定了副本在复制消息时可被允许的最大延迟时间。

5.4. 其他请求

我们讨论了 Kafka 中最常见的三种请求类型：元信息请求，生产请求和拉取请求。这些请求都是使用的是 Kafka 的自定义二进制协议。集群中 Broker 间的通信请求也是使用同样的协议，这些请求是内部使用的，客户端不能发送。比如在选举 Partition Leader 过程中，控制器会发送 LeaderAndIsr 请求给新的 Leader 和其他跟随副本。

这个协议目前已经支持 20 种请求类型，并且仍然在演进以支持更多的类型。

6. 总结

6.1. 副本机制

• 每个 Partition 都有一个 Leader，零个或多个 Follower。
• Leader 处理一切对 Partition （分区）的读写请求；而 Follower 只需被动的同步 Leader 上的数据。
• 同一个 Topic 的不同 Partition 会分布在多个 Broker 上，而且一个 Partition 还会在其他的 Broker 上面进行备份。

6.2. 选举机制

Follower 宕机，啥事儿没有；Leader 宕机了，会从 Follower 中重新选举一个新的 Leader。 生产者/消费者如何知道谁是 Leader

• Kafka 将这种元数据存储在 Zookeeper 服务中。
• 生产者和消费者都和 Zookeeper 连接并通信。