前言

Github：https://github.com/HealerJean

博客：http://blog.healerjean.com

1、集群成员关系

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

Kafka 组件订阅 Zookeeper 的 /brokers/ids 路径(broker 在 Zookeeper 上的注册路径)，当有 broker 加入集群或退出集群时，这些组件就 可以获得通知。

zookeeper.connect=localhost:2181/brokers/ids

1.1、关闭broker

在关闭 broker 时，它对应的节点也会消失，不过它的 ID 会继续存在于其他数据结构中。

例如，主题的副本列表里就可能包含这些 ID。在完全关闭一个 broker 之 后，如果使用相同的 ID 启动另一个全新的 broker，它会立即加入集群，并拥有与旧 broker 相同的分区和主题。

1.2、broker 停机

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

1.3、启动相同ID 的 broker

如果你要启动另一个具有相同ID 的 broker，会得到一个错误——新 broker 会试着进行注 册，但不会成功，

2、控制器

控制器其实就是一个 broker，只不过它除了具有一般 broker 的功能之外，还负责 分区首领 的选举，集群里第一个启动的 broker 通过在 Zookeeper 里创建一个临时节点 /controller 让自己成为控制器。

其他 broker 在启动时也 会尝试创建这个节点，不过它们会收到一个“节点已存在”的异常，然后“意识”到控制 器节点已存在，也就是说集群里已经有一个控制器了。其他 broker 在控制器节点上创建 Zookeeper watch 对象，这样它们就可以收到这个节点的变更通知。这种方式可以确保集群 里一次只有一个控制器存在。

2.1、控制器断开连接后重新选举

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

第一个在 Zookeeper 里成功创建控制器节点的 broker 就会成为新的控制器，其他 节点会收到“节点已存在”的异常，然后在新的控制器节点上再次创建 watch 对象。

每个新选出的控制器通过 Zookeeper 的条件递增操作获得一个全新的、数值更大的 controller epoch。其他broker在知道当前controller epoch后，如果收到由控制器发出的包含较旧 epoch 的消息，就会忽略它们。

2.2、broder分区首领离开集群后控制器选举分区首领

当控制器发现一个 broker 已经离开集群(通过观察相关的 Zookeeper 路径)，它就知道，那些失去首领的分区需要一个新首领(这些分区的首领刚好是在这个 broker 上)。

控制器遍 历这些分区，并确定谁应该成为新首领(简单来说就是分区副本列表里的下一个副本)， 然后向所有包含新首领或现有跟随者的 broker 发送请求。该请求消息包含了谁是新首领以及谁是分区跟随者的信息。 随后，新首领开始处理来自生产者和消费者的请求，而跟随者 开始从新首领那里复制消息。

2.3、broker加入集群

当控制器发现一个 broker 加入集群时，它会使用 broker ID 来检查新加入的 broker 是否包含现有分区的副本。如果有，控制器就把变更通知发送给新加入的 broker 和其他 broker， 新 broker 上的副本开始从首领那里复制消息。

2.4、总结

⬤ Kafka 使用 Zookeeper 的临时节点来选举控制器，并在节点加入集群或退出集群时通知控制器。

⬤ 控制器使用 epoch 来避免“脑裂”。“脑裂”是指两个节点同时认为自己是当前的控制器。

⬤ 控制器负责在节点加入或离开集群时进行分区首领选举。

3、复制

复制功能是 Kafka 架构的核心。在 Kafka 的文档里，Kafka 把自己描述成“一个分布式的、 可分区的、可复制的提交日志服务”。复制之所以这么关键，是因为它可以在个别节点失 效时仍能保证 Kafka 的可用性和持久性。

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

3.1、副本的两种类型

副本有以下两种类型。

首领副本：每个分区都有一个首领副本。为了保证一致性，所有生产者请求和消费者请求都会经过这个副本。

跟随者副本：首领以外的副本都是跟随者副本。跟随者副本不处理来自客户端的请求，它们唯一的任务就是从首领那里复制消息，保持与首领一致的状态。如果首领发生崩溃，其中的一个跟随者会被提升为新首领。

3.2、跟随者复制过程

首领的另一个任务是搞清楚哪个跟随者的状态与自己是一致的。跟随者为了保持与首领的 状态一致，在有新消息到达时尝试从首领那里复制消息，不过有各种原因会导致同步失 败。例如，网络拥塞导致复制变慢，broker 发生崩溃导致复制滞后，直到重启 broker 后复 制才会继续。

为了与首领保持同步，跟随者向首领发送获取数据的请求，这种请求与消费者为了读取消息而发送的请求是一样的。首领将响应消息发给跟随者。请求消息里包含了跟随者想要获 取消息的偏移量，而且这些偏移量总是有序的。

一个跟随者副本先请求消息 1，接着请求消息 2，然后请求消息 3，在收到这 3 个请求的响 应之前，它是不会发送第 4 个请求消息的。如果跟随者发送了请求消息 4，那么首领就知 道它已经收到了前面 3 个请求的响应。通过查看每个跟随者请求的最新偏移量，首领就会 知道每个跟随者复制的进度。

如果跟随者在 10s 内没有请求任何消息，或者虽然在请求消 息，但在 10s 内没有请求最新的数据，那么它就会被认为是不同步的。 如果一个副本无法 与首领保持一致，在首领发生失效时，它就不可能成为新首领——毕竟它没有包含全部的消息。相反，持续请求得到的最新消息副本被称为同步的副本。在首领发生失效时，只有同步副本才有可能被选为新首领。

跟随者的正常不活跃时间或在成为不同步副本之前的时间是通过 replica.lag.time.max.ms 参数来配置的。这个时间间隔直接影响着首领选举期间的客户端行为和数据保留机制。

3.3、首选首领

除了当前首领之外，每个分区都有一个首选首领——创建主题时选定的首领就是分区的首选首领。

之所以把它叫作首选首领，是因为在创建分区时，需要在 broker 之间均衡首领 (我的理解：将首领分散到各个broker)。 因此，我们希望首选首领在成为真正的首领时，broker 间的负载最终会得到均衡。

默认情况下，Kafka 的 auto.leader.rebalance. enable 被设为 true，它会检查首选首领是不是当前首领，如果不是，并且该副本是同步的，那么就会触发首领选举，让首选首领成为当前首领。

3.3.1、找到首选首领

从分区的副本清单里可以很容易找到首选首领(可以使用 kafka.topics.sh 工 具查看副本和分区的详细信息，。清单里的第一个副本一般就是首选首领。

不管当前首领是哪一个副本，都不会改变这 个事实，即使使用副本分配工具将副本重新分配给其他 broker。要记住，如 果你手动进行副本分配，第一个指定的副本就是首选首领，所以要确保首选首领被传播到其他 broker 上(我的理解：告诉其他broker，我已经是首领了)，避免让包含了首领的 broker 负载过重，而其他 broker 却无法为它们分担负载。

4、处理请求

4.1、Broker处理请求过程

1、broker 会在它所监听的每一个端口上运行一个Acceptor 线程，这个线程会创建一个连接， 并把它交给 Processor 线程去处理。

2、Processor 线程(也被叫作“网络线程”)的数量是可 配置的。网络线程负责从客户端获取请求消息，把它们放进请求队列，然后从响应队列获 取响应消息，把它们发送给客户端。

3、请求消息被放到请求队列后，IO 线程会负责处理它们。下面是几种最常见的请求类型。

①Acceptor线程：负责监听端口并创建连接，并将连接交给Processor线程处理

②Processor线程：负责从客户端获取请求消息，并放入请求队列，然后从响应队列获取响应消息，把它们发送给客户端。线程数量可以配置

③IO线程：负责处理请求，并产生响应

Kafka broker处理请求的过程如图

4.1.1、生产请求和获取请求都必须发送给分区的首领副本。

⬤ 如果 broker 收到一个针对特定分区的 请求，而该分区的首领在另一个 broker 上，那么发送请求的客户端会收到一个“非分区 首领”的错误响应。

⬤ 当针对特定分区的获取请求被发送到一个不含有该分区首领的 broker 上，也会出现同样的错误。Kafka 客户端要自己负责把生产请求和获取请求发送到正确的 broker 上。

4.1.2、客户端怎么知道该往哪里发送请求

那么客户端怎么知道该往哪里发送请求呢?客户端使用了另一种请求类型，也就是元数据 请求。这种请求包含了客户端感兴趣的主题列表。

服务器端的响应消息里指明了这些主题 所包含的分区、每个分区都有哪些副本，以及哪个副本是首领。元数据请求可以发送给任意一个 broker，因为所有 broker 都缓存了这些信息。

一般情况下，客户端会把这些信息缓存起来，并直接往目标 broker 上发送生产请求和 获取请求。它们需要时不时地通过发送元数据请求来刷新这些信息(刷新的时间间隔通 过 metadata.max.age.ms 参数来配置)，从而知道元数据是否发生了变更 ，——比如，在新 broker 加入集群时，部分副本会被移动到新的 broker 上。

另外，如果客户端收到“非首领”错误，它会在尝试重发请求之前先刷新元数据，因为这个错误说明了客户端正在使用过期的元数据信息，之前的请求被发到了错误的 broker 上。

4.1.3、Kafka高效文件存储

写message

1、消息从 java 堆转入page cache(即物理内存)。

2、由异步线程刷盘,消息从page cache刷入磁盘。

读message

1、消息直接从page cache转入socket发送出去。

2、当从page cache没有找到相应数据时，此时会产生磁盘 IO,从磁盘Load消息到page cache,然后直接从socket发出去

Kafka高效文件存储设计特点

1、Kafka 把 topic中一个parition大文件分成多个小文件段，通过多个小文件段，就容易定期清除或删除已经消费完文件，减少磁盘占用。

2、通过索引文件稀疏存储，可以大幅降低index文件元数据占用空间大小。

3、通过索引信息可以快速定位 message和确定response的最大大小。

4、通过 index 元数据全部映射到 memory缓存，可以避免segment file的IO磁盘操作。

4.2、生产请求

生产请求：生产者发送的请求，它包含客户端要写入 broker 的消息。

我们在第 3 章讨论如何配置生产者的时候，提到过 acks 这个配置参数——该参数指定了需要多少个 broker 确认才可以认为一个消息写入是成功的。不同的配置对“写入成功”的界 定是不一样的，

1、如果 acks=1，那么只要首领收到消息就认为写入成功;

2、如果 acks=all，那 么需要所有同步副本收到消息才算写入成功;

3、如果 acks=0，那么生产者在把消息发出去之 后，完全不需要等待 broker 的响应。

包含首领副本的 broker 在收到生产请求时，会对请求做一些验证。

1、发送数据的用户是否有主题写入权限?

2、请求里包含的 acks 值是否有效(只允许出现 0、1 或 all)?

3、如果 acks=all，是否有足够多的同步副本保证消息已经被安全写入?(我们可以对broker 进行配置，如果同步副本的数量不足，broker 可以拒绝处理新消息。在第 6 章介 绍 Kafka 持久性和可靠性保证时，我们会讨论更多这方面的细节。)之后，消息被写入本地磁盘。在 Linux 系统上，消息会被写到文件系统缓存里，并不保证 它们何时会被刷新到磁盘上。Kafka 不会一直等待数据被写到磁盘上——它依赖复制功能 来保证消息的持久性。

在消息被写入分区的首领之后，broker 开始检查 acks 配置参数——如果 acks 被设为 0 或 1， 那么 broker 立即返回响应;如果 acks 被设为 all，那么请求会被保存在一个叫作炼狱的缓冲区里，直到首领发现所有跟随者副本都复制了消息，响应才会被返回给客户端。

4.3、获取请求

获取请求：在消费者和跟随者副本需要从 broker 读取消息时发送的请求。broker 处理获取请求的方式与处理生产请求的方式很相似。客户端发送请求，

1、向 broker 求主题分区里具有特定偏移量的消息，好像在说:“请把主题 Test 分区 0 偏移量从 53 开始 的消息以及主题 Test 分区 3 偏移量从 64 开始的消息发给我。

2、”客户端还可以指定 broker 最多可以从一个分区里返回多少数据。这个限制是非常重要的，因为客户端需要为 broker 返 回的数据分配足够的内存。如果没有这个限制，broker 返回的大量数据有可能耗尽客户端 的内存。

4.3.1、请求偏移量

我们之前讨论过，请求需要先到达指定的分区首领上，然后客户端通过查询元数据来确保请求的路由是正确的。首领在收到请求时，它会先检查请求是否有效——比如，指定的偏移量在分区上是否存在?如果客户端请求的是已经被删除的数据，或者请求的偏移量不存 在，那么 broker 将返回一个错误。

如果请求的偏移量存在，broker将按照客户端指定的数量上限从分区里读取消息，再把消息返回给客户端。Kafka 使用零复制技术向客户端发送消息——也就是说，Kafka 直接把消息从文件(或者更确切地说是 Linux 文件系统缓存)里发送到网络通道，而不需要经过任 何中间缓冲区。这是 Kafka 与其他大部分数据库系统不一样的地方，其他数据库在将数据 发送给客户端之前会先把它们保存在本地缓存里。这项技术避免了字节复制，也不需要管理内存缓冲区，从而获得更好的性能。

客户端除了可以设置 broker 返回数据的上限，也可以设置下限。例如，如果把下限设置为 10KB，就好像是在告诉 broker:“等到有 10KB 数据的时候再把它们发送给我。”在主题消 息流量不是很大的情况下，这样可以减少 CPU 和网络开销。客户端发送一个请求，broker 等到有足够的数据时才把它们返回给客户端，然后客户端再发出请求，而不是让客户端每 隔几毫秒就发送一次请求，每次只能得到很少的数据甚至没有数据。(如图 5-3 所示。)对 比这两种情况，它们最终读取的数据总量是一样的，但前者的来回传送次数更少，因此开 销也更小。

4.1.2、消费者读取数据需要等待所有broker跟随者复制完成

我们不会让客户端一直等待 broker 累积数据。在等待了一段时间之后，就可以把可用的数据拿回处理，而不是一直等待下去。所以，客户端可以定义一个超时时间，告 诉 broker:“如果你无法在 X 毫秒内累积满足要求的数据量，那么就把当前这些数据返回 给我。

大部分客户端只能读取已经被写入所有同步副本的消息。分区首领知道每个消息会被复制到哪个副本上，在消息还没有 被写入所有同步副本之前，是不会发送给消费者的——尝试获取这些消息的请求会得到空的响应而不是错误。

为什么？

如果首领发生崩溃，另一 个副本成为新首领，那么这些消息就丢失了。 此时，如果我们允许消费者读取这些消息，可能就 会破坏一致性。

试想，一个消费者读取并处理了这样的一个消息，而另一个消费者发现这 个消息其实并不存在。所以，我们会等到所有同步副本复制了这些消息，才允许消费者读 取它们(如图 5-4 所示)。

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

4.4、其他请求

到此为止，我们讨论了 Kafka 最为常见的几种请求类型:元数据请求、生产请求和获取请求。

重要的是，我们讨论的是客户端在网络上使用的通用二进制协议。Kafka 内置了由 开源社区贡献者实现和维护的 Java 客户端，同时也有用其他语言实现的客户端，如 C、 Python、Go 语言等。Kafka 网站上有它们的完整清单，这些客户端就是使用这个二进制协 议与 broker 通信的。

broker 之间也使用同样的通信协议。它们之间的请求发生在Kafka 内部，客户端不 应该使用这些请求。例如，当一个新首领被选举出来，控制器会发送 LeaderAndIsr 请求给 新首领(这样它就可以开始接收来自客户端的请求)和跟随者(这样它们就知道要开始跟 随新首领)。

4.4.1、增加新的请求类型

Kafka 协议可以处理 20 种不同类型的请求，而且会有更多的类 型加入进来。协议在持续演化——随着客户端功能的不断增加，我们需要改进协议来满足 需求。

例如，之前的 Kafka 消费者使用 Zookeeper 来跟踪偏移量，在消费者启动的时候， 它通过检查保存在 Zookeeper 上的偏移量就可以知道从哪里开始处理消息。因为各种原 因，我们决定不再使用 Zookeeper 来保存偏移量，而是把偏移量保存在特定的 Kafka 主题 上。为了达到这个目的，我们不得不往协议里增加几种请求类型:OffsetCommitRequest、 OffsetFetchRequest 和 ListOffsetsRequest。现在，在应用程序调用 commitOffset() 方法 时，客户端不再把偏移量写入 Zookeeper，而是往 Kafka 发送 OffsetCommitRequest 请求。

主题的创建仍然需要通过命令行工具来完成，命令行工具会直接更新 Zookeeper 里的主题 列表，broker 监听这些主题列表，在有新主题加入时，它们会收到通知。我们正在改进 Kafka，增加了 CreateTopicRequest 请求类型，这样客户端(包括那些不支持 Zookeeper 客 户端的编程语言)就可以直接向 broker 请求创建新主题了。

4.4.2、修改请求类型

除了往协议里增加新的请求类型外，我们也会通过修改已有的请求类型来给它们增加新功 能。例如，从 Kafka 0.9.0 到 Kafka 0.10.0，我们希望能够让客户端知道谁是当前的控制器， 于是把控制器信息添加到元数据响应消息里。我们还在元数据请求消息和响应消息里添加了 一个新的 version 字段。

现在，0.9.0 版本的客户端发送的元数据请求里 version 为 0(0.9.0 版 本客户端的 version 不会是 1)。不管是 0.9.0 版本的 broker，还是 0.10.0 版本的 broker，它们 都知道应该返回 version 为 0 的响应，也就是不包含控制器信息的响应。

0.9.0 版本的客户端不需要控制器的信息，而且也没必要知道如何去解析它。0.10.0 版本的客户端会发送 version 为 1 的元数据请求，0.10.0 版本的 broker 会返回 version 为 1 的响应，里面包含了控制器的 信息。

如果 0.10.0 版本的客户端发送 version 为 1 的请求给 0.9.0 版本的 broker，这个版本的 broker 不知道该如何处理这个请求，就会返回一个错误。这就是为什么我们建议在升级客户 端之前先升级 broker，因为新的 broker 知道如何处理旧的请求，反过来则不然。

我们在 0.10.0 版本的 Kafka 里加入了 ApiVersionRequest——客户端可以询问 broker 支持哪 些版本的请求，然后使用正确的版本与 broker 通信。如果能够正确使用这个新功能，客户 端就可以与旧版本的 broker 通信，只要 broker 支持这个版本的协议。

5、物理存储

1、Kafka 的基本存储单元是分区。分区无法在多个 broker 间进行再细分，也无法在同一个 broker 的多个磁盘上进行再细分。所以，分区的大小受到单个挂载点可用空间的限制

2、在配置 Kafka 的时候，管理员指定了一个用于存储分区的目录清单——也就是 log.dirs 参 数的值，该参数一般会包含每个挂载点的目录。

3、接下来我们会介绍 Kafka 是如何使用这些目录来存储数据的。

5.1、分区分配

在创建主题时，Kafka 首先会决定如何在 broker 间分配分区。假设你有 6 个 broker，打算 创建一个包含 10 个分区的主题，并且复制系数为 3。那么 Kafka 就会有 30 个分区副本， 它们可以被分配给 6 个 broker。在进行分区分配时，我们要达到如下的目标。

◯ 在broker 间平均地分布分区副本。对于我们的例子来说，就是要保证每个 ` broker` 可以 分到 5 个副本。

◯ 确保每个分区的每个副本分布在不同的 broker 上。假设分区 0 的首领副本在 broker 2 上， 那么可以把跟随者副本放在 broker 3 和 broker 4 上，但不能放在 broker 2 上，也不能两个都放在 broker 3 上。

◯ 如果为 broker 指定了机架信息，那么尽可能把每个分区的副本分配到不同机架的 broker 上。这样做是为了保证一个机架的不可用不会导致整体的分区不可用。

5.1.1、按照数字顺序分配

1、为了实现这个目标，我们先随机选择一个 broker(假设是 4)，然后使用轮询的方式给每 个 broker 分配分区来确定首领分区的位置。于是，首领分区 0 会在 broker 4 上，首领分区 1 会在 broker 5 上，首领分区 2 会在 broker 0 上(只有 6 个 broker)，并以此类推。然后，

2、我们从分区首领开始，依次分配跟随者副本。如果分区 0 的首领在 broker 4 上，那么它的 第一个跟随者副本会在 broker 5 上，第二个跟随者副本会在 broker 0 上。分区 1 的首领在 broker 5 上，那么它的第一个跟随者副本在 broker 0 上，第二个跟随者副本在 broker 1 上。

5.1.2、配置了机架信息分配

如果配置了机架信息，那么就不是按照数字顺序来选择 broker 了，而是按照交替机架的方式 来选择 broker。假设 broker 0、broker 1 和 broker 2 放置在同一个机架上，broker 3、broker 4 和 broker 5 分别放置在其他不同的机架上。

我们不是按照从 0 到 5 的顺序来选择 broker，而 是按照 0，3，1，4，2，5 的顺序来选择，这样每个相邻的 broker 都在不同的机架上(如图 5-5 所示)。

于是，如果分区 0 的首领在 broker 4 上，那么第一个跟随者副本会在 broker 2 上， 这两个 broker 在不同的机架上。如果第一个机架下线，还有其他副本仍然活跃着，所以分区 仍然可用。这对所有副本来说都是一样的，因此在机架下线时仍然能够保证可用性。

5.1.3、为分区分配目录

为分区和副本选好合适的 broker 之后，接下来要决定这些分区应该使用哪个目录。

我们单独为每个分区分配目录，规则很简单:计算每个目录里的分区数量，新的分区总是被添加到数量最小的那个目录里。也就是说，如果添加了一个新磁盘，所有新的分区都会被创建 到这个磁盘上。因为在完成分配工作之前，新磁盘的分区数量总是最少的。

要注意，在为 broker 分配分区时并没有考虑可用空间和工作负载问题，但在将分区分配到磁盘上时会考虑分区数量，不过不考虑分区大小。也就是说， 如果有些 broker 的磁盘空间比其他 broker 要大(有可能是因为集群同时使 用了旧服务器和新服务器)，有些分区异常大，或者同一个 broker上有大小 不同的磁盘，那么在分配分区时要格外小心。在后面的章节中，我们会讨论 Kafka 管理员该如何解决这种 broker 负载不均衡的问题。

5.2、文件管理

保留数据是 Kafka 的一个基本特性，Kafka 不会一直保留数据，也不会等到所有消费者都 读取了消息之后才删除消息。相反，Kafka 管理员为每个主题配置了数据保留期限，规定 数据被删除之前可以保留多长时间，或者清理数据之前可以保留的数据量大小。

因为在一个大文件里查找和删除消息是很费时的，也很容易出错，所以我们把分区分成若 干个片段。默认情况下，每个片段包含 1GB 或一周的数据，以较小的那个为准。

在 broker 往分区写入数据时，如果达到片段上限，就关闭当前文件，并打开一个新文件。当前正在写入数据的片段叫作活跃片段。活动片段永远不会被删除，

​

⬤ 所以如果你要保留数据 1 天，但片段里包含了 5 天的数据，因为在片段被关闭之 前这些数据无法被删除，那么这些数据会被保留 5 天。

⬤ 如果你要保留数据一周，而且每天使用一个新片段，那么你就会 看到，每天在使用一个新片段的同时会删除一个最老的片段——所以大部分时间该分区会 有 7 个片段存在。

我们在第 2 章讲过，broker 会为分区里的每个片段打开一个文件句柄，哪怕片段是不活跃 的。这样会导致打开过多的文件句柄，所以操作系统必须根据实际情况做一些调优。

5.3、文件格式

我们把 Kafka 的消息和偏移量保存在文件里。保存在磁盘上的数据格式与从生产者发送过 来或者发送给消费者的消息格式是一样的。因为使用了相同的消息格式进行磁盘存储和网 络传输， Kafka 可以使用零复制技术给消费者发送消息，同时避免了对生产者已经压缩过 的消息进行解压和再压缩。

除了键、值和偏移量外，消息里还包含了消息大小、校验和、消息格式版本号、压缩算法 (Snappy、GZip 或 LZ4)和时间戳(在 0.10.0 版本里引入的)。时间戳可以是生产者发送消息的时间，也可以是消息到达 broker 的时间，这个是可配置的。

如果生产者发送的是压缩过的消息，那么同一个批次的消息会被压缩在一起，被当作“包装消息”进行发送(如图 5-6 所示)。

于是，broker 就会收到一个这样的消息，然后再把它发送给消费者。消费者在解压这个消息之后，会看到整个批次的消息，它们都有自己的时 间戳和偏移量。

也就是说，如果在生产者端使用了压缩功能(极力推荐)，那么发送的批次越大，就意味 着在网络传输和磁盘存储方面会获得越好的压缩性能，同时意味着如果修改了消费者使用的消息格式(例如，在消息里增加了时间戳)，那么网络传输和磁盘存储的格式也要随之修改，而且 broker 要知道如何处理包含了两种消息格式的文件。

Kafka 附带了一个叫 DumpLogSegment 的工具，可以用它查看片段的内容。它可以显示每个消息的偏移量、校验和、魔术数字节、消息大小和压缩算法。运行该工具的方法如下:bin/kafka-run-class.sh kafka.tools.DumpLogSegments如果使用了 --deep-iteration 参数，可以显示被压缩到包装消息里的消息。

5.4、索引

消费者可以从 Kafka 的任意可用偏移量位置开始读取消息。主要依赖于 kafka消息的存储：

分片 segment（日志片段) + 索引(index) 的机制

5.4.1、kafka消息存储

partition分区下面会分为多个 segment片段（分片），每个分片对应着.log和.index文件。log文件超过1G的时候会新创建一个log文件。 segment file组成：由2大部分组成，分别为index file和data file，此2个文件一一对应，成对出现，后缀”.index”和“.log”分别表示为segment索引文件、数据文件.

segment 文件命名规则：partion 全局的第一个 segment 从0开始，后续每个 segment 文件名为上一个segment文件最后一条消息的offset值。数值最大为64位long大小，19 位数字字符长度，没有数字用0填充。

5.4.2、在partition中如何通过offset查找message

例如读取offset=368776的message，需要通过下面2个步骤查找。

1、第一步查找segment file 上述图2为例，其中00000000000000000000.index表示最开始的文件，起始偏移量(offset)为0.第二个文件00000000000000368769.index的消息量起始偏移量为368770 = 368769 + 1.同样，第三个文件00000000000000737337.index的起始偏移量为737338=737337 + 1，其他后续文件依次类推，以起始偏移量命名并排序这些文件，

只要根据offset 二分查找文件列表，就可以快速定位到具体文件。 当offset=368776时定位到00000000000000368769.index/log

2、第二步通过segment file查找message 通过第一步定位到segment file，当offset=368776时，依次定位到00000000000000368769.index的元数据物理位置和00000000000000368769.log的物理偏移地址，然后再通过00000000000000368769.log顺序查找直到offset=368776为止。

368776 - 368769 = 7，以此再定位到 6,1407 对应元数据物理位置，然后找到1407物理位置00000000000000368769.index所在的 Message368775 1407 行，接着顺序查找即可查找到Message368776 1508对应的数据

5.2.3、稀疏索引

index文件中并没有为每一条message建立索引。而是采用了稀疏存储的方式，每隔一定字节的数据建立一条索引，这样的话就是避免了索引文件占用过多的空间和资源，从而可以将索引文件保留到内存中。缺点是没有建立索引的数据在查询的过程中需要小范围内的顺序扫描操作。

5.5、清理

一般情况下，Kafka 会根据设置的时间保留数据，把超过时效的旧数据删除掉。

1、不过，试想一下这样的场景 如果你使用 Kafka 保存客户的收货地址，那么保存客户的最新地址比 保存客户上周甚至去年的地址要有意义得多，这样你就不用担心会用错旧地址，而且短时 间内客户也不会修改新地址。

2、另外一个场景，一个应用程序使用 Kafka 保存它的状态，每次状态发生变化，它就把状态写入 Kafka。在应用程序从崩溃中恢复时，它从 ` Kafka` 读取消息来恢复最近的状态。在这种情况下，应用程序只关心它在崩溃前的那个状态，而不关心运行过程中的那些状态。

主题可以配置自己的保留策略，可以将消息保留到不再使用它们为止。

例如，用于跟踪用户活动的数据可能需要保留几天，而应 用程序的度量指标可能只需要保留几个小时。 也可以通过配置把主题当作紧凑型日志，只有最后一个带有特定键的消息会被保留下来。这种情况对于变更日志类型的数据来说比较适 用，因为人们只关心最后时刻发生的那个变更。

Kafka 通过改变主题的保留策略来满足这些使用场景。早于保留时间的旧事件会被删除， 为每个键保留最新的值，从而达到清理的效果。很显然，只有当应用程序生成的事件里 包含了键值对时，为这些主题设置 compact 策略才有意义。如果主题包含 null 键，清理 就会失败。

5.5.1、清理的工作原理

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

干净的部分：这些消息之前被清理过，每个键只有一个对应的值，这个值是上一次清理时保留下来的。

污浊的部分：这些消息时上一次清理之后写入的。稍后会被清理

如果kafka启动时启用了清理功能（log.cleaner.enabled参数），每个broker会启动一个清理管理器线程和多个清理线程，它们负责执行清理任务。这些线程会选择污浊率（污浊消息占分区总大小的比例）较高的分区进行清理。

为了清理分区，清理线程会读取分区的污浊部分，并在内存里创建一个map。map里的每个元素包含了消息键的散列值和消息的偏移量，键的散列值是16B，加上偏移量总共是24B。

如果要清理一个1GB的日志片段，并假设每个消息的大小为1KB，那么这个片段就包含一百万个消息，而我们只需要24MB的map就可以清理这个片段。如果有重复的键，可以重用散列项，从而使用更少的内存。这是非常高效的。

kafka配置中可以对map使用的内存大小进行配置。每个线程都有自己的map，而这个参数指的是所有线程可使用的内存总大小。如果为map分配了1GB内存，并使用了5个清理线程，那么每个线程可以使用200MB内存来创建自己的map。kafka并不要求分区的整个污浊部分来适应这个map的大小，但要求至少有一个完整的片段必须符合。如果不符合，那么kafka就会报错，要么配置分配更多的内存，要么减少清理线程数量，让每个线程可用内存变大。如果只有少部分片段可以完全符合，kafka将从最旧的片段开始清理，等待下一次清理剩余的部分。

清理线程在创建好偏移量map后，开始从干净的片段处读取消息，从最旧的消息开始，把它们的内容与map里的内容进行比对。它会检查消息的键是否存在于map中，如果不存在，那么说明消息的值是最新的，就把消息复制到替换片段上。如果键已存在，消息会被忽略，因为在分区的后部已经有一个具有相同键的消息存在。

在复制完所有消息之后，我们就将替换片段与原始片段进行交换，然后开始清理下一个片段。完成整个清理过程之后，每个键对应一个不同的消息，这些消息的值都是最新的。

清理的思想就是根据Key的重复来进行整理，注意，它不是数据删除策略，而是类似于压缩策略，如果key送入了值，对于业务来说，key的值应该是最新的value才有意义，所以进行清理后只会保存一个key的最新的value。

这个适用于一些业务场景，比如说key代表用户lD，Value用户名称，如果使用清理功能就能够达到最新的用户的名称的消息。

5.5.2、 被删除的事件

如果只为每个键保留最近的一个消息，那么当需要删除某个特定键所对应的所有消息时， 我们该怎么办?这种情况是有可能发生的，比如一个用户不再使用我们的服务，那么完全 可以把与这个用户相关的所有信息从系统中删除。

为了彻底把一个键从系统里删除，应用程序必须发送一个包含该键且值为 null 的消息。清理线程发现该消息时，会先进行常规的清理，只保留值为 null 的消息。

该消息(被称为墓 碑消息)会被保留一段时间，时间长短是可配置的。在这期间，消费者可以看到这个墓碑消息，并且发现它的值已经被删除。 于是，如果消费者往数据库里复制 Kafka 的数据，当它看到这个墓碑消息时，就知道应该要把相关的用户信息从数据库里删除。在这个时间段过后，清理线程会移除这个墓碑消息，这个键也将从 Kafka 分区里消失。

重要的是，要留给消费者足够多的时间，让他看到墓碑消息，因为如果消费者离线几个小时并错过了墓碑 消息，就看不到这个键，也就不知道它已经从 Kafka 里删除，从而也就不会去删除数据库 里的相关数据了。

5.5.3、何时会清理主题

就像 delete 策略不会删除当前活跃的片段一样，compact 策略也不会对当前片段进行清理。只有旧片段里的消息才会被清理。

在 0.10.0 和更早的版本里，Kafka 会在包含脏记录的主题数量达到 50% 时进行清理。这样 做的目的是避免太过频繁的清理(因为清理会影响主题的读写性能)，同时也避免存在太多脏记录(因为它们会占用磁盘空间)。浪费 50% 的磁盘空间给主题存放脏记录，然后进 行一次清理，这是个合理的折中，管理员也可以对它进行调整。

我们计划在未来的版本中加入宽限期，在宽限期内，我们保证消息不会被清理。对于想看 到主题的每个消息的应用程序来说，它们就有了足够的时间，即使时间有点滞后。