[大数据]Kafka简介

1.简介

1.1.概述

Kafka是最初由Linkedin公司开发，是一个分布式、分区的、多副本的、多订阅者，基于zookeeper协调的分布式日志系统（也可以当做MQ系统），常见可以用于web/nginx日志、访问日志、消息服务等等，Linkedin于2020年贡献给了Apache基金会并成为顶级开源项目。
主要应用场景是：日志手机系统和消息系统。
Kafka主要设计目标如下：

1. 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。
2. 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条信息的传输。
3. 支持Kafka Server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。
4. 同时支持离线数据处理和实时数据处理。
5. Scale out：支持在线水平扩展。

1.2.消息系统介绍

一个消息系统负责将数据从一个应用传递到另外一个应用，应用只需关注于数据，无须关注数据在两个或多个应用间是如何传递的。分布式消息传递基于可靠的消息队列，在客户端应用和消息系统之间异步传递消息。有两种主要的消息传递模式：点对点传递模式、发布-订阅模式。大部分的消息系统选用发布-订阅模式。Kafka就是一种发布-订阅模式。

1.3.点对点消息传递模式

在点对点消息系统中，消息持久化到一个队列中。此时，将有一个或多个消费者消费队列中的数据。但是一条消息只能被消费一次。当一个消费者消费了队列中的某条数据之后，该条数据则从消息队列中删除。该模式即使有多个消费者同时消费数据，也能保证数据处理的顺序。
生产者发送一条消息到queue，只有一个消费者能收到。

1.4.发布-订阅消息传递模式

在发布-订阅消息系统中，消息被持久化到一个topic中。与点对点消息系统不同的是，消费者可以订阅一个或多个topic，消费者可以消费该topic中所有的数据，同一条数据可以被多个消费者消费，数据呗消费后不会立马删除。在发布-订阅消息系统中，消息的生产者称为发布者，消息者称为订阅者。
发布者发送到topic的消息，只有订阅了topic的订阅者才会收到消息。

2.Kafka的优点

2.1.解耦

在项目启动之初预测将来项目会碰到什么需求，是极其困难的。消息系统在处理中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

2.2.冗余（副本）

有些情况下，处理数据的过程会失败。除非数据被持久化，否则将会造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失的风险。许多消息队列所采用的“插入-获取-删除”范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

2.3.扩展性

因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要修改代码、不需要调节参数。扩展就像调大电力按钮一样简单。

2.4.峰值处理能力

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

2.5.可恢复性

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

2.6.顺序保证

在大多使用场景下，数据处理的顺序很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。Kafka保证一个Partition内的消息的有序性。

2.7.缓冲

在任何重要的系统中，都会有需要不同的处理时间的元素。例如，加载一张图片比应用过滤器花费更少的时间。消息队列通过一个缓冲层来帮助任务最高效率的执行–写入队列的处理尽可能的快速。该缓冲有助于控制和优化数据流经过系统的速度。

2.8.异步通信

在很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

3.常用Message Queue对比

3.1.RabbitMQ

RabbitMQ是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP、XMPP、SMTP、STOMP，也正因如此，它非常重量级，更适合与企业级的开发。同时实现了Broker构架，这意味着消息在发送给客户端时现在中心队列排队。对路由、负载均衡或者数据持久化都有很好的支持。

3.2.Redis

Redis是一个基于Key-Value队的NoSQL数据库，开发维护很活跃。虽然他是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。

3.3.ZeroMQ

ZeroMQ号称最快的消息队列系统，尤其针对大吞吐量的需求场景。ZeroMQ能够实现RabbitMQ不擅长的高级/复杂的队列，但是开发人员需要自己组合多种技术框架，技术上的复杂度是对这个MQ能够应用成功的挑战。ZeroMQ具有一个独特的非中间件的模式，你不需要安全和运行一个消息服务器或中间件，因为你的应用程序将扮演这个服务器角色。你只需要简单的引用ZeroMQ程序库，可以使用NuGet安装，然后你就可以愉快的在应用程序之间发送消息了。但是ZeroMQ仅提供非持久性的队列，也就是说如果宕机，数据将会丢失。

3.4.ActiveMQ

ActiveMQ是Apache下的一个子项目。类似于ZeroMQ，他能够以代理人和点对点的技术实现队列，同时类似于RabbitMQ，它少量的代码就可以高效地实现高级应用场景。

3.5.Kafka/Jafka

Kafka是Apache下的一个子项目，是一个高性能跨语言分布式发布/订阅消息队列系统，而Jafka是在Kafka之上孵化而来的，即Kafka的一个升级版。具有以下特性：快速持久化，可以在O(1)的系统开销下进行消息持久化；高吞吐，在一台普通的服务器上即可以达到10W/s的吞吐速率；完全的分布式系统，Broker、Producer、Consumer都原生自动支持分布式，自动实现负载均衡；支持Hadoop数据并行加载，对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka通过Hadoop的并行加载机制统一了在线和离线的消息处理。Apache Kafka相对于ActiveMQ是一个非常轻量级的消息系统，除了性能非常好之外，还有一个工作良好的分布式系统。

4.Kafka中的术语解释

4.1.概述

上图中一个topic配置了3个partition。Partition1有两个offset：0和1。Partition2有4个offset。Partition3有1个offset。副本的id和副本所在的机器的id恰好相同。
如果一个topic的副本数为3，那么Kafka将在集群中为每个partition创建3个相同的副本。集群中的每个broker存储一个或多个partition。多个producer和consumer可同时生产和消费数据。

4.2.broker

Kafka集群包含一个或多个服务器，服务器节点称为broker。
broker存储topic的数据。如果某topic有N个partition，集群有N个broker，那么每个broker存储该topic的一个partition。
如果某topic有N个partition，集群有（N+M）个broker，那么其中有N个broker存储该topic的一个partition，剩下的M个broker不存储该topic的partition数据。
如果某topic有N个partition，集群中broker数目少于N个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致Kafka集群数据不均衡。

4.3.Partition

topic中的数据分割为一个或多个partition。每个topic至少有一个partition。每个partition中的数据使用多个segment文件存储。partition中的数据是有序的，不同partition间额数据丢失了数据的顺序。如果topic有多个partition，消费数据时就不能保证数据的顺序。在需要严格保证消息的消费顺序的场景下，需要将partition数目设为1。

4.4.Producer

生产者即数据的发布者，该角色将消息发布到Kafka的topic中。broker接收到生产者发送的消息后，broker将消息追加到当前用于追加数据的segment文件中，生产者发送的消息，存储到一个partition中，生产者也可以指定数据存储的partition。

4.5.Consumer

消费者可以从broker中读取数据。消费者可以消费多个topic中的数据。

4.6.Consumer Group

每个Consumer属于一个特定的Consumer Group(可为每个Consumer指定group name，若不指定group name则属于默认的group)。

4.7.Leader

每个partition有多个副本，其中有且仅有一个作为Leader，Leader是当前负责数据的读写的partition。

4.8.Follower

Follower跟随Leader，所有写请求都通过Leader路由，数据变更会广播给所有Follower，Follower与Leader保持数据同步。如果Leader失效，则从Follower中选举出一个新的Leader。当Follower与Leader挂掉、卡住或者同步太慢，leader会把这个follower从“in sync replicas” (ISR)列表中删除，重新创建一个Follower。

5.使用案例

5.1.消息

Kafka很好地替代了传统的message broker（消息代理）。Message brokers可用于各种场合（如将数据生成器与数据处理解耦，缓冲未处理的消息等）。与大多数消息系统相比，Kafka拥有更好的吞吐量、内置分区、具有复制和容错的功能，这使它成为一个非常理想的大型消息处理应用。
根据我们的经验，通常消息传递使用较低的吞吐量，但可能要求较低的端到端延迟，Kafka提供强大的持久性来满足这一要求。
在这方面，Kafka可以与传统的消息传递系统（ActiveMQ和RabbitMQ）相媲美。

5.2.跟踪网站活动

Kafka的初始用例是将用户活动跟踪灌到重建为一组实时发布-订阅源。这意味着网站活动（浏览网页、搜索或其他的用户操作）将被发布到中心topic，其中每个活动类型有一个topic。这些订阅源提供一系列用例，包括实时处理、实时监控、对加载到Hadoop或离线数据仓系统的数据进行离线处理和报告等。
每个用户浏览网页时都生成了许多活动信息，因此活动跟踪的数据量通常非常大。

5.3.度量

Kafka通常用于监控数据。这涉及到从分布式应用程序中汇总数据，然后生成可操作的集中数据源。

5.4.日志聚合

许多人使用Kafka来替代日志聚合解决方案。日志聚合系统通常从服务器手机物理日志文件，并将其置于一个中心系统（可能是文件服务器或HDFS）进行处理。Kafka从这些日志文件中提取信息，并将其抽象为一个更加清晰的消息流。这样可以实现更低的延迟处理且易于支持多个数据源及分布式数据的消耗。与Scribe或Flume等以日志为中心的系统相比，Kafka具备同样出色的性能、更强的耐用性（因为复制功能）和更低的端到端的延迟。

5.5.流处理

许多Kafka用户通过管道来处理数据，有多个阶段：从Kafka topic中消费原始输入数据，然后聚合，修饰或通过其他方式转化为新的topic，以供进一步消费或处理。例如，一个推荐新闻文章的处理管道可以从RSS订阅源抓取文章内容并将其发布到“文章”topic；然后对这个内容进行标准化或者重复的内容，并将处理完的文章内容发布到新的topic；最终它会尝试将这些内容推荐给用户。这种处理管道基于各个topic创建实时数据流图。从0.10.0.0开始，在Apache Kafka中，kafka Streams可以用来执行上述的数据处理，它是一个轻量但功能强大的流处理库。除Kafka Streams外，可供替代的开源流处理工具还包括Apache Storm和Apache Samza。

5.6.采集日志

Event sourcing是一种应用程序设计风格，按时间来记录状态的更改。Kafka可以存储非常多的日志数据，为基于event sourcing的引用程序提供强有力的支持。

5.7.提交日志

Kafka可以从外部为分布式系统提供日志提交功能。日志有助于记录节点和行为间的数据，采用重新同步机制可以从失败节点恢复数据。Kafka的日志压缩功能支持这一用法。这一点与Apache BookKeeper项目类似。

6.设计思想

6.1.动机

Kafka设计的目的是能够作为一个统一的平台来处理大公司可能拥有的所有实时数据馈送。要做到这点，必须要考虑到想当广泛的用例：

1. Kafka必须具有高吞吐量来支持高容量事件流，例如实时日志聚合。
2. Kafka需要能够正常处理大量的数据积压，以便能够支持来自离线系统的周期性数据加载。
3. Kafka必须处理低延迟分发，来处理更传统的消息传递用例。
4. Kafka必须支持对这些反馈尽心分区，分布式，以及实时处理来创建新的分发馈送等特性。由此产生了我们的分区模式和消费者模式。
5. Kafka在数据流被推送到其他数据系统进行服务的情况下，要求系统在出现机器故障时必须能够保证容错。

6.2.持久化

Kafka对消息的存储和缓存严重依赖文件系统。人们对于“磁盘速度慢”的普遍印象，使得人们对持久化的架构能够提供强有力的性能产生怀疑。事实上，磁盘的速度比人们预期的要慢得多，也快得多，这取决于人们使用磁盘的方式。而且设计合理的磁盘结构通常可以和网络一样快。
关于磁盘性能的关键事实是，磁盘的吞吐量和过去十年里磁盘的寻址延迟不同。因此，使用6个7200rpm、SATA接口、RAID-5的磁盘列阵在JBOD配置下的顺序写入的性能约为600MB/秒，但随即写入的性能仅约为100k/秒，相差6000倍以上。因为线性的读取和写入是磁盘使用模式中最有规律的，并且由操作系统进行了大量的优化。现代操作系统提供了read-ahead和write-behind技术，read-ahead是以大的data block为单位预先读取数据，而write-behind是将多个小型的逻辑写合并成一次大型的物理磁盘写入。
为了弥补这种性能差异，现代操作系统在越来越注重使用内存对磁盘进行cache。现在操作系统主动将所有空闲内存用作disk caching，代价是在内存回收时性能会有所降低。所有对磁盘的读写操作都会通过这个统一的cache。如果不使用直接I/O，该功能不能轻易关闭。因此即使进程维护了in-process cache，改数据也可能会被复制到操作系统的pagecache中，事实上所有内容都被存储了两份。
此外，Kafka建立在JVM之上，任何了解Java内存使用的都知道两点：

1. 对象的内存开销非常高，通常是所存储的数据的两倍（甚至更多）。
2. 随着堆中数据的增加，Java的垃圾回收变得越来越复杂和缓慢。
   受这些因素影响，相对于维护in-memory cache或者其他结构，使用文件系统和pagecache显得更有优势–我们可以通过自动访问所有空闲内存将可用缓存的容量至少翻倍，并且通过存储紧凑的字节结构而不是独立的对象，有望将缓存容量再翻一番。这样使得32GB的机器缓存容量可以达到28-30GB，并且不回产生额外的GC负担。此外，即使服务重新启动，缓存依旧可用，而in-process cache则需要在内存中重建（重建一个10GB的缓存可能需要10分钟），否则进程就要从cold cache的状态开始（这意味着进程最初的性能表现十分糟糕）。这同时也极大的简化了代码，因为所有保持cache和文件系统之间一致性的逻辑现在都被放到了OS中，这样做比一次性的进程内缓存更准确、更高效。如果你的磁盘使用更倾向于顺序读取，那么read-ahead可以有效的使用每次从磁盘中读取到的有用数据预先填充cache。
   这里给出了一个非常简单的设计：当比于维护尽可能多的in-memory cache，并且在空间不足的时候匆忙将数据flush到文件系统，我们把这个过程倒过来。所有数据一开始就被写入到文件系统的持久化日志中，而不用在cache空间不足的时候flush到磁盘。实际上，这表明数据被转移到了内核的pagecache中。

6.3.常量时间就足够了

消息系统使用的持久化数据结构通常是和BTree相关联的消费者队列或者其他用于存储消息源数据的通用随机访问数据结构。BTree是最通用的数据结构，可以在消息系统能够支持各种功能事务性和非事务性语义。虽然BTree的操作复杂度是O(logN)，但成本也相当高。通常我们认为O(logN)基本等同于常数时间，但这条在磁盘操作中不成立。磁盘寻址是每10ms一跳，并且每个磁盘同时只能执行一次寻址，因此并行性受到了限制。因此即使是少量的磁盘寻址也会很高的开销。由于存储系统将非常快的cache操作和非常慢的物理磁盘操作混合在一起，当数据随着fixed cache增加时，可以看到树的性能通常是非线性的–比如数据翻倍时性能下降不只两倍。
所以直观来看，持久化队列可以建立在简单的读取和向文件后追加两种操作之上，这和日志解决方案相同。这种架构的优点在于所有的操作复杂度都是O(1)，而且读操作不会阻塞写操作，读操作之间也不会互相影响。这有着明显的性能优势，由于性能和数据大小完全分离开来–服务器现在可以充分利用大量廉价、低转速的1+TB SATA硬盘。虽然这些硬盘的寻址性能很差，但他们在大规模读写方面的性能是可以接受的，而且价格是原来的三分之一、容量是原来的三倍。
在不产生任何性能损失的情况下能够访问几乎无限的硬盘空间，这意味着我们可以提供其他消息系统不常见的特性。例如：在Kafka中，我们可以让消息保留相对较长的一段时间（比如一周），而不是试图在被消费后立即删除。正如我们后面将要提到的，这给消费者带来了很大的灵活性。

6.4.Efficiency

Kafka在性能上做了很大的努力。其主要的使用场景是处理WEB活动数据，这个数据量非常大，因为每个页面都有可能大量的写入。此外我们假设每个发布message至少被一个consumer（通常很多个consumer）消费，因此我们尽可能的去降低消费的代价。
我们还发现，从构建和运行许多相似系统的经验上来看，性能是多租户运行的关键。如果下游的基础设施服务很轻易被应用层冲击形成瓶颈，那么一些小的改变也会造成问题。通过非常快的（缓存）技术，我们能确保应用层冲击基础设施之前，将负载稳定下来。当尝试去运行支持集中式集群上成百上千个应用程序的集中式服务时，这一点很重要，因为应用层使用方式几乎每天都会发生变化。
对于磁盘性能，一旦消除了磁盘访问模式不佳的情况，该类系统性能低下的主要原因就剩下了两个：大量的小型I/O操作，以及过多的字节拷贝。
小型的I/O操作发生在客户端和服务端之间以及服务端自身的持久化操作中。
为了避免这种情况，我们的协议是建立在一个“消息块”的抽象基础上，合理将消息分组。这使得网络请求将多个消息打包成一组，而不是每次发送一条信息，从而使整组消息分担网络中往返的开销。Consumer每次获取多个大型有序的消息块，并由服务器端一次将消息块一次加载到它的日志中。
这个简单的优化对速度有着数量级的提升。批处理允许更大的网络数据包，更大的顺序读写磁盘操作，连续的内存块等等，所有这些都是Kafka将随机流消息顺序写入到磁盘，再由consumers进行消费。
另一个低效率的操作是字节拷贝，在消息量较少时，这不是什么问题。但是在高负载的情况下，影响就不容忽视。为了避免这种情况，我们使用producer，broker和consumer都共享的标准化的二进制消息格式，这样数据块不用修改就能在他们之间传递。
broker和consumer都共享的标准化的二进制消息格式，这样数据块不用修改就能在他们之间传递。
broker维护的消息日志本身就是一个文件目录，每个文件都由一系列以相同格式写入到磁盘的消息集合组成，这种写入格式被producer和consumer公用。保持这种通用格式可以对一些很重要的操作进行优化：持久化日志块的网络传输。现代的unix操作系统提供了一个高度优化的编码方式，用于将数据从pagecache转移到socket网络连接中；在Linux中系统调用了sendfile做到这一点。
为了理解sendfile的意义，了解数据从文件到套接字的常见数据传输路径就非常重要：
1.操作系统从磁盘读取数据到内核空间的pagecache。
2.应用程序读取内核空间的数据到用户空间的缓冲区。
3.应用程序将数据（用户空间的缓冲区）写回内核空间到套接字缓冲区（内核空间）。
4.操作系统将数据从套接字缓冲区（内核空间）复制到通过网络发送的NIC缓冲区。

这显然是低效的，有四次copy操作和两次系统调用。使用sendfile方法，可以允许操作系统将数据从pagecache直接发送到网络，这样避免重新复制数据。所以这种优化方式，只需要最后一步copy操作，将数据复制到NIC缓冲区。
我们期望一个普遍的应用场景，一个topic被多消费者消费。使用上面提交的zero-copy（零拷贝）优化，数据在使用时只会被复制到pagecache中一次，节省了每次拷贝到用户空间内存中，再从用户空间进行读取的消耗。这使得消息能够以接近网络连接速度的上限进行消费。
pagecache和sendfile的组合使用意味着，在一个kafka集群中，大多数consumer消费时，是看不到磁盘上的读取活动，因为数据将完全由缓存提供。

6.5.端到端的批量压缩

在某些情况下，数据传输的瓶颈不是CPU，也不是磁盘，而是网络带宽。对于需要通过广域网在数据中间之间发送消息的数据管道尤其如此。当然，用户可以在不需要Kafka支持下一次一个的压缩消息。但是这样回造成非常差的压缩比和消息重复类型的冗余，比如JSON中的字段名称或Web日志中的用户代理或公共字符串值。高性能的压缩是一次压缩多个消息，而不是压缩单个消息。

6.6.Producer

Load balancing

生产者直接发送数据到主分区的服务器上，不需要经过任何中间路由。为了让生产这实现这个功能，所有的kafka服务器节点都能响应这样的元数据请求：哪些服务器是活着的，主题的那些分区是主分区，分配在哪个服务器上，这样生产者就能适当地直接发送它的请求到服务器上。
客户端控制消息发送数据到哪个分区，这个可以实现随机的负载均衡方式，或者使用一些特定的语义的分区函数。我们有提供特定分区的接口让用于根据指定的键值进行hash分区（当然也有选项可以重写分区函数），例如，如果使用用户ID作为key，则用户相关的所有数据都会被分发到同一个分区上。这允许消费者在消费数据时做一些特定的本地化处理。这样的分区风格经常被设计用于一些本地处理比较敏感的消费者。

Asynchronous send

批处理是提升性能的一个主要驱动，为了允许批量处理，kafka生产者会尝试在内存中汇总数据，并用一次请求批次提交信息。批处理，不仅仅可以配置指定的消息数量，也可以指定等待特定的延迟时间（如64k或10ms），这允许汇总更多的数据后再发送，在服务器端也会减少更多的IO操作。该缓冲是可配置的，并给出了一个机制，通过权衡少量额外的延迟时间获取更好的吞吐量。

6.7.Consumer

Kafka consumer通过向broker发出一个“fetch”请求来获取它想要消费的partition。consumer的每个请求都在log中指定了对应的offset，并接收从该位置开始的一大块数据。因此，consumer对于该位置的控制就显得极为重要，并且可以在需要的时候通过回退该位置再次消费对应的数据。

push vs pull

究竟是由consumer从broker那里pull数据，还是由broker将数据push到consumer。Kafka在这方面采取了一种较为传统的设计方式，也是大多数的消息系统所共享的方式：即producer吧数据push到broker，然后consumer从broker中pull数据。也有一些logging-centric的系统，比如Scribe和Apache Flume，沿着一条完全不同的push-based的路径，将数据push到下游节点。这两种方法都有优缺点。然而，由于broker控制着数据传输速率，所以push-based系统很难处理不同的concumer。让broker控制数据传输速率主要是为了让consumer能够以可能的最大速率消费；不幸的是，这导致着在push-based的系统中，当消费速率低于生产效率时，consumer往往会不堪重负（本质上类似于拒绝服务攻击）。pull-based系统有一个很好的特性，那就是当consumer速率落后于producer时，可以在适当的时间赶上来。还可以通过使用某种backoff协议来减少这种现象：即consumer可以通过backoff表示它已经不堪重负了，然而通过获取负载情况来充分使用consumer（但永远不超载）这一方式实现起来比它看起来更棘手。前面以这种方式构件系统的尝试，引导者Kafka走向了更传送的pull模型。
另一个pull-based系统的优点在于：它可以大批量生产要发送给consumer的数据。而push-based系统必须选择立即发送请求或者积累更多的数据，然后在不知道下游的consumer能否立即处理它的情况下发送这些数据。如果系统调整为低延迟状态，这就会导致一次只发送一条消息，以至于传输的数据不再被缓冲，这种方式是极度浪费的。而pull-based的设计修复了该问题，因为consumer总是将所有可用的（或者刀刀配置的最大长度）消息pull到log当前位置的后面，从而使得数据能够得到最佳的处理而不会引入不必要的延迟。
简单的pull-based系统的不足之处在于：如果broker中没有数据，consumer可能会在一个紧密的循环中结束轮询，实际上busy-waiting直到数据到来。为了避免busy-waiting，Kafka在pull请求中加入参数，使得consumer在一个“long pull”中阻塞等待，直到数据到来（还可以选择等待给定字节长度的数据来确保传输长度）。
你可以想象它可能的只基于pull的，end-to-end的设计。例如：producer直接将数据写入一个本地的log，然后broker从producer那里pull数据，最后consumer从broker中pull数据。通常提到的还有“store-and-forward”式producer，这是一种很有趣的设计，但是它和Kafka设定的有数以千记的生产者的应用场景不太相符。在实践中，Kakfa通过大规模运行的带有强大的SLAs的pipeline，而省略producer的持久化过程。

消费者的位置

持续追踪已经被消费的内容是消息系统的关键性能点之一。
大多数消息系统都在broker上保存被消费消息的元数据。也就是说，当消息被传递给consuer，broker要么理解在本地记录该事件，要么等待consumer的确定后再记录。这是一种相当直接的选择，而且事实上对于单机服务器来说，也没有与其他地方能够存储这些状态信息。由于大多数消息系统用于存储的数据结构规模都很小，所以这也是一个很实用的选择–因为只要broker知道哪些消息被消费了，就可以在本地立即进行删除，一直保持较少的数据量。
也许不太明显，但要让broker和consumer就被消费的数据保持一致性也不不是一个小问题。如果broker在每条信息被发送到网络的时候，立即将其标记为consumed，那么一旦consumer无法处理该消息（可能由consumer崩溃或者请求超时或者其他原因导致），该消息就会丢失。为了解决消息丢失的问题，许多消息系统增加了确认机制：即当消息被发送出去的时候，消息仅被标记为sent而不是consumed。这个策略修复了消息丢失的问题，但也产生了新问题。首先，如果consumer处理了消息但在发送确认之前出错了，那么该消息就会被消费两次。第二个是关于性能的，现在broker必须为每条信息保存多个状态（优先对其加锁，确保该消息只被发送一次，然后将其永久的标记为consumed，以便将其移除）。还有更棘手的问题要处理，比如如何处理已经发送但一直得不到确认的消息。
Kafka使用完全不同的方式解决消息丢失问题。Kafka的topic被分割成了一组完全有序的partition，其中每一个partition在任意给定的时间内只能被每个订阅了这个topic的consumer组中的一个consumer消费。这意味着partition中的每个consumer的位置仅仅是一个数字，即下一条要消费的消息的offset。这使得被消费的消息的状态信息相当少，每个partition只需要一个数字。这个状态信息还可以作为周期性的checkpoint。这以非常低的代价实现了和消息确认机制等同的效果。
这种方式还有一个附加的好处。consumer可以回退之前的offset来再次消费之前的数据，这个操作违反了队列的基本原则，但事实证明对大多数consumer来说这是一个必不可少的特性。例如，如果consumer的代码有bug，并且在bug被发现之前已经有一部分数据被消费了，那么consumer可以在bug修复后通过回退到之前的offset来再次消费这些数据。

离线数据加载

可伸缩的持久化特性允许consumer只进行周期性的消费，例如批量数据加载，周期性将数据加载到诸如Hadoop和关系型数据库之类的离线系统中。
在Hadoop的应用场景中，我们通过将数据加载分配到多个独立的map任务来实现并行化，每一个map负责一个node/topic/partition，从而达到充分并行化。Hadoop提供了任务管理机制，失败的任务可以重新启动而不会有重复数据的风险，只需要简单的从原来的位置重启即可。

6.6.消息交付语义

Kafka在producer和consumer之间提供语义保证。Kafka可以提供的消息交付语义保证有多种：

1. At most once – 消息可能会丢失但绝不重传。
2. At least once – 消息可以重传但绝不丢失。
3. Exactly once – 每一条消息只能被传递一次。

值得注意的是，这个问题被分成了两部分：发布消息的持久性保证和消费消息的保证。
很多系统声称提供了“Exactly once”的消费交付语义，然而阅读他们的细则很重要，因为这些声称大多数都是误导性的（即它们没有考虑consumer或producer可能失败的情况，以及存在多个consumer进行处理的情况，或者写入磁盘的数据可能丢失的情况）。
Kafka的语义是直接了当的。发布消息时，我们会有一个消息的概念被“committed”到log中。一旦消息被提交，只要有一个broker备份了该消息写入的partition，并保持“alive”状态，该消息就不会丢失。假设存在完美无缺的broker，然后来试着理解Kafka对producer和consumer的语义保证。如果一个producer在试图发送消息的时候发生了网络故障，则不确定网络错误发生在消息提交之前还是之后。这与使用自动生成的键插入到数据库表中的语义场景很相似。
在0.11.0.0之前的版本汇总，如果producer没有收到表明消息已经被提交的响应，那么producer除了将消息重传之外别无选择。这里提供的是at-least-once的消息交付语义，因为如果最初的请求事实上执行成功了，那么重传过程中该消息就会被再次写入到log当中。从0.11.0.0版本开始，Kafka producer新增了幂等性的传递选项，该选项保证重传不会在log中产生重复条目。为了实现这个目的，broker给每个producer都分配了一个ID，并且producer给每条被发送的消息分配了一个序列号来避免产生重复的消息。同样也是从0.11.0.0版本开始，producer新增了使用类似事务性的语义将消息发送到多个topid partition的功能：也就是说，要么所有的消息都被成功的写入到了log，要么一个都没有写进去。这种语义的主要应用场景就是Kafka topic之间的exactly-once的数据传递。
并非所有使用场景都需要这么强的保证。对于延迟敏感的应用场景，我们允许生产者指定它需要的持久性级别。如果producer指定了他想要等待的消息被提交，则可以使用10ms的量级。然而，producer也可以指定它想要完全异步地执行发送，或者它只想等待直到leader节点拥有该消息（follower节点有没有无所谓）。
现在从consumer的视角来描述语义，所有的副本都有相同的log和相同的offset。consumer负责控制它在log中的位置。如果consumer永远不崩溃，那么它可以将这个位置只存储在内存中。但如果consumer发生了故障，我们希望这个topic partition被另一个进程接管，那么新进程需要选择一个合适的位置开始进行处理。假设consumer要读取一些消息–它有几个处理消息和更新位置的选项。

1. consumer可以先读取消息，然后将它的位置保存到log中，然后再对消息进行处理。在这种情况下，消费者进程可能会在保存其位置之后，带还没有保存消息处理的输出之前发生崩溃。而在这种情况下，即使在此位置之前的一些消息没有被处理，接管处理的进程将从保存的位置开始。在consumer发生故障的情况下，这对应于“at-most-once”的语义，可能会有消息得不到处理。
2. consumer可以先读取消息，然后处理消息，最后再保存它的位置。在这种情况下，消费者进程可能会在处理了消息之后，但还没有保存位置之前发生崩溃。而在这种情况下，当新的进程接管后， 它最初收到的一部分消息都已经被处理过了。在consumer发生故障的情况下，这对应于“at-least-once”的语义。在许多应用场景中，消息都设有一个主键，所以更新操作是幂等的（想通过的消息接收两次时，第二次写入会覆盖掉第一次写入的记录）。

那么 exactly once语义呢？当从一个kafka topic中消费并输入到另一个topic时（正如一个Kafka Streams应用中所做的那样），可以使用上文提到的0.11.0.0版本中的新事物型producer，并将consumer的位置存储为一个topic中的消息，所以我们可以在输出topic接收已经被处理的数据的时候，在同一个事务中向Kafka写入offset。如果事务被中断，则消费者的位置将恢复到原来的值，而输出topic上产生的数据对其他消费者是否可见，取决于事务的“隔离级别”。在默认的“read_uncommitted”隔离级别中，所有消息对consumer都是可见的，即使他们是中止的事务的一部分，但是在“read_committed”的隔离级别中，消费者只能访问已提交的事务中的消息（以及任何不属于事务的消息）。
在写入外部系统的引用场景中，限制在于需要在consumer offset的存储和consumer的输出结构的存储之间引入two-phase commit。但这可以用更简单的方法处理，而且通常的做法是让consumer将其offset存储与其输出相同的位置。这也是中更好的方式，因为大多数consumer想写入的输出系统都不支持two-phase commit。举个例子，Kafka Connect连接器，它将所读取的数据和数据的offset一起写入到HDFS，以保证数据和offset都被更新，或者两者都不被更新。对于其他很多需要这些较强语义，并且没有主键来避免消息重复的数据系统，Kafka也遵循类似的模式。
因此，事实上Kafaka在Kafka Streams中支持了exactly-once的消息交付功能，并且在topic之间进行数据传递和处理时，通常使用事务型producer/consumer提供exactly-once的消息交付功能。到其他目标系统的exactly-once的消息交付通常需要与该类系统协作，但Kafka提供了offset，使得这种场景的实现变得可行。否则，Kafka默认保证at-least-once的消息交付，并且Kafka允许用户通过禁用producer的重传功能和让consumer在处理一批消息之前提交offset，来实现at-most-once的消息交付。

6.7.Replication

Kafka允许topic的partition拥有若干副本，在server端可以配置partition的副本数量。当集群中的节点出现故障时，能自动进行故障转移，保证数据的可用性。
创建副本的单位是topic的partition，正常情况下，每个分区都有一个leader和零或多个followers。总的副本数是包含leader的总和。所有的读写操作都由leader处理，一般partition的数量都比broker的数量多的多，各分区的leader均匀的分布在brokers中。所有的followers节点都同步leader节点的日志，日志总的消息和偏移量都和leader中的一致。当然，在任何给定时间，leader节点的日志末尾时可能有几个消息尚未被备份完成。
Followers节点就像普通的consumer那样从leader节点那里拉取消息并保存在自己的日志文件中。Followers节点可以从leader节点那里批量拉取消息日志到自己的日志文件中。
与大多数分布式系统一样，自动处理故障需要精确定义节点“alive”的概念。Kafka判断节点是否存活有两种方式。
1.节点必须可以维护和Zookeeper的连接，Zookeeper通过心跳机制检查每个节点的连接。
2.如果节点是个follower，它必须能及时的同步leader的写操作，并且延时不能太久。

满足这两个条件的节点处于“in sync”状态，区别于“alive”和“failed”。Leader会追踪所有“in sync”的节点。如果有节点挂掉了，或是写超时，或是心跳超时，leader就会把它从同步副本副本列表中移除。同步超时和写超时的时间由replica.lag.time.max.ms配置确定。
现在，只有当消息被所有的副本节点加入到日志中时，才算是提交，只有提交的消息才会被consumer消费，这样就不用担心一旦leader挂掉了消息会丢失。另一方面，producer也可以选择是否等待消息被提交，这取决他们的设置在延迟时间和持久性之间的权衡，这个选项是由producer使用的acks设置控制。请注意，topic可以设置同步备份的最小数量，producer请求确认消息是否被写入到所有的备份时，可以用最小同步数量判断。如果producer对同步的备份数没有严格的要求，即使同步的备份数量低于最小同步数量（例如，仅仅只有leader同步了数据），消息也会被提交，然后被消费。
在所有时间里，Kafka保证只要有至少一个同步中的节点存活，提交的消息就不会丢失。
节点挂掉后，经过短暂的故障转移后，Kafka将仍然保持可用性，但在网络分区（network partitions）的情况下可能不能保持可用性。

备份日志：Quorums、ISRs和状态机

Kafka的核心是备份日志文件。备份日志文件是分布式数据系统最基础的要素之一，实现的方法也有很多种。其他系统也可以用kafka的备份日志模块来实现状态机风格的分布式系统。
备份日志按照一系列有序的值（通常是编号为0、1、2…）进行建模。有很多方法可以实现这一点，但最简单和最快的方法是由leader节点选择需要提供有序的值，只要leader节点还存活，所有的follower只需要拷贝数据并按照leader节点的顺序排序。
当然，如果leader永远不会挂掉，那我们就需要follower了。但是如果leader crash，我们就需要从follower中选举出一个新的leader。但是folowers自身也有可能落后或者crash，所以我们必须确保leader的候选者们是一种数据同步最新的follower节点。
如果选择写入时候需要保证一定数量的副本写入成功，读取时需要保证读取一定数量的副本，读取和写入之间有重叠。这样的读写机制称为Quorum。
这种权衡的一种常见方法是对提交策略和leader选举使用多数投票机制。Kafka没有采取这种方式，但是我们还是要研究一下这种投票机制，来理解其中蕴含的权衡。假设我们有2f+1个副本，如果在leader宣布消息提交之前必须由f+1副本收到该消息，并且如果我们从这至少f+1个副本之中，有着最完整的日志记录的follower里来选择一个新的leader，那么在故障次数少于f的情况下，选举出的leader保证具有所有提交的消息。这是因为在任意f+1个副本中，至少有一个副本一定包含了所有提交的消息。该副本的日志监视最完整的，因此将被选为新的leader。这个算法都必须处理许多其他的细节（例如精确定义怎么使日志更加完整，确保在leader down掉期间，保证日志一致性或者副本服务器的副本集的改变），但是现在我们将忽略这些细节。
这种大多数投票方法有一个非常好的优点：延迟是取决于最快的服务器。也就是说，如果副本数是3，则备份完成的等待时间取决于最快的Follower。
大多数投票的缺点是，多数的节点挂掉让你不能选择leader。要冗余单点故障需要三分数据，并且要冗余两个故障需要五份的数据。在一个系统中，仅仅靠冗余来避免单点故障是不够的，但是每写5次，对磁盘空间需求是5倍，吞吐量下降到1/5，这对于处理海量数据问题是不切实际的。这可能是为什么quorum算法更常用于共享集群配置（如ZooKeeper），而不适用于原始数据存储的原因，例如HDFS中namenode的高可用是建立在基于投票的元数据，这种代价高昂的存储方式不适用于数据本身。
Kafka采取了一种稍微不同的方法来选择它的投票集。Kafka不适用大多数投票选择leader。Kafka动态维护了一个同步状态的备份的集合（a set of in-sync replicas）,简称ISR，在这个集合中的节点都是都是和leader保持高度一致的，只有这个集合的成员才有资格被选举为leader，一条信息必须被这个集合所有节点读取并追加到日志中了，这条信息才能视为提交。这个ISR集合发生变化会在ZooKeeper持久化，正因为如此，这个集合中的任何一个节点都有资格被选为leader。这对于Kafka使用模型中，有很多分区和并确保主从关系是很重要的。因为ISR模型和f+1副本，一个Kafka topic冗余f个节点故障而不会丢失任何已经提交的信息。
在实际中，为了冗余f节点故障，大多数投票和ISR都会在提交消息前确认相同数量的备份被收到（例如在一次故障生存之后，大多数的quorum需要三个备份节点和一次确认，ISR只需要两个备份节点和一次确认），大多数投票方法的一个优点是提交时能避免最慢的服务器。但是，通过允许客户端选择是否阻塞消息提交来改善，和所需的备份数较低而产生的额外的吞吐量和磁盘空间时值得的。
另一个重要的设计区别是，Kafka不要求崩溃的节点恢复所有的数据，在这种空间中的复制算法经常依赖于存在“稳定存储”，在没有违反潜在的一致性的情况下，出现任何故障再恢复情况下都不会丢失。这个假设有两个主要的问题。首先，我们在持久性数据系统的实际操作中观察到的最常见的问题是磁盘错误，并且它们通过不能保证数据的完整性。其次，即使磁盘错误不是问题，我们也不希望在每次写入时都要求使用fsync来保证一致性，因为这会使性能降低两到三个数量级。我们的协议能确保备份节点重新加入ISR之前，即使它挂时没有新的数据，它也必须完整再一次同步数据。

Unclean leader选举：如果节点全挂了？

请注意，Kafka对于数据不会丢失的保证，是基于至少一个节点在保持同步状态，一旦分区上的所有备份节点都挂了，就无法保证了。
但是，实际在运行的系统需要去考虑一旦所有的备份都挂了，怎么去保证数据不会丢失，这里有两种实现的方法：

1. 等待一个ISR的副本重新恢复正常服务，并选择这个副本作为领leader（它有极大可能拥有全部数据）。
2. 选择第一个重新恢复正常服务的副本（不一定是ISR中的）作为leader。

这是可用性和一致性的简单妥协，如果我只等待ISR的备份节点，那么只要ISR备份节点都挂了，我们的服务将一直会不可用，如果它们的数据损坏了或者丢失了，那就会是长久的宕机。另一方面，如果不是ISR中的节点恢复服务并且我们允许它成为leader，那么它的数据就是可信的来源，即使它不能保证记录了每一个已经提交的信息。Kafka默认选择第二种策略，当所有的ISR副本都挂掉时，会选择一个可能不同步的备份作为elader，可以配置属性unclean.leader.election.enable禁用此策略，那么就会使用第一种策略即停机时间由于不同步。
这种困境不止有Kafka遇到，它存在于任何quorum-based规则中。例如，在大多数投票算法当中，如果大不多述服务器永久性的挂了，那么要么选择丢失100%的数据，要么违背数据的一致性选择一个存活的服务器作为数据可信的来源。

可用性和持久性保证

向Kafka写数据时，producers设置ack是否提交完成，0：不等待broker返回确认消息，1：leader保存成功返回，-1（all）：所有的备份都保存成功返回，请注意，设置“ack = all”并不能保证所有的副本都写入了消息。默认情况下，当acks = all时，只要ISR副本同步完成，就会返回消息已经写入。例如，一个topic仅仅设置了两个副本，那么只有一个ISR副本，那么当设置acks = all时返回写入成功时，剩下了的那个副本数据也可能数据没有写入。尽管这确保了分区的最大可用性，但是对于偏好数据持久性而不是可用性的一些用户，可能不想用这种策略，因此，我们提供了两个topic配置，可用于优化配置消息数据持久性：
1.禁用unclean leader选举机制-如果所有的备份节点都挂了，分区数据就会不可用，直到最近的leader恢复正常。这种策略优先于数据丢失的风险。
2.指定最小的ISR集合大小，只有当ISR的大小小于最小值，分区才能接受写入操作，以防止仅写入单个备份的消息丢失造成消息不可用的情况，这个设置只有在生产者使用acks = all的情况下才会生效，这至少保证消息被ISR副本写入。此设置是一致性和可用性之间的折中，对于设置更大的最小ISR大小保证了更好的一致性，因为它保证将消息写入了更多的备份，减少了消息丢失的可能性。但是，这会降低可用性，因为如果ISR副本的数量低于最小阈值，那么分区将无法写入。

备份管理

以上关于备份日志的讨论只涉及单个日志文件，即一个topic分区，事实上，一个Kafka集群管理者成百上千这样的partitions。kafka尝试以轮询调用的方式将集群内的partition负载均衡，避免大量topic拥有的分区几种在少数几个节点上。同样，我们也试图平衡leadership，以至于每个节点都是部分partition的leader节点。
优化主从关系的选举过程也是重要的额，这是数据不可用的关键窗口。原始的实现是当有节点挂了后，进行主从关系选举时，会对挂掉的节点的所有partition的领导权重新选举。相反，我们会选择一个broker作为“controller”节点。controller节点负责检测brokers级别故障，并负责在broker故障的情况下更改这个故障的Broker中的partition的leadership。这种方式可以批量的通知主从关系的变化，使得对于拥有大量的partition的broker，选举过程的代价更低并且速度更快。如果controller节点挂了，其他存活的broker都可能成为新的controller节点。

6.8.日志压缩

日志压缩可确保Kafka始终至少为单个topic partition的数据日志中的每个message key保留最新的已知值。这样的设计解决了应用程序崩溃、系统故障后恢复或者应用在运行维护过程中重启后重新加载缓存的场景。
目前，Kafka日志保留方法很简单，当旧的数据保留时间超过指定时间、日志达到规定大小后就丢弃。这样的策略非常适用于处理那些暂存的数据，例如记录每条消息之间相互独立的日志，然而在实际使用过程中还有一种非常重要的场景–根据key进行数据变更（例如更改数据库表内容），使用以上方法显然不行。
举一个处理这样的流式数据的具体例子。假设有一个topic，里面的内容包含用户的email地址；每次用户更新它们的email地址时，Kafka发送一条消息到这个topic，这里使用用户Id作为消息的key值。现在，我们在一段时间内为id为123的用户发送一些消息，每个消息对应email地址的改变（其他ID消息省略）：

123 => bill@microsoft.com
        .
        .
        .
123 => bill@gatesfoundation.org
        .
        .
        .
123 => bill@gmail.com

日志压缩提供了更精细的保留机制，所以Kafka至少保留每个key的最后一次更新（例如：bill@gmail.com）。这样Kafka保证日志包含每一个key的最终值而不只是最近变更的完整快照。这意味着下游的消费者可以获得最终的状态而无需拿到所有的变化的消息信息。
让我们先看几个有用的使用场景，然后再看看如何使用它。
1.数据库更改订阅。通常需要在多个数据系统设置拥有一个数据集，这些系统中通常有一个是某种类型的数据库（无论是RDBMS或者新流行的key-value数据库）。例如，你可能有一个数据库，缓存，搜索引擎群或者Hadoop集群。每次变更数据库，也同时要变更缓存、搜索引擎以及hadoop集群。在只需处理最新日志的实时更新的情况下，你只需要最近的日志。但是，如果你希望能够重新加载缓存或恢复搜索失败的节点，你可能需要一个完整的数据集。
2.事件源。这是一种应用程序设计风格，它将查询处理与应用程序设计相结合，并使用变更的日志作为应用程序的主要存储。
3.日志高可用。执行本地计算的进程可以通过注销对其本地状态所做的更改来实现容错，以便另一个进程可以重新加载这些更改并在故障时继续进行。一个具体的例子就是在流查询系统中进行计数，聚合和其他类似"group by"的操作。实时流处理框架Samza，使用这个特性便是由于这个原因。

在这些场景中，主要需要处理变化的实时feed，但是偶尔当机器崩溃或者需要重新加载或重新处理数据时，需要处理所有数据。日志压缩允许在同一topic下同时使用这两个用例。
想法很简答，我们有无限的日志，以上每种情况记录变更日志，我们从一开始就捕获每次变更。使用这个完整的日志，我们可以通过回放日志来恢复到任何一个时间点的状态。然而这种假设的情况下，完整的日志是不实际的，对于那些每一行记录会变更多次的系统，即使数据集很小，日志也会无限的增长下去。丢弃旧日志的简单操作可以限制空间的增加，但是无法重建状态–因为旧的日志被丢弃，可能一部分记录的状态无法重建（这些记录所有的状态变更都在旧日志中）。
日志压缩机制是更细粒度的、每个记录都保留的机制，而不是基于时间的粗粒度。这个理念是选择性的删除哪些有更新的变更的记录的日志。这样最终日志至少包含每个key的记录的最后一个状态。
这个策略可以为每个Topic设置，这样一个集群中，可以一部分Topic通过时间和大小保留日志，另外一些可以通过压缩策略保留。

日志压缩基础

这个是一个高级别的日志逻辑图，展示了Kafka日志的每条信息的offset逻辑结构。

Log head中包含传统的Kafka日志，它包含了连续的offset和所有的消息。日志压缩增加了处理tail log 的选项。上图展示了日志压缩的Log tail的情况。tail中的消息保存了初次写入时的offset。即使该offset的消息被压缩，所有offset仍然在日志中是有效的。在这个场景下，无法区分和下一个出现的更高offset的位置。如上面的例子中，36,37,38是属于相同位置的，从他们开始读取日志都将从38开始。
压缩也允许删除。通过消息的key和空负载（null payload）来识别该消息可从日志中删除。这个删除标记将会引起所有之前拥有相同key的消息被移除（包括拥有key相同的新消息）。但是删除标记比较特殊，它将在一定周期后被从日志中删除来释放空间。这个时间点被称为“delete retention point”，如上图。
压缩操作通过在后台周期性的拷贝日志段来完成的。清除操作不会阻塞读取，并且可以被配置不超过一定IO吞吐来避免影响Producer和Consumer。实际的日志段压缩过程有点像这样：

日志压缩的保障措施如下：

1. 任何滞留在日志head中的所有消费者能看到写入的所有消息；这些消息都是有序的offset。topic使用min.compaction.lag.ms来保障消息写入之前必须经过的最小时间长度，才能被压缩。这限制了一条消息在Log Head中的最短存在时间。
2. 始终保持消息的有序性。压缩永远不会重新排序消息，只是删除了一些。
3. 消息的Offset不会变更。这是消息在日志中的永久标志。
4. 任何从头开始处理日志的Consumer至少会拿到每个key的最终状态。另外，只要Consumer在小于Topic的delete.retention.ms设置的时间段内到达Log head，将会看到所有删除记录的所有删除标记。换句话说，因为移除删除标记和读取时同时发生的，Consumer可能会因为落后超过delete.retention.ms而导致错过删除标记。

日志压缩的细节

日志压缩由Log Cleaner执行，后台线程池重新拷贝日志段，移除那些key存在于Log Head中的记录。每个压缩线程如下工作：

1. 选择log head与log tail比率最高的日志。
   2.在head log中为每个key的最后offset创建一个简单的概要。
2. 它从日志的开始到结束，删除哪些在日志中最新出现的key的旧的值。新的、干净的日志将会立即被交到日志中，所以只需要一个额外的日志段空间（不是日志的完整副本）。
3. 日志head的概要本质上是一个空间密集型的哈希表，每个条目使用24个字节。所以如果有8G的整理缓冲区，则能迭代处理大约366G的日志头部（假设消息大小为1K）。