消息队列之Kafka——从架构技术重新理解Kafka

Apache Kafka® 是 一个分布式流处理平台 . 这到底意味着什么呢?

我们知道流处理平台有以下三种特性:

1. 可以让你发布和订阅流式的记录。这一方面与消息队列或者企业消息系统类似。
2. 可以储存流式的记录，并且有较好的容错性。
3. 可以在流式记录产生时就进行处理。

Kafka适合什么样的场景?

它可以用于两大类别的应用:

1. 构造实时流数据管道，它可以在系统或应用之间可靠地获取数据。 (相当于message queue)
2. 构建实时流式应用程序，对这些流数据进行转换或者影响。 (就是流处理，通过kafka stream topic和topic之间内部进行变化

Kafka有四个核心的API:

• The  Producer API  允许一个应用程序发布一串流式的数据到一个或者多个Kafka topic。
• The  Consumer API  允许一个应用程序订阅一个或多个 topic ，并且对发布给他们的流式数据进行处理。
• The  Streams API  允许一个应用程序作为一个 流处理器 ，消费一个或者多个topic产生的输入流，然后生产一个输出流到一个或多个topic中去，在输入输出流中进行有效的转换。
• The  Connector API  允许构建并运行可重用的生产者或者消费者，将Kafka topics连接到已存在的应用程序或者数据系统。比如，连接到一个关系型数据库，捕捉表（table）的所有变更内容。

在Kafka中，客户端和服务器使用一个简单、高性能、支持多语言的 TCP 协议.此协议版本化并且向下兼容老版本， 我们为Kafka提供了Java客户端，也支持许多其他语言的客户端。

————————————————————————————————————————————————

以上摘自Apache Kafka官网

而本文关注的焦点是：构造实时流数据管道，即message queue部分。也就是我们常使用的“消息队列”部分，这部分本身也是Kafka最初及最基本的底层设计。

让我们回到最初Kafka还没有设计出来的时候，通过重新设计Kafka，一步步了解为什么Kafka是我们现在看到的样子，到时我们将了解到Kafka作为消息队列会高吞吐量、分布式、高容错稳定。我们把这个项目命名为： Kafka-R 。

现在我们开始设计 Kafka-R ，我们正式设计 Kafka-R 之前需要考虑设计目标，也就是我的 Kafka-R 设计出来到底是用来干嘛的，适用于什么业务场景，解决什么需求痛点。

可以很快想到：数据交换。这是消息队列的基本功能与要求。

然后呢？可以作为个大平台，支持多语言，最好能满足大公司的业务需求，而且最好是实时的，至少是低延迟。

概括起来就是：我们设计 Kafka-R 的目标是可以作为一个统一的平台来处理大公司可能拥有的所有实时数据馈送。

为了满足我们的 Kafka-R 的设计目标，那么 Kafka-R 需要具备以下这些特征：

具有高吞吐量来支持高容量事件流。

能够正常处理大量的数据积压，以便支持来自离线系统的周期性数据加载。

系统必须处理低延迟分发，来处理更传统的消息传递用例。

数据馈送分区与分布式，以及实时。

系统在出现机器故障时能够保证容错。

一、数据的存储方式——in-memory&in-disk

有两种选择：第一种，使用in-memory cache，并在空间不足的的时候将数据flush到文件系统中。

另外一种，使用in-disk，一开始把所有的数据写入文件系统的持久化日志中。

我们的 Kafka-R 采用in-disk。实际上在此情况数据被转移到了内核的pagecache中。

“磁盘速度慢”是人们的普遍印象，那么 Kafka-R 的数据存储和缓存基于文件系统，这样的性能能够接受吗？

而事实是，磁盘的速度比人们预期的要慢得多，也快得多，取决于人们使用磁盘的方式。

我们知道磁盘有顺序读和随机读两种模式，之间的性能差异很大，但具体差距多少呢？

使用6个7200rpm、SATA接口、RAID-5的磁盘阵列在JBOD配置下的顺序写入的性能约为600MB/秒，但随机写入的性能仅约为100k/秒，相差6000倍。

线性的读取和写入是磁盘使用模式中最有规律的，并且操作系统进行了大量的优化。现代操作系统提供了read-ahead和write-behind技术，read-ahead是以大的data block为单位预先读取数据，而write-hehind将多个小型的逻辑写合并成一次大型的物理磁盘写入。

磁盘除了访问模式，还有两个低效率操作影响系统的性能：大量的小型I/O操作，过多的字节拷贝。

那么我们怎么处理这些问题呢？

针对于大量的小型I/O操作， Kafka-R 使用“消息块”将消息合理分组。使网络请求将多个消息打包成一组，而不是每次发送一条消息，从而使整组消息分担网络往返的开销。

另一个过多的字节拷贝， Kafka-R 使用producer，broker和consumer都共享的标准化通用的二进制消息格式，这样数据块不用修改就能在他们之间传递。

保持这种通用的格式有什么用呢？

可以对持久化日志块的网络传输进行优化。现代的unix操作系统提供了一个高度优化的编码方式，用于将数据从pagecache转移到socket网络连接中。

数据从文件到套接字的常见数据传输过程：磁盘->pagecache->用户空间缓存区->套接字缓冲区（内核空间）->NIC缓存区

1. 操作系统从磁盘读区数据到内核空间的pagecache

2. 应用程序读取内核空间的数据到用户空间的缓存区

3. 应用程序将数据（用户空间的缓存区）写会内核空间到套接字缓冲区（内核空间）

4. 操作系统将数据从套接字缓冲区（内核空间）复制到能够通过网络发送的NIC缓冲区

共进行了4次copy操作和2次系统调用，显然很低效。在Linux系统中使用zero-copy（零拷贝）优化，其中之一 sendfile ，使用后的数据传输过程是这样：磁盘->pagecache->NIC缓存区。

我们的Kafka-R通过使用zero-copy优化技术，可以用尽可能低的消费代价让多个consumer消费。数据在使用时只会被复制到pagecache中一次，这样消息能够以接近网络连接的速度上限进行消费。

二、数据结构——BTree&日志解决方案

日志解决方案即简单读取与追加来操作文件。

我们的 Kafka-R 采用日志解决方案。

我们知道BTree是通用的数据结构，其广泛用于随机的数据访问。BTree的操作时间复杂度是O(log N)，基本等同于常数时间，但在磁盘上则不成立。

每个磁盘同时只能执行一次寻址，并行性受到限制。少量的磁盘寻址也有很高的开销。数据翻倍时性能下降不止两倍。

而日志解决方案的数据存储架构，所有的操作时间复杂度都是O(1)，并且读不会阻塞写，读之间也不会相互影响。

由于性能和数据的大小是完全分离的，则服务器可以使用大量廉价、低转速的1+TB SATA硬盘，即使这些硬盘的寻址性能很差，在大规模读写的性能也可以接受，而且三分之一的价格三倍的容量。

三、获取数据方式——push-based&pull-based

由consumer从broker那里pull数据呢？还是从broker将数据push到consumer?

我们的 Kafka-R 采用pull-based方式。

这是大多数消息系统所共享的传统的方式：即producer把数据push到broker，然后consumer从broker中pull数据。

push-based系统优点：

1. 让consumer能够以最大速率消费。

push-based系统缺点：

1. 由于broker控制着数据传输速率，所以很难处理不同的consumer。

2. 当消费速率低于生产速率时，consumer往往会不堪重负（本质类似于拒绝服务攻击）。

3. 必须选择立即发送请求或者积累更多的数据，然后在不知道下游的consumer能否立即处理它的情况下发送这些数据。特别系统为低延迟状态下，这样会极度糟糕浪费。

pull-based系统优点：

1. 可以大批量生产要发送给consumer的数据。

pull-based系统缺点：

1. 如果broker中没有数据，consumer可能会在一个紧密的循环中结束轮询，实际上会busy-waiting直到数据到来。

为了避免busy-waiting，我们的 Kafka-R 的pull参数重加入参数，使得consumer在一个“long pull”中阻塞等待，知道数据到来（还可以选择等待给定字节长度的数据来确保传输长度）。

四、消费者的位置——consumed&offset

Kafka-R 的消费过程：consumer通过向broker发出一个“fetch”请求来获取它想要消费的partition。consumer的每个请求在log中指定了对应的offset，并接收从该位置开始的一大块数据。

consumed指通过状态标示已经被消费的数据。

大多数消息系统都在broker上保存被消费消息的元数据。当消息被传递给consumer，broker要么立即在本地记录该事件，要么等待consumer的确认后再记录。

消费者的位置问题其实就是broker和consumer之间被消费数据的一致性问题。如果broker再每条消息被发送到网络的时候，立即将其标记为consumd，那么一旦consumer无法处理该消息（可能由consumer崩溃或者请求超时或者其他原因导致），该消息就会丢失。为了解决消息丢失的问题，许多消息系统增加了确认机制：即当消息被发送出去的时候，消息被标记为sent而不是consumed；然后broker会等待一个来自consumer的特定确认，再将消息标记为consumed。这个策略修复了消息丢失的问题，但也产生了新问题。首先，如果consumer处理了消息但在发送确认之前出错了，那么该消息就会被消费两次。第二个是有关性能的，broker必须为每条消息保存多个状态（首先对其加锁，确保该消息只被发送一次，然后将其永久的标记为consumed，以便将其移除）。还有更棘手的问题，比如如何处理已经发送但一直等不到确认的消息。

Kafka-R 使用offse来处理消息丢失问题。topic被分割成一组完全有序的partition，其中每一个partition在任意给定的时间内只能被每个订阅了这个topic的consumer组中的一个consumer消费。意味着partition中每一个consumer的位置仅仅是一个数字，即下一条要消费的消息的offset。这样就可以按非常低的代价实现和消息确认机制等同的效果。consumer还可以回退到之前的offset再次消费之前的数据，这样的操作违背了队列的基本原则，但事实证明对consumer来说是个很重要的特性。如果consumer代码由bug，并且在bug被发现之前有部分数据被消费了，consumer可以在bug修复后通过回退到之前的offset再次消费这些数据。

五、leader选举——多数投票机制f+1&ISR

Kafka-R 动态维护了一个同步状态的备份的集合（a set of in-sync replicas），简称ISR。

在了解ISR之前我们需要先了解in-sync。

Kafka-R 判断节点是否存活有两种方式：

1. 节点必须可以维护和ZooKeeper的连接，ZooKeeper通过心跳机制检查每个节点的连接。

2. 如果节点是个follower，它必须能及时的同步leader的写操作，并且延时不能太久。

只有满足上面两个条件的节点就处于“in sync”状态。leader会追踪所有“in sync”的节点，如果有节点挂掉了，或是写超时，或是心跳超时，leader就会把它从同步副本列表中移除。

在ISR集合中节点会和leader保持高度一致，只有这个集合的成员才有资格被选举为leader，一条消息必须被这个集合所有节点读取并追加到日志中了，这条消息才能视为提交。

ISR集合发生变化会在ZooKeeper持久化，所以这个集合中的任何一个节点都有资格被选为leader。

多数投票机制f+1顾名思义：假设我们有2f+1个副本，如果在leader宣布消息提交之前必须有f+1个副本收到该消息，并且如果我们从这只少f+1个副本之中，有着最完整的日志记录的follower里来选择一个新的leader，那么在故障数小于f的情况下，选举出的leader保证具有所有提交的消息。

多数投票算法必须处理许多细节，比如精确定义怎样使日志更加完整，确保在leader down期间，保证日志一致性或者副本服务器的副本集改变。

多数投票机制有一个非常好的优点：延迟取决于较快的服务器。也就是说，如果副本数是3，则备份完成的等待时间取决于最快的follwer。

因此提交时能避免最慢的服务器，这也是多数投票机制的优点。

同样多数投票的缺点也很明显，多数的节点挂掉后不能选择出leader。而通过冗余来避免故障率，会降低吞吐量，不利于处理海量数据。

是一种Quorum读写机制（如果选择写入时候需要保证一定数量的副本写入成功，读取时需要保证读取一定数量的副本，读取和写入之间有重叠）。

Kafka-R 保证只要有只少一个同步中的节点存活，提交的消息就不会丢失。

在一次故障生存之后，大多数的quorum需要三个备份节点和一次确认，ISR只需要两个备份节点和一次确认。

创建副本的单位是topic的partition,正常情况下，每个分区都有一个leader和零或多个follower。总的副本数是包括leader与所有follwer的总和。所有的读写操作都由leader处理，一般partition的数量都比broker的数量多的多，各分区的leader均匀分布在broker中。所有的follower节点都同步leader节点的日志，日志中的消息和偏移量都和leader保持一致。

六、Uclean leader选举——ISR副本&第一个副本

如果节点全挂了的服务恢复。

Kafka-R 对于数据不会丢失时基于只少一个节点保持同步状态，而一旦分区上的所有备份节点都挂了，就无法保证了。

Kafka-R 默认“第一个副本”策略。

ISR副本：等待一个ISR的副本重新恢复正常服务，并选择这个副本作为新leader（极大可能拥有全部数据）

第一个副本：选择第一个重新恢复正常服务的副本（不一定是ISR）作为leader。

这是可用性和一致性之间的简单妥协，如果只等待ISR的备份节点，只要ISR备份节点都挂了，那么服务都一直会不可用，如果他们的数据损坏了或者丢失了，那就会是长久的宕机。另一方面，如果不是ISR中的节点恢复服务并且我们允许它成为leader，那么它的数据就是可信的来源，即使它不能保证记录了每一个已经提交的消息。

可以配置属性unclean.leader.election.enable禁用次策略，那么就会使用“ISR副本”策略即停机时间优于不同步，以修改默认配置。

通过以上的架构技术的分析和选型，我们就大致设计出了我们的消息队列 Kafka-R 。