Kafka设计原理Design

1．设计意图 Motivation

    我们设计Kafka能够作为一种处理各种实时数据流的统一平台，我们为此设想了广泛的使用场景。

• 它必须能够支持高吞吐特性，以支持那些高容量的事件流，例如一个实时的日志收集场景；
• 它必须能够轻松得处理大量积压的数据，以支持那些定时从离线系统中加载大量数据的场景；
• 这同时也意味着它必须能够低延时地进行消息分发，以支持那些传统方式上的消息服务使用场景；

   我们希望Kafka能够分区、分布式和实时得处理数据流，并创建和得到处理后的新数据流feeds 。这些目标推动我们设计和实现了Kafka的分区特性和消费模型。

   最后地，在一些需要把Kafka数据流feeds注入到其它的数据服务系统的场景中，我们知道它还必须能够在发生机器故障条件下保证数据的容错性。

   为了支持所有的这些特性，我们在Kafka的设计实现过程中使用到了一些独特的元素。例如使用一个数据库日志的方式，而不是传统意义上的消息系统。我们在下面会对一些设计元素进行说明。

2．持久化设计 Persistence

1) Kafka可以非常快地存取文件系统

   Kafka重度得依赖文件系统存储和缓存消息数据。人们通常上会认为“磁盘读写是很慢的”，这也让人们对一个基于数据持久化结构设计和实现的系统是否可以提供高性能能力，抱着怀疑态度。而实际上，磁盘IO速度既可以很缓慢，也可以非常迅速，这完全取决于人们是怎么使用它。因此，在一个进行了恰当的磁盘数据存储结构设计的系统中，几乎经常可以达到像你的网络传输速度一样快（撑满你的网络）。

   在过去的十年中，衡量磁盘性能的最重要因素就是硬盘的吞吐量，而这是取决于硬盘的磁盘寻道延时。在一个验证性的测试中，基于一组SATA 7.2k转速且RAID5保护的磁盘，它的顺序写IO性能测试结果是600MB/sec，然而它的随机写IO的测试结果只有区区100k/sec，相差了6000倍。

   在ACM Queue article所做的一个测试中，他们实际上发现在一些使用场景中的顺序磁盘访问性能甚至比内存的随机读还要快。

   更进一步得说，我们的应用是基于JVM运行的，大家对Java memory usage会了解到的两个事实是：对象存储的内存耗费是非常大的，经常是实际数据容量的两倍以上；Java的垃圾回收机制变得越来越精巧难使用，而且在堆内数据增涨后会变慢。

   因此，使用文件系统和系统内核页缓存机制的效果会优于维护一个基于内存实现的缓存系统。通过访问全部的系统空闲内存，结合上存储一种字节结构紧凑设计的数据，几乎可以提升数倍的性能。例如，在一台32GB的物理服务器上，可以在不使用任何内存垃圾回收管理机制的条件下使用到大约30GB的系统缓存，而且即使在应用重新启动后，这些系统级缓存仍然是有效可用的，然而在常规意义上的进程级别缓存使用上，你必须要重建缓存（10GB Cache大约需要10min）。

   在Kafka的持久化设计中，所有的数据都是立即被写入持久化日志文件的，越过了从内存刷新到磁盘的步骤。实际上，这意味着消息数据是被直接传送给了内核的页面缓存（Kernel’s pagecache）。如果要了解更多关于这种pagecache-centric风格的设计知识，请参见这篇Varnish设计机制的说明文章（article）。

2) Kafka拥有常量时间存取的能力

   通常在消息中间件系统中使用到的数据持久化结构是B树。B树是一种非常多用途的数据结构，同时被用于消息系统各种事务性或非事务性的语义环境中。不过即使B树的算法复杂度只有O(logN)，但他们的确带来了相当高的成本。人们通常认为O(logN)和常量时间是实质等价的，然而这在磁盘IO读写上可不是真的。磁盘的一性导道操作需要10ms，而且一块磁盘同一时间只能做一次寻道动作，几乎没有并发空间。所以即使少量的磁盘寻道工和，也会带来很高的负载。还有很多公司使用到了专用的存储产品，这些专业设备在慢速的磁盘之上结合使用了很多快速缓存访问机制，提供了一个可观的存储性能。但是，这种专业存储设备可用的缓存空间是固定的，当你需要存取的数据量翻番时，你将会发现所能得到的最终存取性能甚至要下降一多半。

   我们可以直观地看到，一个实现了数据持久化的队列，提供简单的数据读和数据追加写到文件末尾，例如在一个日志管理系统中。这种数据结构的最大优势是所有操作的算法复杂度都是O(1)，磁盘读、写也不会互相阻塞。这使得Kafka具有了一个显著的性能优势，因为在这里性能与数据量实现了完全得解耦。一台服务器现在可以轻松利用到一组廉价的低转速、大容量（1+TB）磁盘能够提供的各种优势，虽然这些磁盘只有可怜的寻道速度但仍然能够基于大块数据的读、写提供可以接受的性能。这些磁盘普遍只有SAS磁盘价格的1/3和3倍以上的容量。

   拥有访问几乎无限的磁盘空间的能力，却不会有任何性能惩罚，这意味着我们可以基于Kafka实现一些在传统消息中间件中很少看到的特性。例如，在传统消息中间件系统中往往会在消息一旦被获取后立即尝试删除该消息数据，而Kafka能够为消息数据保留一个相对来说很长的时间（如一周）。仅这一个特性，就为消息消费端提供了大量的灵活性。

3．效率设计 Efficiency

避免产生大量的小I/O

   为了提升运行效率，尽可能得避免产生大量的小I/O，Kafka采用建立信息集合的方式把一组消息集中在一起批量处理，这也较一次只发送一条消息的方式减少了网络往返的开销。在服务器端每次是把消息块追加到日志文件，在消费者端则是每次都是顺序得获取一个大消息块数据。

   这种简单的优化就已经产生了显著得效果。通过批量处理消息数据，会产生更大的网络数据包、更多的磁盘顺序操作、连续的内存块等。这些特性使得Kafka可以把一个由随机消息突发产生的数据流转换为顺序地线性写盘操作后，再转给消费端处理。

字节级别数据复制的优化

   另外一个会严重影响性能效率的因素是字节级别的拷贝。这在一个小数据量的系统中不是个问题，但是在一个高负载系统中却会严重影响到效率。为了避免这一问题，我们设计和使用了一种标准的二进制消息数据格式，消息的producer, broker, consumer都遵守和共享这一标准格式。所以数据块（data chunks）可以在它们之间直接进行传输而不需要做任何修改。每个broker所维护着的那些由一个消息集的序列产生的日志文件，也是使用的这种标准的二进制数据格式，因此我们就可以直接通过网络传输持久化日志的数据块。

   Linux/Unix的sendfile机制。Kafka在通过网络传输数据时，还对Linux系统级别的网络数据传输机制做了调优，启用了系统的sendfile机制，减少数据在系统内核态、用户态间的冗余复制步骤，直接把数据从kernel pagecache写入网卡buffer，进行网络数据传输。这种优化设计，可以使得尤其是在一种有多个consumer消费同一个topic的场景中，几乎可以让consumers以可以使用的极限网络速率来获取消息数据。如果大家对这种结合使用pagecache和sendfile的方式需要了解更多资料，详见article 。

端到端的批量压缩 End-to-end Batch Compression

   在一些场景中，如在数据中心之间进行数据复制，系统的瓶颈往往不是CPU或者磁盘，而是网络带宽的限制，尤其是在需要通过广域网进行数据管道流传输的时候变得更加明显。虽然用户可以先把数据压缩一下再发送给Kafka，但这种基于每个消息数据的压缩办法很难获得系统级虽整体的性能提升。Kafka的办法是支持递归的消息集，一批消息可以被聚集在一起、压缩后发送给服务器。这批消息直接以压缩后的格式进行存储，即使在日志文件中也是这样，只有在consumer获取并处理这批消息时才会被解压缩。从producer到consumer，所以我们称之为End-to-end Batch Compression 。

端到端的批量压缩的实现原理可以参见官网这篇文章here 。

在producer端需要增加以下两个配置项：

配置项	描述	默认值
compressed.topics	comma separated list of topics for which compression should be enabled. This doesn't mean anything when compression.codec = 0	empty
compression.codec	Controls the compression codec to be used by the producer. O means no compression. 1 means GZIP compression. 2 means Snappy compression	0

当Kafka中一部分topics使用了端到端压缩，而其它topics没有的时候：

	compression.topics=empty	compression.topics="topicA,topicB"
compression.codec=0	All topics are uncompressed since compression is disabled	All topics are uncompressed since compression is disabled
compression.codec=1	All topics are compressed	Only the topics topicA and topicB are compressed

目前Kafka支持GZIP和Snappy compression codecs 。

4．生产者 The Producer

负载均衡

   在Kafka集群中的每个节点都会向Producer提供metadata元数据的查询服务，帮助producer发现和定位自己需要使用的活跃分区是哪一个。然后producer便可以把处理请求直接发往该topic的活跃分区，不需再经由什么中间的路由层次。

异步发送

   将多条消息暂且在producer客户端buffer起来，并将他们批量的发送到broker，小数据IO太多，会拖慢整体的网络延迟，批量延迟发送事实上提升了网络效率。这个批量处理的参数是可以配置的，例如数据量达到 64k 或者设置超时时间为 10 ms。

5．消费者 The Consumer

 consumer端向broker发送"fetch"请求,并告知其获取消息的offset;，然后consumer将会获得一定条数的消息数据。consumer端也可以重置offset来重新消费消息。

Push vs. pull

   在kafka中,采用了pull方式,即consumer在和broker建立连接之后,主动去pull(或者说fetch)消息;这种模式的优点是,首先consumer端可以根据自己的消费能力适时的去fetch消息并处理,且可以控制消息消费的进度(offset);此外,消费者可以良好的控制消息消费的数量batch fetch 。没有数据时，消费者在无用的忙碌，等待数据到来；为了避免这种情况，Kafka允许消费者阻塞在一个长poll等待直到数据到来。

 其它的以日志处理为中心的系统中，Scribe and Apache Flume等采用的是push方式。

Consumer Position

   追踪哪些被消费是消息系统的关键点；许多消息系统扩展性差立即删除是务实的选择。

   broker和consumer要在哪些消息已经被消费上面达到一致，并不是一件轻松的事情。如果每次消费者消费掉后就删掉数据，那么当消费者死掉，就可能会丢失数据。为解决这个问题，消息系统都加了确认机制。这种策略解决了丢数据问题，却引发了新的问题：如果消费者已经处理了一个消息，却在发送确认时失败了，就会导致再次处理该消息；性能问题--现在broker需要对每个消息保留多个状态。

   kafka的处理机制不同，topic是完全有序的partiiton集合，每个partition在一个时刻只有一个consumer消费。这意味着需要被消费的消息的位置在每个partiiton中的位置仅仅是一个单一的整数（下一个要处理的消息位置）。这使得哪些消息已经被消费过的这种状态，变得很简易，不过是每个分区上的一个整数。另外的一个好处在于，消费者可以根据自己的需要，例如遇到了应用bug造成部分消息处理错误，退回到某个旧的offset上去重新消费！

Offline Data Load

   Kafka可扩展的持久化特性对各种离线数据加载的使用场景也都有非常好得支持，例如周期性的向hadoop或数据仓库中批量加载数据。例如在hadoop中，通过把each node/topic/partition绑定起来，我们可以进行并行化地数据加载。Hadoop提供任务管理，如果其中一个数据加载任务失败了，只需简单得对该任务重新启动执行一次就ok了——没有任何数据重复的风险，因为只是退到原来的位置重新启动。

6．消息分发机制 Message Delivery Semantics

至多一次（失败不重传）；至少一次(不会丢失，可能重传)；精确一次。

   基本上是关于两个问题的处理办法：发布消息的持久化保证和消费消息的保证。exactly once，kafka中并没有严格的去实现(基于2阶段提交,事务)，这一点会在未来版本中实现，不过在大多数使用场景中并不需要这么完全的保证。

生产者角度

   在对延时敏感的用例中，我们可以设置producer的可靠性工作级别，比如设置花上10ms以等待确认消息发送出去了、或者是按照异步模式发送，或者设置为一直等待到leader分区已经收到消息为止。

从消费者角度看Kafka的消息分发机制

   在Kafka集群中，所有的复制分区都有相同offsets的相同日志数据。Consumer控制日志消费的offset。如果consumer端发生任何意外故障，导致需要启动新的进程继续处理消息数据，该怎么确保可以选择正确的position继续处理消息数据呢。

1) Consumer可以在获取到message后立即存到本地日志中，最后再处理该message。在这种条件下，如果consumer刚刚拿到一条消息并写入日志后，还没完成消息的处理工作，进程就挂掉了，那么在consumer进程重启后无疑会把这条消息当作已经处理完成的，而实际上是被漏掉了。这种消费方式对应着”at -most-once”模式。

2) Consumer可以在获取到message后立即进行消息处理工作，最后再把position信息写入日志。我们可以想像得到，如果一个consumer在处理了一个message后、在将其position写入日志前，进程挂掉了。那么无疑在consumer进程重新启动后，会把前面这个message再处理一遍。这种消费方式对应着”at-least-once”模式。其实在很多用例中，消息本身都是有一个唯一主键的，因此基于相同消息的updates也是幂等的（一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同）。

3) 那么怎么实现”exactly-once”呢。传统的办法是两阶段提交，比如将存储consumer position作为一个阶段，将存储consumers output作为另一个阶段。但这还有一种更简单的实现办法是，直接让consumer把它的offset和output存储在同一个地方。后者可能会更好一些，因为有很多输出系统并不支持consumer的两阶段提交操作。例如，hadoop ETL在向HDFS文件系统中填充数据时，是将数据和数据在HDFS中的offset一起存储的，这就保证了数据的offset是和读到的数据是被一同更新的。我们也是按照这种模式来满足需要严格保证机制的用例使用需求，或是满足那些因为messages没有有效的主键而无法处理重复消息的用例使用需求。

   因此，Kafka是默认的实现和提供了”at-least-once”的消息分发保证，然后又允许用户自行设定使用”at-most-once”功能。“exactly-once”的实现需要与目标存储系统的合作，不过Kafka提供的offsets使得这个实现变得简单。

7．复制 Replication

   kafka将每个partition数据复制到多个server上,任何一个partition有一个leader和多个follower(可以没有)。备份的个数可以通过broker配置文件来设定，Kafka自动维护leader和follower的失效转移。leader处理所有的read-write请求，follower需要和leader保持同步。Follower和consumer一样，消费消息并保存在本地日志中。leader负责跟踪所有的follower状态，如果follower"落后"太多或者失效，leader将会把它从replicas同步列表中删除。当所有的follower都将一条消息保存成功，此消息才被认为是"committed"，此时consumer才能消费它。即使只有一个replicas实例存活，仍然可以保证消息的正常发送和接收，只要zookeeper集群存活即可(不同于其他分布式存储，比如hbase需要"多数派"存活才行)。当然了，在Producer端是可以通过参数”request.required.acks”来控制自己是否要等待消息返回”committed”的响应。

   当leader失效时，需在followers中选取出新的leader，当然了只有处于”in-sync”状态的followers才有参选资格。可能此时follower落后于leader，因此需要选择一个"up-to-date"的follower。选择follower时需要兼顾一个问题就是新leader server上所已经承载的partition leader的个数。如果一个server上有过多的partition leader，意味着此server将承受着更多的IO压力。在选举新leader时需要考虑到"负载均衡"。Follower需要能够维护和ZooKeeper之间一个有效的会话，否则也会被判定为”unalive”。一个配置参数”replica.lag.time.max.ms”控制着一个follower数据同步滞后所能允许的最大延时。

Replicated Logs: Quorums, ISRs, and State Machines

   日志复制是Kafka系统的核心功能，同时也是分布式数据系统中最基本组成单元。日志复制机制同样地在状态机风格的分布式系统中有大量应用。无论是仲裁、中断服务程序还是状态机，每一种方法都需要去维护较多的复制日志的份数，尤其是当需要处理多于一个node失效的故障时。日志复制得越多，分布式集群的性能效率降低得越多。

   Kafka没有直接采用常规方法下的leader选举办法，而是使用了一种更轻巧的仲裁策略，它选择在ZooKeeper中维护一组处于”in-sync”状态的副本集(ISR)。Kafka写入分区的数据，只有这一组副本集都写成功后才会成功。所以这组副本集中任一个follower都有资格升级为leader。这组”in-sync”状态的副本集的任何信息变化都会持久化到ZooKeeper中去，用于Kafka选举leader时的负载均衡与失效转移策略。

   在Kafka中的另一项重要设计是Kafka不要求那些受到损坏的节点的数据做到完全的无损恢复。即使之前受损的节点丢失了部分未来得及持久化到磁盘中的数据，但是按照Kafka集群的服务管理协议，对于需要重新加入集群的节点，它需要做到”fully re-sync again“。

Unclean leader election: What if they all die?

   有一种极端状况是处于”in-sync”状态的副本集全部挂掉了，这时该采取什么策略推选新的leader。Kafka使用的方法是，选择第一个恢复可用的副本作为leader，无论它之前是否处在”in-sync”状态。这么设计的考虑是，我们根本无法预料那些已经挂掉了的前”in-sync“状态的副本集是否还能免恢复得了。所以权衡之下，先选择一个可用的副本作为leader以恢复集群对外的服务才显得更加重要。哪怕因此造成那么一部分的数据一致性问题。而这种处理方式普遍存在于各种基于仲裁模式实现的系统中。

可用性和持久性的保证 Availability and Durability Guarantees

   当向Kafka写数据时，producers是可以配置是否获得对消息发送的确定，配置参数的取值范围包括0，1和-1（即all）。其中request.required.acks=-1时不是意味着要等到所有的副本都返回的确认，而只是等待处于in-sync状态的副本集们全部返回了确认。

Kafka提供了两个topic-level的可用性配置项，它们是：

n 关闭”Unclean leader election“功能，即使让集群服务不可用，也不能接受任何的数据丢失风险；

n 设定一个最小ISR值，该值限定了处于in-sync状态的副本数量不能小于这个数值，否则集群服务就会停止对外服务。

副本管理 Replica Management

在实际的大规模分布式应用系统中，一个Kafka集群中很可能会有几百或几千年partitions，所以副本管理还意味着要考虑到分区的负载均衡要求，避免把大吞吐量的topics全放置在了少数几个节点上去。

此外leader的选举程序也是一个重要的优化内容。一般来说，如果一个节点整体性的失效了，那在它上面运行的很多个partitions，每个partition都要进行一次新的leader选举。在partitons数量很多时，这也是一个难于管理的工作。所以Kafka从brokers中选了一个作为集群的”controller”，这个controller负责检测集群brokers的健康状态并且负责在某个broker失效时处理它上面受影响partitions的leader选举管理工作。当controller也失效时，就从其它健康的brokers中另选一个作为新的controller。

8．日志压缩 Log Compaction

log compaction是一种更细的每条记录的保存机制，而不是比较粗的基于时间的保存。思路是：有选择的删除记录，我们有每个记录更新的更新；这种方式每条记录有最新的状态。

retention policy可以每个topic设置；可以看到一个集群上的topic可以有基于时间保留的、基于大小保留的、基于compaction保留的。任何从0offset读的处理会看到至少所有记录的最后状态。

配置参数（disabled by default）：

logcleaner. log.cleaner.enable=true; log.cleanup.policy=compact;

log compaction限制：不能配置还有多少日志没有compact;还不能与压缩的topic兼容。

9．配额 Quotas

从Kafka0.9开始，提供了基于消息发送请求和消息消费请求的配额管理功能。配额功能可以按每个client-id定义字节传输比率的阈值。一个client-id逻辑上等于一个应用，它是可以同时扮演多个producers以及consumers的，而Kafka配额限制是基于client-id进行的整性限制。如果一个名为”test-client”的client-id有10MB/sec的传输配额上限，那么这个带宽是被拥有相同client-id的实例们共享的。

我们为什么需要使用Kafka的配额管理功能？首先是解决那些设计和实现都很糟糕的客户端，避免它们的低劣行为对kafka集群形成事实上的DOS攻击结果；其次是可以用于面向多租户环境下的应用提供基于SLAs保证的api服务。

Kafka Quotas是基于每个broker进行配置的，并没有提供一种跨集群多brokers的配置同步机制。这样做的目的也是出于简化功能实现的需求。

那么，如果某一个client达到了Quotas的上限时，会遇到什么问题？基本上不是有任何问题，因为Kafka会检测并发现该配额事件，然后在该client收或发消息的过程中加入一定的延时，使其最终被控制在配额数值之下，而同时其消息发送或消费行为并不会受到影响。

当client-ids需要使用更高的配额时，除了去修改每个broker配置并重启应用之外，还有一种办法是修改ZooKeeper的配置文件/config/clients，在配置文件里设置一个集群默认使用的配额数值。所有brokers都会获取该配置信息并立即生效。这种配额配置办法，也不需要去重启集群的服务。