原来这才是Kafka

集群成员关系

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

控制器

控制器其实就是一个 broker。集群里第一个启动的 broker 通过在 Zookeeper 里创建一个临时节点 /controller 让自己成为控制器。其他 broker 在控制器节点上创建 Zookeeper watch 对象，如果控制器被关闭或者与 Zookeeper 断开连接，它们会尝试让自己成为新的控制器。每个新选出的控制器通过 Zookeeper 的条件递增操作获得一个全新的、数值更大的 controller epoch。

控制器遍历分区，并确定谁应该成为新首领，随后，新首领开始处理生产者和消费者的请求，而跟随者开始从首领那里复制消息。

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

复制

复制功能是 Kafka 架构的核心。Kafka 把自己描述成一个分布式的、可分区的、可复制的提交日志服务。

Kafka 使用主题来组织数据，每个主题被分为若干个分区，每个分区有多个副本。每个 broker 可以保存成百上千个属于不同主题和分区的副本。

副本有两种类型：

• 首领副本：每个分区都有一个首领副本。为了保证一致性，所有生产者请求和消费者请求都会经过这个副本。
• 跟随者副本：首领以外的副本都是跟随者副本。跟随者副本不处理来自客户端的请求，它们唯一的任务就是从首领那么复制消息，保持与首领一致的状态。

为了与首领保持同步，跟随者向首领发送获取数据的请求，这种请求与消费者为了读取消息而发送的请求是一样的。请求消息里包含了跟随者想要获取消息的偏移量，而这些偏移量总是有序的。通过查看每个跟随者请求的最新偏移量，首领就会知道每个跟随者复制的进度。如果跟随者在10s内没有请求任何消息，或者虽然在请求消息，但是10s内没有请求最新的数据，那么它就被认为是不同步的。

处理请求

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

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

原来这才是Kafka

生产请求和获取请求都必须发送给分区的首领副本。Kafka 客户端负责把生产请求和获取请求发送到争取的 broker 上。客户端使用了另一种请求类型，也就是元数据请求。服务器端的响应消息里指明了这些主题所包含的分区、每个分区都有哪些副本，以及哪个副本是首领。一般情况下，客户端会把这些信息缓存起来。

原来这才是Kafka

Kafka 使用零复制技术向客户端发送消息 —— 也就是说，Kafka 直接把消息从文件（或者更确切的说是 Linux 文件系统缓存）里发送到网络通道，而不需要经过任何中间缓冲区。

并不是所有保存在分区首领上的数据都可以被客户端读取。大部分客户端只能读取已经被写入所有同步部分的消息。

原来这才是Kafka

物理存储

Kafka 的基本存储单元是分区。在配置 Kafka 的时候，管理员指定了一个用户存储分区的目录清单。

分区分配

在创建主题时，Kafka 会决定如何在 broker 间分配分区。为分区和副本选好 broker 之后，会决定哪些分区使用哪些目录。规则很简单：计算每个目录里的分区数量，新的分区总是被添加到数量最小的哪个目录里。

Kafka 管理员为每个主题配置了数据保留期限，规定数据被删除之前可以保留多长时间，或者清理数据之前可以保留数据量大小。

文件管理

因为一个大文件里查找和删除消息是很费时的，所以把分区分成若干个片段。在 broker 往分区写入数据时，如果达到片段上限，就关闭当前文件，并打开一个新文件。当前正在写入数据的片段叫做活跃片段。

文件格式

Kafka 把消息和偏移量保存在文件里。

原来这才是Kafka

索引

消费者可以从 Kafka 的任意可用偏移量位置开始读取消息。Kafka 为每个分区维护了一个索引。索引把偏移量映射到片段文件和偏移量在文件里的位置。

索引也被分成片段，所以再删除消息时，也可以删除相应的所以。如果索引出现损坏，Kafka 会通过重新读取消息并录制偏移量和位置来重新生成索引。

清理的工作原理

每个日志片段可以分为两个部分：

• 干净的部分：这些消息之前被清理过。
• 污浊的部分：这些消息是上一次清理之后写入的。

原来这才是Kafka

如果在 Kafka 启动时启动了清理功能，每个 broker 会启动一个清理管理器线程和多个清理线程，它们负责清理任务。为了清理分区，清理线程会读取分区的污浊部分，并在内存里创建一个 map。map 里的每个元素包含了消息键的散列值和消息的偏移量。

1.png

获取以上Java高级架构最新视频，欢迎

加入Java进阶架构交流群：142019080。直接点击链接加群。https://jq.qq.com/?_wv=1027&k=5lXBNZ7