一、zookeeper
Kafka 使用 Zookeeper 来维护集群成员的信息。每个 broker 都有一个唯一标识符, 这个标识符可以在配置文件里指定, 也可以自动生成。 在 broker 启动的时候, 它通过创建临时节点把自己的 ID 注册到 zookeeper。Kafka 组件订阅 Zookeeper 的/brokers/ids 路径(broker 在 zookeeper 上的注册路径) , 当有 broker 加入集群或退出集群时，这些组件就可以获得通知。
如果你要启动另一个具有相同 ID 的 broker, 会得到一个错误。
在 broker 停机、出现网络分区或长时间垃圾回收停顿时，broker 会从 Zookeeper 上断开连接, 此时 broker 在启动时创建的临时节点会自动从 Zookeeper 上移除。监听 broker 列表的 Kafka 组件会被告知该 broker 已移除。
在关闭 broker 时, 它对应的节点也会消失, 不过它的 ID 会继续存在于其他数据结构中。例如，主题的副本列表里就可能包含这些 ID。在完全关闭一 个 broker 之后, 如果使用相同的 ID 启动另一个全新的 broker，它会立即加入集群, 并拥有与旧 broker 相同的分区和主题。
二、控制器
控制器其实就是一个 broker，只不过它除了具有一般 broker 的功能之外，还负责分区首领的选举。集群里第一个启动的 broker 通过在 Zookeeper 里创建一个临时节点/controller 让自己成为控制器。其他 broker 在启动时也会尝试创建这个节点，不过它们会收到一个“节点已存在”的异常，说明集群里已经有一个控制器了。其他 broker 在控制器节点上创建 Zookeeperwatch 对象，这样它们就可以收到这个节点 的变更通知。这种方式可以确保集群里一次只有一个控制器存在。
如果控制器被关闭或者与 Zookeeper 断开连接，zookeeper 上的临时节点就会消失。集群里的其他 broker 通过 watch 对象得到控制器节点消失的通知，它们会尝试让自己成为新的控制器。第一个在 Zookeeper 里成功创建控制器节点的 broker 就会成为新的控制器，其他节点会收到“节点已存在”的异常，然后在新的控制器节点上再次创建 watch 对象。每个新选出的控制器通过Zookeeper 的条件递增操作获得一个全新的、数值更大的controller epoch，其他broker在知道当前controller epoch后，如果收到由控制器发出的包含较旧epoch的消息，就会忽略它们。
当控制器发现一个 broker 已经离开集群，它就知道，那些失去首领的分区需要一个新首领 (这些分区的首领刚好是在这个 broker 上)。控制器遍历这些分区，并确定谁应该成为新首领 (简单来说就是分区副本列表里的下一个副本) , 然后向所有包含新首领或现有跟随者的 broker 发送请求。该请求消息包含了谁是新首领以及谁是分区跟随者的信息。随后，新首领开始处理来自生产者和消费者的情求，而跟随者开始从新首领那里复制消息。
当控制器发现一个 broker 加入集群时，它会使用 broker ID 来检査新加入的 broker 是否包含现有分区的副本。如果有，控制器就把变更通知发送给新加入的 broker 和其他 broker，新 broker 上的副本开始从首领那里复制消息。
简而言之，Kafka 使用 Zookeeper 的临时节点来选举控制器，并在节点加入集群或退出集群时通知控制器。控制器负责在节点加入或离开集群时进行分 区首领选举。控制器使用epoch来避免脑裂。
三、复制
复制功能是 Kafka 架构的核心。Kafka 把自己描述成“一个分布式的、可分区的、可复制的提交日志服务”。复制之所以这么关键，是因为它可以在个别节点失效时仍能保证 Kafka 的可用性和持久性。
Kafka 使用主题来组织数据，每个主题被分为若干个分区，每个分区有多个副本。那些副本被保存在 broker 上, 每个 broker 可以保存成百上千个属于不同主题和分区的副本。
副本有以下两种类型。
- 首领副本：每个分区都有一个首领副本。为了保证一致性，所有生产者请求和消费者请求都会经过这个副本。
- 跟随副本：首领以外的副本都是跟随者副本。跟随者副本不处理来自客户端的请求，它们唯一的任务就是从首领那里复制消息，保持与首领一致的状态。如果首领发生崩溃，其中的一个跟随者会被提升为新首领。
首领的另一个任务是搞清楚哪个跟随者的状态与自己是一致的。跟随者为了保持与首领的状态一致，在有新消息到达时尝试从首领那里复制消息，不过有各种原因会导致同步失败。
为了与首领保持同步, 跟随者向首领发送获取数据的请求，这种请求与消费者为了读取消息而发送的请求是一样的。首领将响应消息发给跟随者。请求消息里包含了跟随者想要获取消息的偏移量，而且这些偏移量总是有序的 。
一个跟随者副本先请求消息1，接着请求消息2，然后请求消息3，在收到这3个请求的响应之前，它是不会发送第 4 个请求消息的。如果跟随者发送了请求消息 4，那么首领就知道它已经收到了前面 3 个请求的响应。通过査看每个跟随者请求的最新偏移量，首领就会知道每个跟随者复制的进度。如果跟随者在 10s 内没有请求任何消息，或者虽然在请求消息，但在 10s 内没有请求最新的数据，那么它就会被认为是不同步的。如果一个副本无法与首领保持一致，在首领发生失效时，它就不可能成为新首领，因为它没有包含全部的消息。相反，持续请求得到的最新消息副本被称为同步副本。在首领发生失效时，只有同步副本才有可能被选为新首领。
首选首领：创建主题时选定的首领就是分区的首选首领。之所以把它叫作首选首领，是因为在创建分区时，需要在 broker 之间均衡首领，避免让包含了首领的broker负载过重。分区的副本清单里的第一个副本一般就是首选首领，不管当前首领是哪一个副本，都不会改变这个事实。
四、处理请求
broker 的大部分工作是处理客户端、分区副本和控制器发送给分区首领的请求。客户端发起连接并发送请求，broker 处理请求并作出响应。broker 按照请求到达的顺序来处理它们一一这种顺序保证让 Kafka 具有了消息队列的特性，同时保证保存的消息也是有序的。
Kafka 处理请求的内部流程
broker 会在它所监听的每一个端口上运行一个 Acceptor 线程，这个线程会创建一个连接，并把它交给 Processor 线程去处理。 Processor 线程（也被叫作“网络线程”）的数量是可配置的。网络线程负责从客户端获取请求消息，把它们放进请求队列，然后从响应队列获取响应消息，把它们发送给客户端。请求消息被放到请求队列后，IO 线程会负责处理它们。
生产请求：生产者发送的请求，它包含客户端要写入 broker 的消息。
获取请求：在消费者和跟随者副本需要从 broker 读取消息时发送的请求。
生产请求和获取请求都必须发送给分区的首领副本。如果 broker 收到一个针对特定分区的请求，而该分区的首领在另一个 broker 上，那么发送请求的客户端会收到一个“非分区首领”的错误响应。当针对特定分区的获取请求被发送到一个不含有该分区首领的 broker 上，也会出现同样的错误。
那么客户端怎么知道该往哪里发送请求呢？客户端使用了另一种请求类型，也就是元数据请求。这种请求包含了客户端感兴趣的主题列表。服务器端的响应消息里指明了这些主题所包含的分区、每个分区都有哪些副本，以及哪个副本是首领。元数据请求可以发送给任意一个broker，因为所有broker都缓存了这些信息。一般情况下，客户端会把这些信息缓存起来，并直接往目标 broker 上发送生产请求和获取请求。它们需要时不时地通过发送元数据请求来刷新这些信息。
生产请求
acks 这个配置参数指定了需要多少个 broker 确认才可以认为一个消息写入是成功的。如果 acks=1，那么只要首领收到消息就认为写入成功；如果 acks=all，那么需要所有同步副本收到消息才算写入成功；如果 acks=0，那么生产者在把消息发出去之后，完全不需要等待 broker 的响应。
包含首领副本的 broker 在收到生产请求时，会对请求做一些验证。 .
- 发送数据的用户是否有主题写入权限？
- 请求里包含的 acks 值是否有效（只允许出现0、1 或 all) ?
- 如果 acks=all，是否有足够多的同步副本保证消息已经被安全写入？
之后，消息被写入本地磁盘。在消息被写入分区的首领之后，broker 开始检查 acks 配置参数，如果 acks 被设为 0 或 1, 那么 broker 立即返回响应；如果 acks 被设为 all，那么请求会被保存在一个叫作炼狱的缓冲区里，直到首领发现所有跟随者副本都复制了消息，晌应才会被返回给客户端。
获取请求
客户端发送请求，向 broker 请求主题分区里具有特定偏移量的消息，好像在说： “请把主题 Test 分区 0 偏移量从 53 开始 的消息以及主题 Test 分区 3 偏移量从 64 开始的消息发给我。”
客户端还可以指定 broker 最多可以从一个分区里返回多少数据。
请求需要先到达指定的分区首领上，然后客户端通过查询元数据来确保请求的路由是正确的。首领在收到请求时，它会先检查请求是否有效一一比如，指定的偏移量在分区上是否存在？如果客户端请求的是已经被删除的数据，或者请求的偏移量不存在，那么 broker 将返回一个错误。 如果请求的偏移量存在，broker 将按照客户端指定的数量上限从分区里读取消息，再把消息返回给客户端。 Kafka 使用零复制技术向客户端发送消息一一也就是说， Kafka 直接把消息从文件（或者更确切地说是 Linux 文件系统缓存）里发送到网络通道，而不需要经过任何中间缓冲区。这项技术避免了字节复制，也不需要管理内存缓冲区，从而获得更好的性能。
客户端除了可以设置 broker 返回数据的上限，也可以设置下限。客户端发送一个请求，broker 等到有足够的数据时才把它们返回给客户端，然后客户端再发出请求，而不是让客户端每隔几毫秒就发送一次请求，每次只能得到很少的数据甚至没有数据。
并不是所有保存在分区首领上的数据都可以被客户端读取。大部分客户端只能读取已经被写入所有同步副本的消息（跟随者副本也不行，尽管它们也是消费者，否则复制功能就无法工作）。分区首领知道每个消息会被复制到哪个副本上，在消息还没有被写入所有同步副本之前，是不会发送给消费者的一一尝试获取这些消息的请求会得到空的响应而不是错误。
因为还没有被足够多副本复制的消息被认为是“不安全”的一一如果首领发生崩溃，另一个副本成为新首领，那么这些消息就丢失了。如果我们允许消费者读取这些消息，可能就会破坏一致性。 这也意味着，如果 broker 间的消息复制因为某些原因变慢，那么消息到达消费者的时间也会随之变长（因为我们会先等待消息复制完毕）。延迟时间可以通过参数 replica.lag.time.max.ms 来配置，它指定了副本在复制消息时可被允许的最大延迟时间。
其他请求
到此为止，我们讨论了 Kafka 最为常见的几种请求类型：元数据请求、生产请求和获取请求。broker 之间也使用同样的通信协议。它们之间的请求发生在 Kafka 内部，客户端不应该使用这些请求。例如，当一个新首领被选举出来，控制器会发送 LeaderAndIsr 请求给新首领（这样它就可以开始接收来自客户端的请求）和跟随者（这样它们就知道要开始跟随新首领）。
之前的 Kafka 消费者使用 Zookeeper 来跟踪偏移量，在消费者启动的时候，它通过检查保存在 Zookeeper 上的偏移量就可以知道从哪里开始处理消息。因为各种原因，我们决定不再使用 Zookeeper 来保存偏移量，而是把偏移量保存在特定的 Kafka 主题上。
五、物理存储
Kafka 的基本存储单元是分区。分区无法在多个 broker 间进行再细分，也无法在同一个 broker 的多个磁盘上进行再细分。
在配置 Kafka 的时候，管理员指定了一个用于存储分区的目录清单，log.dirs 参数的值。
分区分配
在创建主题时，Kafka 首先会决定如何在 broker 间分配分区。假设你有 6 个 broker，打算创建一个包含 10 个分区的主题，并且复制系数为 3。那么 Kafka 就会有 30 个分区副本，它们可以被分配给 6 个 broker。在进行分区分配时，我们要达到如下的目标。
- 在 broker 间平均地分布分区副本。对于我们的例子来说，就是要保证每个 broker 可以分到5个副本。
- 确保每个分区的每个副本分布在不同的 broker 上。
- 如果为 broker 指定了机架信息，那么尽可能把每个分区的副本分配到不同机架的 broker 上。
为分区和副本选好合适的 broker 之后，接下来要决定这些分区应该使用哪个目录。我们单独为每个分区分配目录，规则很简单： 计算每个目录里的分区数量，新的分区总是被添加到数量最小的那个目录里。
注意磁盘空间：在为 broker 分配分区时并没有考虑可用空间和工作负载问题，但在将分区分配到磁盘上时会考虑分区数量，不考虑分区大小。 也就是说， 如果有些 broker 的磁盘空间比其他 broker 要大，有些分区异常大，或者同一个 broker 上有大小不同的磁盘，那么在分配分区时要格外小心。
文件管理
保留数据是 Kafka 的一个基本特性，Kafka 不会一直保留数据，也不会等到所有消费者都读取了消息之后才删除消息。相反，Kafka 管理员为每个主题配置了数据保留期限，规定数据被删除之前可以保留多长时间，或者清理数据之前可以保留的数据量大小。
因为在一个大文件里查找和删除消息是很费时的，所以我们把分区分成若干个片段。 默认情况下，每个片段包含 1GB 或一周的数据，如果达到片段上限，就关闭当前文件，并打开一个新文件。当前正在写入数据的片段叫作活跃片段。活动片段永远不会被删除，所以如果你要保留数据 1 天，但片段里包含了 5 天的数据，那么这些数据会被保留 5 天，因为在片段被关闭之前这些数据无法被删除。
文件格式
我们把 Kafka 的消息和偏移量保存在文件里。保存在磁盘上的数据格式与从生产者发送过来或者发送给消费者的消息格式是一样的。因为使用了相同的消息格式进行磁盘存储和网络传输，Kafka 可以使用零复制技术给消费者发送消息，同时避免了对生产者已经压缩过的消息进行解压和再压缩。
如果生产者发送的是压缩过的消息，那么同一个批次的消息会被压缩在一起，被当作“包装消息”进行发送。broker 就会收到一个这样的消息，然后再把它发送给消费者。消费者在解压这个消息之后，会看到整个批次的消息，它们都有自己的时间戳和偏移量。如果在生产者端使用了压缩功能，那么发送的批次越大，就意味着在网络传输和磁盘存储方面会获得越好的压缩性能。
索引
消费者可以从 Kafka 的任意可用偏移量位置开始读取消息。假设消费者要读取从偏移量 100 开始的 1MB 消息，那么 broker 必须立即定位到偏移量 100 （可能是在分区的任意一个片段里），然后开始从这个位置读取消息。为了帮助 broker 更快地定位到指定的偏移量，Kafka 为每个分区维护了一个索引。索引把偏移量映射到片段文件和偏移量在文件里的位置。 索引也被分成片段，所以在删除消息时，也可以删除相应的索引。Kafka 不维护索引的校验和。如果索引出现损坏， Kafka 会通过重新读取消息并录制偏移量和位置来重新生成索引。如果有必要，管理员可以删除索引，这样做是绝对安全的，Kafka 会自动重新生成这些索引。
清理
一般情况下， Kafka 会根据设置的时间保留数据，把超过时效的旧数据删除掉。Kafka 通过改变主题的保留策略来满足这些使用场景。早于保留时间的旧事件会被删除，为每个键保留最新的值，从而达到清理的效果。只有当应用程序生成的事件里包含了键值对时，为这些主题设置 compact 策略才有意义。如果主题包含 null 键，清理就会失败。
清理的工作原理
每个日志片段可以分为以下两个部分。
干净的部分 : 这些消息之前被清理过，每个键只有一个对应的值，这个值是上一次清理时保留下来的。
污浊的部分 : 这些消息是在上一次清理之后写入的。
如果在 Kafka 启动时启用了清理功能（log.cleaner.enabled），每个 broker 会启动一个清理管理器线程和多个清理线程，它们负责执行清理任务。这些线程会选择污浊率（污浊消息占分区总大小的比例）较高的分区进行清理。
为了清理分区，清理线程会读取分区的污浊部分，并在内存里创建一个 map。 map 里的每个元素包含了消息键的散列值和消息的偏移量。
清理线程在创建好偏移量 map 后，开始从干净的片段处读取消息，从最旧的消息开始，把它们的内容与 map 里的内容进行比对。它会检查消息的键是否存在于 map 中，如果不存在，那么说明消息的值是最新的，就把消息复制到替换片段上。 如果键已存在，消息会被忽略，因为在分区的后部已经有一个具有相同键的消息存在。在复制完所有的消息之后，我们就将替换片段与原始片段进行交换，然后开始清理下一个片段。完成整个清理过程之后，每个键对应一个不同的消息——这些消息的值都是最新的。
被删除的事件
为了彻底把一个键从系统里删除，应用程序必须发送一个包含该键且值为 null 的消息。清理线程发现该悄息时，会先进行常规的清理，只保留值为 null 的消息。
何时会清理主题
在 0.10.0 和更早的版本里， Kafka 会在包含脏记录的主题数量达到 50% 时进行清理。这样做的目的是避免太过频繁的清理（因为清理会影响主题的读写性能），同时也避免存在太多脏记录（因为它们会占用磁盘空间）。
五、日志存储
Kafka 中的消息是以主题为基本单位进行归类的，各个主题在逻辑上相互独立。每个主题又可以分为一个或多个分区，分区的数量可以在主题创建的时候指定，也可以在之后修改。每条消息在发送的时候会根据分区规则被追加到指定的分区中，分区中的每条消息都会被分配一个唯一的序列号，也就是通常所说的偏移量。
不考虑多副本的情况，一个分区对应一个日志（Log）。为了防止 Log 过大，Kafka又引入了日志分段（LogSegment）的概念，将Log切分为多个LogSegment，相当于一个巨型文件被平均分配为多个相对较小的文件，这样也便于消息的维护和清理。Log 在物理上只以文件夹的形式存储，而每个LogSegment 对应于磁盘上的一个日志文件和两个索引文件。
向Log 中追加消息时是顺序写入的，只有最后一个 LogSegment 才能执行写入操作，在此之前所有的 LogSegment 都不能写入数据。为了方便描述，我们将最后一个 LogSegment 称为“activeSegment”，即表示当前活跃的日志分段。随着消息的不断写入，当activeSegment满足一定的条件时，就需要创建新的activeSegment，之后追加的消息将写入新的activeSegment。为了便于消息的检索，每个LogSegment中的日志文件（以“.log”为文件后缀）都有对应的两个索引文件：偏移量索引文件（以“.index”为文件后缀）和时间戳索引文件（以“.timeindex”为文件后缀）。
