Kafka消息订阅发布系统设计介绍

Kafka学习总结
一、Kafaka简介
Kafka是一个分布式的消息发布-订阅系统。它的特性如下：
l 通过在O(1)的磁盘数据结构上提供消息持久化，对于即使数以TB的消息存储也能够保持长时间的稳定性能。
l 高吞吐：在商用机器上可以提供每秒数十万条的消息
l 支持在Kafaka服务器集群上进行messages分片，并在把messages在消费集群的机器上分配的同时维护每个分片的顺序信息。
l 支持将数据并行的加载到Hadoop中。
Kafka目标是提供一个可以完成在一个用户站点上所有动作流数据和处理的解决方案。
这种动作（网页浏览量、搜索和其他的用户行为）是在现代网站的许多社交特征中的重要组成部分。这个数据通常根据吞吐量的需要采用日志记录和专门的日志合并解决方案。这种专门的解决方案通过提供记录数据给离线分析系统，比如hadoop,是可行的，但是要想实现实时的数据处理就很局限了。Kafka的目标是通过提供一个在并行加载到hadoop的同时也要具有在一个集群上逐个分区的进行实时消费能力的机制来统一离线和线上处理。
针对动作流处理的使用让Kafaka可以与Facebook的Scribe和Cloudera的Flume相提并论，尽管这些系统的体系结构和基本实现单元都不相同，并且使得Kafaka更像是传统的消息处理系统。make Kafka more comparable to a traditional messaging system.
二、设计
1.项目背景和出发点
Kafaka是一个构建在LinkedIn(一个社交网站)作为活动流处理基础的消息系统。
活动流数据是一个分析任何一个网站浏览量的指标。动作的数据是一些比如像网页浏览量、关于什么内容被展示的信息和搜索之类的东西。这种东西通常用日志记录把相关的动作到一类文件中，并最终阶段性的整合这些文件去进行分析。
在近几年中，然而，动作数据已经成为了网站生产特征的要害部位，所以需要一个稍微复杂一些的基础架构集。
2.动态数据的用例
l 广播你朋友的活动的消息推送特点
l 使用计数分级、投票或者点击率关联分析和排序来决定给出的条目集中哪一个最相关
l 安全性：站点需要屏蔽abusive crawlers, rate-limit apis，检测垃圾邮件和维护其他的检测和锁定外界连接的防御系统。
l 运行检测：大部分的站点需要某些实时的、灵活的监控，以便跟踪性能变化并在出错的情况下给出警告。
l 报表和批处理：将数据加载到数据仓库或者hadoop系统去进行离线分析和就商业活动作出报告
3.动作流数据的特点
因为动作流数据的体积几乎10倍甚至100倍于一个站点的下一个数据源，所以这个高吞吐量的不可变动态数据真的提出了一个巨大的计算挑战。
传统的日志文件聚合计算是一个不错的和可扩展的方法来提供类似于报表或者批处理方面的用例；但是对于实时处理它有太大的时延并且倾向于具有相当高复杂度的操作。另一方面，已存在的消息和队列系统对于实时或者准实时用例是不错的，但是在操作大量未被消费掉的队列方面很差劲，often treating persistence as an after thought。这就给向馈入hadoop一样的离线系统带来了麻烦，可能每天或者每小时只能消费一些数据源。This creates problems for feeding the data to offline systems like Hadoop that may only consume some sources once per hour or per day.Kafaka意在作一个同时支持离线和线上用例的单一队列平台。
Kafka支持相当通用的信息语义。但是没有一个与动作处理绑定，尽管那是激发我们做Kafka的用例。
4.部署
下面的这副图给出了Kafka在LinkedIn的一个简化的部署拓扑图。

注意，一个kafka集群处理来自所有不同数据源的所有动作数据。这给线上和离线处理的consumers提供一个单一的数据管道。这一层作为一个在实时活动和异步处理之间的一个缓存。我们也使用kafka来将所有的数据备份到另外一个数据中心去进行离线处理。
5．主要设计元素
这里有一些使得Kafka不同于大部分其他的消息系统的主要的设计方案：
1. Kafka 被设计成把messages持久化作为一种常见的情况。
2. 吞吐量而不是特性作为最基本的设计限制
3. 关于消息被处理的状态作为消费系统的一部分在维护而不是在kafka的服务器上。
State about what has been consumed is maintained as part of the consumer not the server
4. Kafka是显式的分布式系统。它被假设为producer、broker和consumer都分布于多台机器上
以上的每一种设计都将在下面详细讨论。
6.基本概念
首先是一些基本的术语和概念：
Messages 是通信的基本单位。Messages被一个producer发布给一个topic，也就是说它们被物理地传递给作为一个broker的服务器(一般是另外一台机器)。一些consumer订阅一个topic，每一个被发布的消息就被转交给这些consumer了。
Kafka是显式分布式的——producer、consumer和broker可以全部运行在一个集群 的机器上，并作为一个逻辑组进行协作运行。这对于broker和producer是非常自然的，但是consumer需要一些特别的支持。每一个consumer进程属于一个consumer group并且每一条message恰是被传递给每一个consumer group中的一个进程。因此一个consumer group允许许多进程或者机器逻辑地作为一个单一的consumer。这种consumer group的概念是非常有用的，可以被用做支持在JMS中提供的queue或者是topic的语义。要支持queue的语义，我们将所有的consumer放到一个consumer group里面，在这种情况下每一个message将传递到一个单独的consumer中。要支持topic的语义分析，每一个consumer被放到它自己的consumer group中，接着所有的consumer将接到每一条消息。在我们使用中更通用的一个用例是我们有多个逻辑consumer group，每一个由作为一个逻辑整体的consuming machines组成。Kafka在大规模数据面前有更多的优势，因为不管一个topic有多少consumer，一个消息只存储一次。
7．消息持久化和缓存
不要害怕文件系统
Kafka严重依赖于文件系统去进行messages的存储和缓存。有一个常见的观点就是磁盘是缓慢的，也使人们怀疑在磁盘上进行持久化的结构可以提供比较好的性能。事实上，磁盘或者比人们想象中的更慢或者更快，这取决于它们是被怎样使用的，一个合适的磁盘结构设计通常比网络传输还快。
关于磁盘性能的实际情况是在过去的十年中磁盘的吞吐量已经和磁盘的寻道时间不一致了。结果是在6 7200rpm SATA RAID-5 组上的一次写入性能可以达到 300MB/sec，但是随机写入的性能确是50k/sec——几乎是1000被的差距。这些一次读写在大部分的使用情况中是可以预计到的，因此这可以用过使用read-ahead和write-behind技术，就是在一个大的数据块中预先多次取出数据并把较小的逻辑写入合并到一个大的物理写入，把磁盘性能最佳化。关于这个问题更深入的讨论请参考ACM Queue article，他们事实上发现连续的磁盘访问在一些情况下比对内存的随机读取还要快。
为了祢补这种性能不足，现代操作系统已经倾向于使用主存作为磁盘的缓存。任何的现代操作系统将所有的空余内存作为磁盘的缓存，同时在内存回收的时候伴有一些性能损失。所有的磁盘读写将通过这个统一的缓存。这种特性如果不是直接使用I/O的话不能被轻易的关闭。所以即使是一个在进程中保存的数据，还是可能被复制到OS的页缓存中，事实上把所有的数据都存储两次。
另外我们是建立在JVM之上的，所有花了一些时间在Java内存使用的人都知道两件事：
1. 对象的内存开销特别高，通常会是所存储数据的两倍。
2. Java的内存回收机制随着堆内存的数据的增加变得频繁
作为这些因素的结果，使用文件系统和依赖于页缓存比维持一个内存的存储或者其他的结构有优势——我们至少通过自动访问所有的空闲内存使得可用的缓存加倍，而且可能通过存储一个紧凑的字节结构而不是单独的对象使得可用的缓存又增加一倍的大小。这么做将导致在在一个32GB的机器上有28到30GB的缓存，而且还不会有GC带来的损失。而且这种缓存将保持可用即使服务被重新启动，但是进程中的缓存将需要在内存中重建（这对于一个10GB的缓存将需要大概10分钟的时间）或者需要用一个完全的冷备份启动（这将是一个非常可怕的初始化过程）。它也将极大的简化了编码因为所有在缓存和文件系统里的相关维护逻辑现在都归操作系统里了，这将比在进程中的一次性尝试的效率和正确度都要高。如果你的磁盘支持一次的读取那么read-ahead 将有效地用每一个磁盘上读取的有用数据填充这个缓存。
这表明了一种很简单的设计：我们不是把数据尽量多的维持在内存中并只有当需要的时候在将数据刷到文件系统，我们是反其道而行之。所有的数据不用进行任何的刷数据的调用就立刻被写入到文件系统的一个持久化的日志记录中。事实上这只是意味着转移到了内核的页缓存中，OS将在之后将它刷出。接着我们添加一个配置驱动器刷数据策略来允许系统的用户控制数据被刷入物理磁盘的频率（每多少消息或者每多少秒）来设置一个在临界磁盘崩溃时数据量的一个限制。
这种页缓存为中心的设计在一片关于Varnish的设计的文章中有描述。
8.较长时间的满足
在消息系统元数据中使用的持久化数据结构的通常是B树。B树是可以使用的最万能的数据结构，使得在消息系统中支持一个广泛的各种各样的事务性的和非事务性的语义。这样有着相当高的开销，但是B树操作的复杂度是O(log N)。正常情况下O(log N)被认为是本质上等于恒定时间，但是这对于磁盘操作来讲是不对的。磁盘寻道达到10ms ,并且一个磁盘一次只能做一个寻道，所有并行化被限制住了。因此即使是少量的磁盘寻道也将带来很大的开销。因为存储系统在物理磁盘操作中混合了快速的缓存操作，观测到的树结构的性能通常是字面上的。此外，B树要求一个复杂的页或行同步实现来避免在每一个操作中锁定整个树。实现必须要对row-locking或者有效的序列化所有的读取付出较高的代价。因为对磁盘寻道较高的依赖，不可能有效的利用驱动器的密度优势。人们被强制使用小型的（小于100GB）高转速SAS驱动器来维持一个数据寻找能力的合理度。
直观上，一个持久化队列可以建立在简单地读取和添加到文件中，就像在日志记录方案中常见到的那样。尽管这个结构不能支持丰富的B树实现的语义，但是他有一个优势就是所有的操作复杂度都是O(1)并且读和写之间不会互相阻塞。这显然是个性能优势自从性能完全和数据的大小解耦。一个服务器可以充分利用许多廉价的、低转速的1TB大小的SATA驱动器。虽然它们的寻道性能比较差，但是这些驱动器对于大的写入和读取方面有1/3的价格和3倍的容量的性能可比性。可以访问几乎没有限制的磁盘空间而不出现损失意味着我们可以提供一些在消息系统中不常见的特性。比如，在kafka中，我们可以将消息保存相当长的一段时间而不是在消费完之后立刻将它删除。
9.效率最大化
我们的假设是message量是及其之大的，甚至是一个站点的网页访问量（我们假设网页访问量是我们要处理的活动）总数的几倍。此外，我们还假设每一个发布的消息至少被读取一次（实际上可能是很多次），因此我们为consumption优化而不是为production优化。
这里有两个效率低下的原因：太多的网络请求有和过度的字节复制
为了达到高效率，API建立在一个“message set”的抽象上，这个抽象自然的将消息分组。这使得网络请求把message分组以此将网络折返的开销分摊而不是一次只请求一个消息。
MessageSet实现本身是一个小的封装了字节数组或者文件的API。因此在message处理中没有单独的序列化和反序列化的步骤，message字段是根据需要lazily deserialized的。
被broker维护的message的记录本身只是个被写入磁盘的message sets的目录。这种抽象使得一个字节格式同时被broker和consumer共享（某种程度上还有producer，虽然producer的消息在被记录之前已经被校验和验证过了）。
维护这个通用的格式允许对最重要的操作进行优化：持久化的日志块在网络中的传输。现代的Unix操作系统提供了高度优化过的编码途径将数据从页缓存中传输给一个socket。在Linux中这通过sendfile的系统调用实现。Java通过FileChannel.transferTo api 提供了对这个系统调用的访问。
要想了解sendfile的影响，首先理解通常的数据从文件传输到socket的途径是很重要的。
1. 操作系统从磁盘中读取数据到内核空间的页缓存中
2. 应用程序从内核空间将数据读取到用户空间缓存
3. 应用程序将数据写回到内核空间的一个socket缓存
4. 操作系统从socket缓存将数据复制到网卡缓存，并最终发送到网络中
这是非常低效的，这里出现了四次拷贝，两次系统调用。使用sendfile,通过允许OS将数据直接从页缓存发送到网络来避免这种重复复制。所有在这种途径中，只有最终复制到NIC的缓存中是必须的。
我们希望对一个topic对应的多个consumer有一个通用的用例。使用了上面的0拷贝优化，数据被拷贝到页缓存中一次而不是存在内存中并被每一个consumption重复利用，在每次读取的时候拷贝出内核空间。这就使得消息被以接近网络连接上限的速率consumed。
对于更多在java中对sendfile和0拷贝的支持，请参考这篇文章。
10.消费状态
跟踪什么被消费是一个消息系统必须提供的主要功能之一。这个不是凭直觉的，但是记录这个状态是这个系统主要的性能点之一。状态跟踪需要更新一个持久化的条目并且可能引起随机的读取。因此它可能被存储系统的寻道时间限制而不是写入的带宽。
大部分消息系统在broker上保存哪些消息被消费掉的元数据。也就是说，随着一个消息被传送给consumer，broker进行本地地记录。这是一个相当凭感觉的选择，而且对于单一的服务器来讲的确不清楚它还将传递到哪儿。因为在很多消息系统中用于存储的数据结构扩展性很差，这也是个实用的选择——因为broker知道什么被消费掉了就可以立刻删除它，以保持较小的数据量。
可能不是很明显的是让broker和consumer关于什么已经被消费掉达成协议是一个不简单的问题。如果broker每一次在消息被传送到网络中的时候立刻将它记录为已经consumed，那么如果consumer处理消息失败（可能是因为宕机或者超过请求的时间等等）接着这条消息就会丢失。要解决这个问题，许多消息系统添加了一个确认特性也就是说当消息被传送出去的时候只是被标记为sent而不是consumed;broker等到收到consumer发来的特定的确认之后才把这个消息记录为consumed。这种策略解决了消息丢失的问题，但是带来了新的问题。首先，如果consumer处理了这个消息，但是在发送确认信息时失败了，那么这条信息将被处理两次。第二个问题就是关于性能，现在broker必须保存关于一个消息的多个状态（首先锁定它这样就不会重复发出两次，接着再将它永久的标记为consumed以便于将它删除）。更棘手的问题是怎样处理那种消息被发出去了但是一直没有被确认已经消费掉了的情况。
11消息传递语义
很明显这里有许多种可能的消息交付保障方案来提供：
l 至多传递一次：这是用于上面描述的第一种方案。消息被立刻标记为consumed,这样它们就不会给出两次，但是会出现丢失消息的情况。
l 最少传递一次：这是上面的第二种情况，我们保证每一个消息至少被传递一次，但是在失败的情况下可能被传递两次。
l 准确的传递一次：这是人们真正想要的，每个消息传递一次且仅此一次。
这个问题已经被深入的研究过了，是"transaction commit"问题的变种。提供准确的一次的算法是存在的，two- or three-phase commits和Paxos算法变种就是示例，但是他们带有一些缺点。他们通常需要多次的往返并且可能实施有效性缺乏保证（会不确定的停止）。FLP结果在这些算法中提供了一些基本的限制。
Kafka关于元数据做了两件不同寻常的事。首先是数据流在broker上分片成不同的分片集。这些分片的语义含义被留给producer，producer指定一个消息属于那个分片。在一个分片中消息被按照到达broker的顺序排序，接着按照相同的顺序发送到consumer。这意味着不是存储每一个消息的元数据（比如将它标记为consumed），我们只需要存储对每一个consumer、topic和分片的高水位标志。因此概括consumer状态需要的元数据实际上非常的小。在Kafka中我们将这个高水位线作为偏移量，至于原因在下面的实现部分就会清楚了。
12.consumer状态
Kafka也维护着关于什么被消费给客户端的状态。这为一些简单的用例提供了便利，在一些方面有好处。在最简单的用例是consumer将一些聚合的值录入到一个集中的事务型在线交易处理数据库。在这个用例中consumer可以存储在和数据库修改相同的事务中什么被消费的状态。这解决了分布式的一致性问题，通过移出分布式的部分。一个巧妙的技术也可以对一些非事务的系统有效。一个搜索系统可以用索引分段保存它的consumer状态。虽然它提供了不持久的保证，这意味着索引总是与consumer状态保持一致：如果一个未刷出的索引分段在一次崩溃中丢失，索引将总是从上一个检查点的偏移量开始重新消费。同样的，我们用来冲Kafka中向Hadoop并行加载数据的作业，就做了这样一个技巧。单个的mapper在map任务结束的时候把上一次消费掉的消息的偏移量写入HDFS。如果一个作业失败了并且重新运行，每个mapper仅仅是从存储在HDFS中的偏移量的位置重新开始。
这个决定有一方面的好处。一个consumer可以故意的倒回到一个旧的偏移量并且重新消费数据。这违反了一个队列的常规约定，但却是许多的consumer中必须的一个特性。比如说，如果consumer的代码有问题，而且这是在一些消息已经被处理的之后发现的，这些consumer可以在bug修复以后重新消费那些消息。
13.推还是拉
一个相关问题是是consumer从broker上拉去数据还是broker将数据退送给订阅者。在这一方面，Kafka延续了一个被大部分的消息处理系统采用的传统的设计，也就是数据从producer推送到broker，接着consumer在从broker上拉取数据。许多近期的系统中，比如是scribe和flume,将重点放在日志的整合上，采用了另一种基于路径的推送，这种情况下，每一个节点充当一个broker而数据被顺流而下的推送。这两种方法有利有弊。但是一个基于推送的系统在处理不同的consumer方面有困难，因为broker控制着数据以多达的速率被传送。对于consumer的目标是可以在最大的速率下进行消费；不幸的是在一个推送系统中这意味着当消费的速率低于生产(本质上，就是拒绝服务攻击)的速率的时候consumer将会招架不住。一个基于拉取的系统有一个非常好的性质就是consumer可以简单的落后并在可以的时候再跟上。这可以通过一些退避协议来缓和，通过退避协议consumer可以指示它已经过载了，但是使得传输的速率充分的利用consumer要比想象的难处理。之前用这种方式构建系统的尝试引导我们采用一种更为传统的拉取模型。
14.分布
Kafka通常运行在一个集群的机器上。Broker和consumer通过Zookeeper协调发现topics并且协调消费。这里没有中心的主节点，取而代之的是broker和consumer作为对等节点集中的元素互相配合。集群中的机器集是非常灵活的：broker和consumer都可以在任何时候不用任何手动配置的增加和删除。
目前，在Kafka中的producer和broker之间没有内建的负载均衡；in our own usage we publish from a large number of heterogeneous machines and so it is desirable that the publisher not need any explicit knowledge of the cluster topology。我们依赖于一个硬件负载均衡器在多个broker中分配producer的负载。我们将考虑在将来的版本中添加这一功能来支持消息基于语义的分片（比如将基于某种ID的所有的消息发布到一个特定的broker上以保证在这个ID中更新流的顺序）。
Kafka在consumer和broker之间有一个内建的负载均衡器。要达到这种配合，每一个broker和consumer都要在Zookeeper中注册它的状态并且保存它的元数据。 当这里有一个broker和consumer的改变的时候，Zookeeper的观察者将通知每一个consumer。 接着consumer读取所有当前关于相关的broker和consumer的信息，并决定应该消费来自拿个broker的数据。
这种集群感知的消费平衡有一些优点：
l 它允许在consumer进程排序方面更好的语义支持（从此所有的对于一个特定分片的更新将作为一个单独的流被按顺序处理）
l 它也强制在集群中公平负载这样每一个broker都将被用来消费
l 最后，因为进程之间不会协调除了当一个新的broker或者consumer出现的时候，它可以更加的有效。不是在每一个请求的时候在分片上加锁和解锁（这可能比想象中的还耗费资源），我们仅仅是将一个分片锁定给一个特定的consumer进程直到网络拓扑发生变化。这使得用一个在元数据中懒惰的更新换取了更好的性能。
15.Producer
自动的均衡负载：
在0.6版本中，我们引进了一个内建的在producer和broker之间自动均衡负载器；in our own usage we publish from a large number of heterogeneous machines and so it is desirable that the publisher not need any explicit knowledge of the cluster topology。我们依赖于一个硬件负载均衡器在多个broker中分配producer的负载。使用这个硬件负载均衡器的优势是“healthcheck”服务，这个服务检测到如果一个broker挂了就将producer请求转给另一个正常的broker。在0.6版本中，这个“healthcheck”功能是由集群感知中的producer提供。Producer发现集群中可用的broker和它们每一个上面的分片数目，通过在zookeeper中注册观察者。
因为broker中分片的数目是对每一个topic都是你可配置的，Zookeeper的观察者在以下的事件中进行注册：
l 新的broker出现
l 一个broker删除
l 新的topic被注册
l 一个broker被一个已存在topic注册
内部的，producer维持了一个灵活的到broker的连接池，一个连接到一个
broker。这个连接池保持更新建立和维护到所有的活动的broker，通过zookeeper的回调信号。当一个producer对一个特定的topic的请求到来，一个broker的分片被选出通过partitioner。连接池中可用的producer连接被用来发送数据到选定的broker分片。
16.异步传送
异步非阻塞操作是可扩展的消息系统的基石。在kafka中，producer提供了一个选项来对produce的请求进行异步的分配(producer.type=async)。这允许在一个内存的队列中缓存produce的请求并通过一个时间间隔或者预配置的批量大小来触发进行成批的发送。因为数据通常来自以不同速率产生数据的异源的机器集中，这个异步的缓存帮助生成了到达broker中的统一的通信，以达到更好的网络利用率和更高的吞吐量。
17.语义分片
考虑这样一个应用，用于统计每个成员的网友访问量。这将首先将所有对一个成员的访问事件发送到一个特定的分片，因此使得所有关于这个成员的更新出现在相同consumer线程中的相同的流中。在0.6版本中，我们给集群感知producer添加了可以从语义上将消息映射到可用的kafka节点和分片中。这许可用一些语义上的分片函数基于消息中的某些key对消息流进行分片并在broker机器上传播。那些分片函数可以通过实现kafka.producer.Partitioner 接口就可以了，默认的是一个随机的partitioner 。比如说上面的例子，key如果是member_id 的话，那么分片函数将会是hash(member_id)%num_partitions。
18．对Hadoop和其他批量数据加载的支持
可扩展的持久化考虑到了对批数据加载支持的可能性，比如周期性的将快照存入到一个离线系统做批量处理。我们使用这个将数据加载到我们的数据仓库和hadoop集群。 批处理发生在开始加载数据的阶段和进行无循环图处理并输出的阶段(e.g. as supported here)。支持这个模型的一个基本特性就是可以从一个时间点重新加载数据，以防止出现一些错误。
在Hadoop的案例中，我们通过将装载量分割给单独的map任务，一个对应一个node/topic/partition 组合，实现在加载过程中的完全并行化。Hadoop提供了任务管理，如果任务失败会重新启动而且不会出现重复的数据。