Kafka详细设计及其生态系统

Kafka生态-Kafka Core，Kafka Streams，Kafka Connect，Kafka REST Proxy和Schema Registry

Kafak的核心主要有Broker，Topic，日志，分区和集群。该核心还包括相关的工具，如MirrorMaker。

Kafka生态系统由Kafka Core，Kafka Streams，Kafka Connect，Kafka REST Proxy和Schema Registry组成。Kafka生态系统的大多数附件来自Confluent，而不是Apache。

Kafka Stream是一种Streams API，用于从流中转换，汇总和处理记录，并生成衍生流。 Kafka Connect是创建可重用的生产者和消费者的连接器API（例如DynamoDB的更改流）。通过REST（HTTP），Kafka REST代理用于生产者和消费者。Schema Registry使用Kafka记录的Avro来管理Schema。 Kafka MirrorMaker用于将集群的数据复制到另一个集群。

Kafka生态系统：Connect Source，Connect Sink和Kafka Stream的图

Kafka Connect Sources是记录的来源。 Kafka Connect Sinks是记录的目的地。

Kafka生态系统：Kafka REST代理和Confluent Schema Registry

Kafka Stream - 用于流处理的Kafka流

Kafka Stream API基于核心Kafka原语，拥有自己的生命。 Kafka Streams可实现流的实时处理。 Kafka Streams支持流处理器。流处理器从输入Topic中获取连续的记录流，对输入进行一些处理，转换，聚合，并产生一个或多个输出流。例如，视频播放器应用程序可能会收到观看或暂停视频的输入事件流，并输出用户对视频喜好的流，然后根据最近的用户活动或许多用户的总体活动来做出新的视频推荐以及查看哪些新的视频很热门。Kafka Stream API解决了无序记录、多个流的聚合和数据连接以及允许进行有状态计算的难题等等。

Kafka生态系统：Kafka Stream和Kafka Connect

Kafka生态系统回顾

什么是Kafka Stream(流)？

Kafka流可实现实时流处理。它可以跨多个流进行聚合，连接来自多个流的数据，允许有状态的计算等等。

什么是Kafka Connect(连接)？

Kafka Connect是创建可重用的生产者和消费者的连接器API（例如DynamoDB的更改流）。 Kafka Connect Sources是记录的来源。 Kafka Connect Sinks是记录的目的地。

什么是Schema Registry？

Schema Registry用于管理Kafka记录所使用的Avro的模式。

什么是Kafka Mirror Maker？

Kafka MirrorMaker用于将集群数据复制到另一个集群。

你什么时候可能需要使用Kafka REST代理？

通过REST（HTTP），Kafka REST代理用于生产者和消费者。您可以使用它来轻松集成现有的代码基线。

Kafka架构：底层设计

Kafka设计目的

LinkedIn工程师构建Kafka用来支持实时分析。Kafka旨在为实时流处理提供分析系统。 LinkedIn开发了Kafka作为流数据源实时处理的统一平台。 Kafka背后的目标是建立一个高吞吐量的流式数据平台，支持大量的事件流，如日志聚合，用户活动等。

为了扩展以满足LinkedIn Kafka的需求，它支持分布式，分片和负载均衡。实现扩展需要启发的Kafka分区和消费者模型。Kafka使用分区，分布式，提交日志来对写入和读取进行扩展或缩放。 Kafka的分片称为分区（Kinesis，类似于Kafka，称为分区“碎片”）。

根据维基百科，“数据库碎片是数据库或搜索引擎中数据的水平分区，每个分区称为分片或数据库分片，每个分片都保存在单独的数据库服务器实例上，以传播负载。

Kafka旨在处理来自离线系统以及传统消息用例（低延迟）的定期的大量数据负载。

MOM是面向消息的中间件;就像IBM MQSeries，JMS，ActiveMQ和RabbitMQ等MOM一样，Kafka通过复制和Leader选举对节点故障进行容错。然而，Kafka的设计更像是一个分布式数据库事务日志，而不是传统的消息系统。与许多MOM不同，Kafka复制被构建在低级设计中，而不是事后的想法。

持久性：拥抱文件系统

Kafka依赖文件系统来存储和缓存记录。

硬盘驱动器的顺序写入性能很快（真的很快）。 JBOD只是一堆磁盘驱动器。具有六个7200rpm SATA RAID-5阵列的JBOD写入性能大概为600MB /秒。像Cassandra表一样，Kafka日志是只写结构，意思是数据追加到日志末尾。当使用HDD时，操作系统可进行快速的、可预测的和优化的顺序读写。使用HDD，顺序磁盘存取比随机存储器和SSD更快。

虽然JVM GC的开销可能很高，但Kafka倾向于操作系统来获取大而快速的坚实缓存。而且，现代操作系统使用所有可用的主存进行磁盘缓存。操作系统的文件缓存几乎是免费的，没有操作系统的开销。实现正确的缓存一致性是一个挑战，但Kafka依赖于牢固的OS来实现缓存一致性。使用操作系统进行缓存也减少了缓冲区副本的数量。由于Kafka磁盘使用往往会执行顺序读取，因此操作系统预读缓存令人印象深刻。

Cassandra，Netty和Varnish使用类似的技术。所有这些都在Kafka文档中得到了很好的解释，而且在Varnish网站上也有更多有趣的解释。

大而快速的HDD和长顺序存取

Kafka喜欢用于读写的长顺序磁盘存取。就像Cassandra，LevelDB，RocksDB和其他的，Kafka使用一种日志结构化存储和压缩的形式而不是以磁盘上可变的BTree的形式。像Cassandra一样，Kafka使用tombstone而不是立即删除记录。

由于现在的磁盘有一些速度非常快且无限的空间，所以Kafka可以提供消息传递系统中通常不存在的功能，比如持续使用旧的消息。这种灵活性允许Kafka服务些有趣的应用。

Kafka生产者的负载均衡

生产者向Kafka的Broker索要哪个Kafka Broker拥有哪个Topic分区Leader而不需要路由层的元数据。这种领导关系数据允许生产者直接向Kafka Broker分区领导者发送记录。

生产者客户端控制哪个分区发布消息，并可以根据某些应用程序逻辑选择一个分区。生产者可以通过key，随机循环或使用自定义应用程序特定的分区逻辑来对记录进行分区。

Kafka生产者记录批处理

Kafka生产者支持记录的批处理。批处理可以按批量记录的字节大小进行配置。批量记录可以根据时间自动刷新。

批处理对网络IO吞吐量有好处，并大大加快了吞吐量。

缓冲是可配置的，您可以权衡延迟来获得更好的吞吐量。或在负载很重的系统中可以有更好的平均吞吐量和更低的总体延迟。

批处理允许累积更多的字节进行发送，这相当于Kafka Brokers上一些较大的I / O操作，并提高了压缩效率。为了提高吞吐量，Kafka Producer配置允许基于时间和大小的缓冲。生产者以较少的网络请求发送多条记录，而不是逐个发送每条记录。

Kafka生产者批处理

Kafka压缩

在大型流媒体平台中，瓶颈并不总是CPU或磁盘，而是网络带宽。作为多个服务可以共享NiC卡的容器化和虚拟化的云存在更多的网络带宽问题。此外，当将数据中心与数据中心或WAN通信时，更可能会带来网络带宽问题。

批处理有利于高效的压缩和网络IO吞吐量。

Kafka提供端对端批量压缩，而不是一次压缩一条记录，Kafka可有效一次压缩一批记录。相同的消息批次可以一次性压缩并发送到Kafka代理/服务器，并以压缩形式写入日志分区。您甚至可以配置压缩，以便在Kafka代理将压缩记录提供给消费者之前不发生解压缩。

Kafka支持GZIP，Snappy和LZ4压缩协议。

拉与推和流

Kafka消费者从Broker那里抽取数据。其他系统的Borker推送数据或流数据到消费者。消息通常是一个基于拉的系统（SQS，大多数MOM使用拉模式）。在拉式的系统中，如果一个消费者落后了，那么它可以稍后赶上。

由于Kafka是基于拉模式的，它实现了批处理数据的聚合。就像许多其它基于拉模式的系统一样，Kafka实现了长轮询（SQS和Kafka所做的）。长轮询在请求一段时间后会保持连接打开状态，并等待响应。

基于拉模式的系统必须拉取数据，然后处理它，并且拉取和获取数据之间总是有一个暂停。

基于推送的系统会将数据推送给消费者（scribe，flume，反应流，RxJava，Akka）。基于推送或流式传输的系统在处理缓慢或死亡的消费者方面存在些问题。在推送系统中，如果消费者的消费速度低于生产者的生产速率时可能会被压垮。一些基于推送的系统使用基于背压的回退协议，其允许消费者指示它被所看到的反应流被压垮了。当尝试跟踪消息确认时，不冲垮消费者和对消费者进行恢复通常是棘手的。

基于推送或流式传输的系统可以立即发送请求或累积请求并分批发送（或基于背压的组合）。基于推送的系统总是推送数据。消费者在处理已发送的数据时候累积消息来降低消息处理的延迟。然而，如果消费者在处理过程中死亡，那么Broker如何知道消费者在哪里，数据何时再次发送给另一个消费者,这个问题不容易解决。Kafka通过使用基于拉式的系统来解决这些复杂问题。

传统MOM消费者消息状态跟踪

对于大多数MOM，Broker有责任跟踪哪些消息被标记为消费。消息跟踪不是一件容易的事情。当消费者消费消息时，Broker会跟踪状态。

大多数MOM系统的目标是让Broker在消费后快速删除数据。记住大多数MOM是在磁盘小得多，功能更少，更昂贵的时候写的。

此消息跟踪比听起来更为棘手（确认功能），因为Broker必须保持大量状态来跟踪每条消息的发送、确认并要知道何时删除或重新发送消息。

Kafka消费者消息状态跟踪

记住，Kafka的Topic被分为有序的分区。每个消息在此有序分区中具有偏移量。每个Topic分区一次只被一个消费者群组中的一个消费者来消费。

该分区布局意味着，Broker跟踪每个消息的偏移量而不是消息（如MOM），但只需要每个消费者组的偏移量和分区偏移量的匹对存储。这个偏移量跟踪更少需要跟踪的数据。

消费者将位置数据周期性地（消费者组，分区偏移量对）发送到Kafka代理，并且代理将该偏移量数据存储到偏移量Topic中。

偏移量样式消息确认与MOM相比要便宜得多。此外，消费者更加灵活，可以快速回到更早的偏移量（重播）。如果有bug，那就修复bug，重放消费者并重放Topic。这种倒带功能是Kafka的杀手锏，因为Kafka可以长时间持有Topic日志数据。

消息传递语义

有三种消息传递语义：至多一次，至少一次，仅一次。最多一次时消息可能会丢失但不会重新投递。至少一次是消息永远不会丢失，但可能会重新被投递。仅一次是消息只发送一次。仅一次是首选但更昂贵，并且需要更多的生产者和消费者的簿记。

Kafka消费者和消息传递语义

回想一下，所有副本具有与相同偏移量完全相同的日志分区，并且消费者组维护其在每个主题分区日志中的位置。

为了实现“最多一次”的消费者消息读取，然后通过将其发送到代理来将偏移量保存到分区中，并最终处理该消息。 “最多一次”的问题是消费者可以在保存其位置后但在处理消息前死亡。然后接管或重新启动的消费者将在最后的位置离开，然后有问题的消息不会再被处理。

为了实现“至少一次”的消费者消息读取和处理，最后将偏移量保存到代理。 “至少一次”的问题是消费者在处理消息之后但在保存最后一个偏移位置之前可能会崩溃。然后如果消费者重新启动或由其他消费者来接管，则消费者可以接收到已经处理的消息。 “至少一次”是最常见的消息传递设置，您有责任使消息具有幂等性，这意味着获取相同的消息两次而不会导致问题（两次借记）。

为了在消费者端实现“仅一次”，消费者需要在消费者位置存储和消费者的消息输出存储之间做一个两阶段提交。或者，消费者可以将消息处理输出存储在与最后一个偏移量相同的位置。

Kafka提供了前两个，而且以消费者的角度来实现第三个。

Kafka生产者的持久性和确认

Kafka提供了可持续性操作的可预测性的语义。发布消息时，消息将“提交”到日志中，这意味着所有的ISR都接受了消息。只要至少有一个副本存在，则这个提交策略就可以有很好的持久性。

生产者的连接可能会在发送中间下降，生产者可能无法确定其发送的消息是否会通过，然后生产者会重新发送消息。这个重新发送逻辑就是为什么使用消息Key和使用幂等消息（重复确定）很重要的原因。Kafka并没有保证从生产者重新尝试得到的消息不会重复。

生产者可以重新发送消息，直到收到确认，即确认被收到了。生产者重新发送消息而不需知道其发送的其他消息是否发出了，消除了“一次”和“最多一次”的消息传递语义。

生产者耐用性

生产者可以指定耐用性水平。生产者可以等待一个消息被提交。等待提交确保所有副本都具有该消息的副本。

生产者可以发送未确认（0）。也可以发送只需从分区领导者那获得一个确认(1)。生产者也可以发送并等待所有副本的确认（-1），默认值是-1。

改进的生产者（2017年6月发行的版本）

Kafka现在支持从生产者那里“仅一次”的消息传递、性能的改进和跨分区的原子写入。他们通过生产者发送序列ID来实现这一点，代理将会保持跟踪生产者是否发送了这个序列，如果生产者尝试再发送它，它将会得到一个重复消息的确认，不会保存任何东西到日志中。这种改进不需要API更改。

Kafka生产者的原子日志写（2017年6月发行的版本）

Kafka的另一个改进是Kafka生产者跨分区进行原子写入。原子写入意味着Kafka消费者只能看到提交的日志（可配置）。Kafka有一个协调员，将标记写入主题日志，以表示已经成功处理的内容。事务协调器和事务日志保持原子写入的状态。

原子写入需要一个新的生产者API用于事务。

以下是使用新的生产者API的示例。

用于交易的新的生产者API

Kafka复制

Kafka通过可配置数量的Kafka Broker复制每个主题的分区。默认情况下，Kafka的复制模型不是像大多数MOM一样的螺栓功能，因为Kafka从一开始就要使用分区和多节点。每个主题分区有一个领导和零个或多个追随者。

领导者和追随者被称为副本。复制因子是领导节点加上所有追随者。Kafka Broker平均分享分区领导权。消费者只能从领导者那里读取。生产者只能写给领导者。

追随者上的主题日志分区与领导者的日志保持同步，ISR是领导者的精确副本减去正在运行中所需要复制的记录。追随者就像一个普通的Kafka消费者一样从他们的领导人那里批量拉取记录。

Kafka Broker故障转移

Kafka跟踪哪些Kafka的Broker还活着。为了活着，Kafka Broker必须使用ZooKeeper的心跳机制来保持ZooKeeper会话，并且必须让所有跟随者与领导者同步而不能落后太远。

对于被称为同步的代理活动，ZooKeeper的会话和同步都是必需的。同步副本称为ISR。每个领导者都会跟踪一组“同步副本”。

如果ISR /追随者死亡或落后了，则领导者将从该组ISR中移除跟随者。落后是指在一个replica.lag.time.max.ms时间段后，副本未处于同步。

当所有ISR将消息应用于其日志时，消息被视为“已提交”。消费者只看到已提交的消息。卡夫卡保证：只要有至少一个ISR，承诺的邮件就不会丢失。

复制的日志分区

Kafka的分区是一个复制的日志。复制的日志是分布式数据系统原语。复制的日志对于使用状态机实现其他分布式系统非常有用。复制日志模型在一系列有价值的数据上达成共识。

当领导者活着的时候，所有的追随者只需要从他们的领导中复制值和顺序。如果领导者死亡，Kafka从同步的追随者中选出一个新的领导者。如果一个生产者被告知信息已经被提交了，然后领导者失败了，那么新当选的领导者就必须有这个已提交的消息。

你拥有越多的ISR，在领导失败的时候就会要更多的选举。

Kafka和法定人数

法定人数是所需确认的数量和必须比较的日志数量，以选择领导者，以确保可用性。大多数系统使用多数投票，Kafka不会使用简单的多数投票来提高可用性。

在Kafka里，基于完整的日志来选举领导者。如果我们的复制因子为3，则在领导者声明发送的消息已提交之前，至少要有两个ISR必须处于同步。如果一个新的领导人需要当选，那么新的领导人不能失败超过3次，新的领导人要确保有所有提交的消息。

在跟随者中，必须至少要有一个副本包含了所有已提交的消息。多票法定人数的问题是它不会采取多次失败来获得一个不可操作的集群。

多数ISR的Kafka法定人数

Kafka为每个领导者维护了一些ISR。只有ISR里的成员才有资格进行领导选举。在所有ISR确认写之前，生产者对分区的写入都不会被提交。每当ISR设置更改时，ISR将持久到ZooKeeper中。只有作为ISR成员的副本才有资格当选领导者。

这种风格的ISR仲裁允许生产者在没有大部分节点的情况下继续工作，但只是一个ISR的多数投票。这种风格的ISR仲裁还允许一个副本重新加入到ISR集中并具有其投票数，但是在加入之前必须完全重新同步，即使副本在其崩溃期间丢失了未刷新的数据。

如果所有节点同时死亡了怎么办？

Kafka对数据丢失的保证仅在至少一个副本同步时才有效。 如果正在复制一个分区的领导者下的所有追随者同时死亡，则数据会丢失，Kafka会保证数据无效。如果一个分区的所有副本都down了，则默认情况下，Kafka会选取第一个副本作为领导者（不一定在ISR集中）（config unclean.leader.election.enable = true是默认值）。这种选择有利于可靠性的一致性。

如果一致性比您用例的可用性更重要，那么您可以设置config unclean.leader.election.enable = false，然后如果一个分区的所有副本都down了，则Kafka等待第一个存在的ISR成员（不是第一个副本）选举成为新领导人。

生产者选取耐用性

生产者可以通过将acks设置为 - 没有（0），仅限于（1）或所有副本（-1）来选择持久性。

acks = all是默认值。当所有当前的同步复制（ISR）收到消息时，都会发生ack。

您可以在一致性和可用性之间进行权衡。如果优先于可用性的耐久性，则禁用不好的领导者选举，并指定最小的ISR大小。

最小ISR大小越高，一致性的保证会越好。但是如果ISR集的大小小于最小阈值，则ISR的最小值越小，则会降低分区的可用性因为分区对写入不可用。

配额

Kafka有对消费者和生产者的配额限制来限制他们对带宽的消费。这些配额阻止消费者或生产者摆脱所有Kafka Broker的资源。配额是客户端ID或用户。配额数据存储在ZooKeeper中，所以更改不需要重新启动Kafka的Broker。

Kafka底层设计与架构回顾

你如何防止来自写性能差的消费者的拒绝服务攻击？

使用配额来限制消费者的带宽。

默认生产者耐用性的级别（acks）是什么？

所有。这意味着所有ISR必须将消息写入其日志分区。

如果所有Kafka节点down机，默认情况下会发生什么？

Kafka选择第一个副本不一定在ISR集合中）作为领导者因为unclean.leader.election.enable = true默认用于支持可用性。

为什么Kafka记录批处理很重要？

通过电线以及磁盘优化IO吞吐量。它还通过压缩整个批次来提高压缩效率。

Kafka的一些设计目标是什么？

成为高吞吐量，可扩展的流数据平台，用于实时分析大量事件流，如日志聚合，用户活动等。

截至2017年6月，Kafka的一些新功能是什么？

生产者原子写入，性能改进和生产者不发送重复的消息。

有哪些不同的消息传递语义？

有三种消息传递语义：至多一次，至少一次，仅一次。