Apache Kafka：下一代分布式消息系统

做者 Abhishek Sharma ，译者 梅雪松
原文地址：http://www.infoq.com/cn/articles/apache-kafka

简介

Apache Kafka是分布式发布-订阅消息系统。它最初由LinkedIn公司开发，以后成为Apache项目的一部分。Kafka是一种快速、可扩展的、设计内在就是分布式的，分区的和可复制的提交日志服务。

Apache Kafka与传统消息系统相比，有如下不一样：

一、它被设计为一个分布式系统，易于向外扩展；
二、它同时为发布和订阅提供高吞吐量；
三、它支持多订阅者，当失败时能自动平衡消费者；
四、它将消息持久化到磁盘，所以可用于批量消费，例如ETL，以及实时应用程序。

本文我将重点介绍Apache Kafka的架构、特性和特色，帮助咱们理解Kafka为什么比传统消息服务更好。

我将比较Kafak和传统消息服务RabbitMQ、Apache ActiveMQ的特色，讨论一些Kafka优于传统消息服务的场景。在最后一节，咱们将探讨一个进行中的示例应用，展现Kafka做为消息服务器的用途。这个示例应用的完整源代码在GitHub。关于它的详细讨论在本文的最后一节。

架构

首先，我介绍一下Kafka的基本概念。它的架构包括如下组件：

一、话题（Topic）是特定类型的消息流。消息是字节的有效负载（Payload），话题是消息的分类名或种子（Feed）名。
二、生产者（Producer）是可以发布消息到话题的任何对象。
三、已发布的消息保存在一组服务器中，它们被称为代理（Broker）或Kafka集群。
四、消费者能够订阅一个或多个话题，并从Broker拉数据，从而消费这些已发布的消息。

图1：Kafka生产者、消费者和代理环境

生产者能够选择本身喜欢的序列化方法对消息内容编码。为了提升效率，生产者能够在一个发布请求中发送一组消息。下面的代码演示了如何建立生产者并发送消息。

生产者示例代码：

为了订阅话题，消费者首先为话题建立一个或多个消息流。发布到该话题的消息将被均衡地分发到这些流。每一个消息流为不断产生的消息提供了迭代接口。而后消费者迭代流中的每一条消息，处理消息的有效负载。与传统迭代器不一样，消息流迭代器永不中止。若是当前没有消息，迭代器将阻塞，直到有新的消息发布到该话题。Kafka同时支持点到点分发模型（Point-to-point delivery model），即多个消费者共同消费队列中某个消息的单个副本，以及发布-订阅模型（Publish-subscribe model），即多个消费者接收本身的消息副本。下面的代码演示了消费者如何使用消息。

消费者示例代码：

Kafka的总体架构如图2所示。由于Kafka内在就是分布式的，一个Kafka集群一般包括多个代理。为了均衡负载，将话题分红多个分区，每一个代理存储一或多个分区。多个生产者和消费者可以同时生产和获取消息。

图2：Kafka架构

Kafka存储

Kafka的存储布局很是简单。话题的每一个分区对应一个逻辑日志。物理上，一个日志为相同大小的一组分段文件。每次生产者发布消息到一个分区，代理就将消息追加到最后一个段文件中。当发布的消息数量达到设定值或者通过必定的时间后，段文件真正写入磁盘中。写入完成后，消息公开给消费者。

与传统的消息系统不一样，Kafka系统中存储的消息没有明确的消息Id。

消息经过日志中的逻辑偏移量来公开。这样就避免了维护配套密集寻址，用于映射消息ID到实际消息地址的随机存取索引结构的开销。消息ID是增量的，但不连续。要计算下一消息的ID，能够在其逻辑偏移的基础上加上当前消息的长度。

消费者始终从特定分区顺序地获取消息，若是消费者知道特定消息的偏移量，也就说明消费者已经消费了以前的全部消息。消费者向代理发出异步拉请求，准备字节缓冲区用于消费。每一个异步拉请求都包含要消费的消息偏移量。Kafka利用sendfile API高效地从代理的日志段文件中分发字节给消费者。

图3：Kafka存储架构

Kafka代理

与其它消息系统不一样，Kafka代理是无状态的。这意味着消费者必须维护已消费的状态信息。这些信息由消费者本身维护，代理彻底无论。这种设计很是微妙，它自己包含了创新。

从代理删除消息变得很棘手，由于代理并不知道消费者是否已经使用了该消息。Kafka创新性地解决了这个问题，它将一个简单的基于时间的SLA应用于保留策略。当消息在代理中超过必定时间后，将会被自动删除。

这种创新设计有很大的好处，消费者能够故意倒回到老的偏移量再次消费数据。这违反了队列的常见约定，但被证实是许多消费者的基本特征。

ZooKeeper与Kafka

考虑一下有多个服务器的分布式系统，每台服务器都负责保存数据，在数据上执行操做。这样的潜在例子包括分布式搜索引擎、分布式构建系统或者已知的系统如Apache Hadoop。全部这些分布式系统的一个常见问题是，你如何在任一时间点肯定哪些服务器活着而且在工做中。最重要的是，当面对这些分布式计算的难题，例如网络失败、带宽限制、可变延迟链接、安全问题以及任何网络环境，甚至跨多个数据中心时可能发生的错误时，你如何可靠地作这些事。这些正是Apache ZooKeeper所关注的问题，它是一个快速、高可用、容错、分布式的协调服务。你可使用ZooKeeper构建可靠的、分布式的数据结构，用于群组成员、领导人选举、协同工做流和配置服务，以及广义的分布式数据结构如锁、队列、屏障（Barrier）和锁存器（Latch）。许多知名且成功的项目依赖于ZooKeeper，其中包括HBase、Hadoop 2.0、Solr Cloud、Neo4J、Apache Blur（Incubating）和Accumulo。

ZooKeeper是一个分布式的、分层级的文件系统，能促进客户端间的松耦合，并提供最终一致的，相似于传统文件系统中文件和目录的Znode视图。它提供了基本的操做，例如建立、删除和检查Znode是否存在。它提供了事件驱动模型，客户端能观察特定Znode的变化，例如现有Znode增长了一个新的子节点。ZooKeeper运行多个ZooKeeper服务器，称为Ensemble，以得到高可用性。每一个服务器都持有分布式文件系统的内存复本，为客户端的读取请求提供服务。

图4：ZooKeeper Ensemble架构

上图4展现了典型的ZooKeeper ensemble，一台服务器做为Leader，其它做为Follower。当Ensemble启动时，先选出Leader，而后全部Follower复制Leader的状态。全部写请求都经过Leader路由，变动会广播给全部Follower。变动广播被称为原子广播。

Kafka中ZooKeeper的用途：正如ZooKeeper用于分布式系统的协调和促进，Kafka使用ZooKeeper也是基于相同的缘由。ZooKeeper用于管理、协调Kafka代理。每一个Kafka代理都经过ZooKeeper协调其它Kafka代理。当Kafka系统中新增了代理或者某个代理故障失效时，ZooKeeper服务将通知生产者和消费者。生产者和消费者据此开始与其它代理协调工做。Kafka总体系统架构如图5所示。

图5：Kafka分布式系统的整体架构

Apache Kafka对比其它消息服务

让咱们了解一下使用Apache Kafka的两个项目，以对比其它消息服务。这两个项目分别是LinkedIn和个人项目：

一、LinkedIn的研究

LinkedIn团队作了个实验研究，对比Kafka与Apache ActiveMQ V5.4和RabbitMQ V2.4的性能。他们使用ActiveMQ默认的消息持久化库Kahadb。LinkedIn在两台Linux机器上运行他们的实验，每台机器的配置为8核2GHz、16GB内存，6个磁盘使用RAID10。两台机器经过1GB网络链接。一台机器做为代理，另外一台做为生产者或者消费者。

二、生产者测试

LinkedIn团队在全部系统中配置代理，异步将消息刷入其持久化库。对每一个系统，运行一个生产者，总共发布1000万条消息，每条消息200字节。Kafka生产者以1和50批量方式发送消息。ActiveMQ和RabbitMQ彷佛没有简单的办法来批量发送消息，LinkedIn假定它的批量值为1。结果以下面的图6所示：

图6：LinkedIn的生产者性能实验结果

Kafka性能要好不少的主要缘由包括：

Kafka不等待代理的确认，以代理能处理的最快速度发送消息。
Kafka有更高效的存储格式。平均而言，Kafka每条消息有9字节的开销，而ActiveMQ有144字节。其缘由是JMS所需的沉重消息头，以及维护各类索引结构的开销。LinkedIn注意到ActiveMQ一个最忙的线程大部分时间都在存取B-Tree以维护消息元数据和状态。

三、消费者测试

为了作消费者测试，LinkedIn使用一个消费者获取总共1000万条消息。LinkedIn让全部系统每次拉请求都预获取大约相同数量的数据，最多1000条消息或者200KB。对ActiveMQ和RabbitMQ，LinkedIn设置消费者确认模型为自动。结果如图7所示。

图7：LinkedIn的消费者性能实验结果

四、Kafka性能要好不少的主要缘由包括：

Kafka有更高效的存储格式；在Kafka中，从代理传输到消费者的字节更少。
ActiveMQ和RabbitMQ两个容器中的代理必须维护每一个消息的传输状态。LinkedIn团队注意到其中一个ActiveMQ线程在测试过程当中，一直在将KahaDB页写入磁盘。与此相反，Kafka代理没有磁盘写入动做。最后，Kafka经过使用sendfile API下降了传输开销。
目前，我正在工做的一个项目提供实时服务，从消息中快速并准确地提取场外交易市场（OTC）订价内容。这是一个很是重要的项目，处理近25种资产类别的财务信息，包括债券、贷款和ABS（资产担保证券）。项目的原始信息来源涵盖了欧洲、北美、加拿大和拉丁美洲的主要金融市场领域。下面是这个项目的一些统计，说明了解决方案中包括高效的分布式消息服务是多么重要：

天天处理的消息数量超过1,300,000；
天天解析的OTC价格数量超过12,000,000；
支持超过25种资产类别；
天天解析的独立票据超过70,000。
消息包含PDF、Word文档、Excel及其它格式。OTC订价也可能要从附件中提取。

因为传统消息服务器的性能限制，当处理大附件时，消息队列变得很是大，咱们的项目面临严重的问题，JMSqueue一天须要启动2-3次。重启JMS队列可能丢失队列中的所有消息。项目须要一个框架，不论解析器（消费者）的行为如何，都可以保住消息。Kafka的特性很是适用于咱们项目的需求。

五、当前项目具有的特性：

使用Fetchmail获取远程邮件消息，而后由Procmail过滤并处理，例如单独分发基于附件的消息。
每条消息从单独的文件获取，该文件被处理（读取和删除）为一条消息插入到消息服务器中。
消息内容从消息服务队列中获取，用于解析和提取信息。

示例应用

这个示例应用是基于我在项目中使用的原始应用修改后的版本。我已经删除日志的使用和多线程特性，使示例应用的工件尽可能简单。示例应用的目的是展现如何使用Kafka生产者和消费者的API。应用包括一个生产者示例（简单的生产者代码，演示Kafka生产者API用法并发布特定话题的消息），消费者示例（简单的消费者代码，用于演示Kafka消费者API的用法）以及消息内容生成API（在特定路径下生成消息内容到文件的API）。下图展现了各组件以及它们与系统中其它组件间的关系。

图8：示例应用组件架构

示例应用的结构与Kafka源代码中的例子程序类似。应用的源代码包含Java源程序文件夹‘src’和’config’文件夹，后者包括几个配置文件和一些Shell脚本，用于执行示例应用。要运行示例应用，请参照ReadMe.md文件或GitHub网站Wiki页面的说明。

程序构建可使用Apache Maven，定制也很容易。若是有人想修改或定制示例应用的代码，有几个Kafka构建脚本已通过修改，可用于从新构建示例应用代码。关于如何定制示例应用的详细描述已经放在项目GitHub的Wiki页面。

如今，让咱们看看示例应用的核心工件。

Kafka生产者代码示例：

上面的代码片段展现了Kafka生产者API的基本用法，例如设置生产者的属性，包括发布哪一个话题的消息，可使用哪一个序列化类以及代理的相关信息。这个类的基本功能是从邮件目录读取邮件消息文件，而后做为消息发布到Kafka代理。目录经过java.nio.WatchService类监视，一旦新的邮件消息Dump到该目录，就会被当即读取并做为消息发布到Kafka代理。

Kafka消费者代码示例：

上面的代码演示了基本的消费者API。正如咱们前面提到的，消费者须要设置消费的消息流。在Run方法中，咱们进行了设置，并在控制台打印收到的消息。在个人项目中，咱们将其输入到解析系统以提取OTC订价。

在当前的质量保证系统中，咱们使用Kafka做为消息服务器用于概念验证（Proof of Concept，POC）项目，它的总体性能优于JMS消息服务。其中一个咱们感到很是兴奋的特性是消息的再消费（re-consumption），这让咱们的解析系统能够按照业务需求从新解析某些消息。基于Kafka这些很好的效果，咱们正计划使用它，而不是用Nagios系统，去作日志聚合与分析。

总结

Kafka是一种处理大量数据的新型系统。Kafka基于拉的消费模型让消费者以本身的速度处理消息。若是处理消息时出现了异常，消费者始终能够选择再消费该消息。