可思数据-人工智能媒体资讯平台！

1. 前言

Apache Kafka 发展至今，已经是一个很成熟的消息队列组件了，也是大数据生态圈中不可或缺的一员。Apache Kafka 社区非常的活跃，通过社区成员不断的贡献代码和迭代项目，使得 Apache Kafka 功能越发丰富、性能越发稳定，截止本篇文章 Apache Kafka 发布了 V2.2.0 版本。

那么，今天就来聊一聊 Kafka 应用实践与生态集成的相关内容。

2. 如何知道 Kafka 是否适合你?

项目立项时，会做技术调研，那么如何知道你选择的 Kafka 技术是否能够满足你?据 Confluent 公司调研报告可知，Apache Kafka 在过去几年中在功能和覆盖范围方面取得了很大成就。它被财富 500 强中的三分之一用于生产，包括全球十大银行中的七家，十大保险公司中的八家，以及美国十大电信公司中的九家。接下来，为大家介绍 Kafka 示例来帮助大家了解常见的使用模式。并且希望大家能找到与自己的工作流程有交集的地方，这样大家就可以开始利用 Kafka 的强大功能了。

下面让先来看看 Kafka 提供的两个核心功能：

2.1 消息系统

消息系统常见的两种模式：

队列：队列消费者充当了工作组的角色，每条消息记录只传递给一个工作进程，从而有效的划分工作流程;

发布与订阅：订阅者通常是彼此独立的，每个订阅者都可以获得每条消息的副本。

这两种模式都是有效和实用的，通过队列将工作内容分开，用于容错和扩展;发布与订阅能够允许多租户，来使系统解耦。而 Apache Kafka 的有点之一在于它将队列、发布与订阅结合到了一个强大的消息系统中。

2.2 流处理

Apache Kafka 拥有强大，可扩展的消息系统，只需要一种简单的方法来处理消息流。而在 Kafka 中，Stream API 提供这一功能，它是一个 Java 客户端类库，提供比 Producer 和 Consumer 更高级别的抽象 API 。

这使得它使用起来非常的方便：

无状态操作，例如过滤和转换流消息;

有状态操作，例如时间窗口上的连接和聚合。

Stream API 处理消息的序列化与反序列化，同时维护有状态操作所需要的状态。

2.3 典型的 Kafka 案例

旅游行业：例如，在一个旅游网站，酒店和航班的价格是一直在变化的，系统的一些组件 ( 价格告警、分析等 ) 需要了解这些变化。你在 Kafka 的 Topic 上发布更改，并且需要通知的每个组件都充当一个消费者。每个消费者应用所组成的节点形成一个消费者组。给消费者组所消费的 Topic 的发送消息动态记录，这样每个消费者均可获取消息记录，同时每个消费者内能够有效的划分工作内容。

用户分析：页面查看、搜索、用户行为分析等，这些实际上是 Kafka 在 LinkedIn 设计的原始初衷。用户点击网站活动内容，每个活动类型均有一个 Topic，可以实时的反馈，以便深入了解用户参与度、下载量、页面流量等。

GPS：例如，能够实时获取智能手机设备的位置数据，并且希望能够实时处理这些数据来显示车辆路径、行驶距离等。传入数据到 Kafka 的 Topic 中，并使用 Stream API 来进行处理。当需要在特定时间段内提取和处理给定用户的所有位置数据时，使用窗口进行状态处理会有不错的效果。

3. Kafka 的内部存储工作原理是什么?

如何你确定了 Kafka 技术适合你当前的项目，满足你的业务需求。你可能会很好奇，Kafka 的内部存储工作原理是什么呢?接下来，将给大家分析 Kafka 是如何存储其数据的。

3.1 Kafka 存储单元是分区

Topic 中的分区是有序写入的，且不可变的消息序列。分区不能跨多个 Broker 或者多个磁盘来进行分割。

3.2 保留策略来管理 Topic 中消息

在你创建的 Topic 中，你可以指定保留多少数据或者保留多长时间的数据，然后 Kafka 会按照顺序来清除这些消息 ( 不管消息是否有被使用 ) 。

3.3 分区片段

Kafka 需要定期查找磁盘上待清除的数据，对于分区消息单个非常长的文件，该操作会很慢并且容易出错。为了解决这个问题，Kafka 实行了分区分片策略。当 Kafka 将消息写入分区时，它会写入到一个片段，如果该片段到达阀值，则会新开一个新的片段来写入。片段以偏移量来命名，片段的偏移量是大于前一个片段的偏移量且小于或者等于当前片段中的偏移量。

3.4 片段日志是存储消息的位置

每条消息都包含值、偏移量、时间戳、主键 ( KEY ) 、消息大小、压缩编解码器、校验、以及消息格式的版本。磁盘上的数据格式与 Broker 通过网络从 Producer 端接收的格式完全相同，然后由 Consumer 去获取数据，这使得 Kafka 能够通过零拷贝技术有效的传输数据。

3.5 片段索引将消息偏移量映射到它们在日志中的位置

索引文件是内存映射的，偏移量查找时使用二进制搜索来查找小于或等于最近的目标偏移量。索引文件由 8 个字节组成，4 个字节用于存储基本偏移量，另外 4 个字节来存储位置。

3.6 Kafka 将压缩的消息包装在一起

发送压缩消息的 Producer 端会将压缩批处理，并将其作为包装消息的有效负载发送。和之前一样，磁盘上的数据与 Broker 通过网络从 Producer 端接收并发送给其 Consumer 的数据完全相同。

3.7 Kafka 内部存储工作原理小结

Kafka 的存储单元是分区;

分区通过片段来进行分割;

片段包含两个文件：索引和日志文件;

索引将每个偏移量映射到它们所在日志中的消息位置，用于查找消息;

压缩消息批处理作为包装消息的有效负载;

存储在磁盘上的数据与 Broker 通过网络从 Producer 端接收并发给 Consumer 的数据相同。

4. Kafka API 之间的选择与竞争

Kafka 的核心尽管在一段时间内保持相对的稳定，但是 Kafka 生态圈仍然在快速的发展。最初的 Kafka，包含 Producer 和 Consumer，很容易理解。现在 Kafka 处理 Producer 和 Consumer，还有 Kafka Connect 、Kafka Streams 、以及 KSQL 。

4.1 如何正确的选择 Kafka API

Kafka Producer API：应用直接生成数据，例如移动设备、PC 、其他硬件等。

Kafka Connect Source API：应用程序桥接在我们无法控制的数据存储介质，例如 MongoDB 、ElasticSearch 、RESTAPI 等。

Kafka Streams API/KSQL：如果希望像 SQL 一样操作实时流数据，可以通过 KSQL 来完成;如果需要编写复杂的业务逻辑，可以使用 Kafka Streams API 来完成。

Kafka Consumer API：直接读取流数据，并对其执行实时操作，例如推送商品促销活动、发送邮件、获取游戏行为等。

Kafka Connect Sink API：读取实时流并将其存储到目标介质中，例如 Kafka 到 S3、Kafka 到 HDFS、Kafka 到 HBase 等。

选择不同的 API 来实现不同的业务需求，例如，如果希望为实现的需求编写大量的自定义代码，Kafka Consumer API 和 Kafka Connect Sink API 完全是可以互换的。总而言之，上述 API 可以帮助你在实际的业务中以最少的代码量来实现最有效的工作流程。

4.2 各个 API 的优势和局限

4.2.1 Kafka Producer API

优势： Kafka Producer API 使用起来非常的简单，异步发送数据，获取回调结果。非常适合直接发送数据流的应用程序，例如日志、点击流、物联网等。

局限：可以扩展和构建 Kafka Producer API 以便执行更多的操作，但是这需要开发人员编写更多的附加逻辑。例如，试图使用 Kafka Producer API 在数据库和 Kafka 之间执行 ETL 操作时，如何跟踪偏移量 ( 即当 Producer 端停止后，如何正确恢复你的 Producer 应用程序 ) ?如何在若干个 Producer 之间分配 ETL 的负载?这种情况，我们使用 Kafka Connect Source API 会更好一些。

4.2.2 Kafka Connect Source API

优势：Kafka Connect Source API 是一个构建在 Kafka Producer API 之上的完整框架。它的构建是为了让开发人员能够获得更好的 API，以便为并行处理生成并分配任务。另外，可以使用各种各样的连接器，利用这些连接器来处理大多数数据介质，且无需编写任何代码。

局限：适配的数据源连接器均是专属的，如果你当前的数据源在已有的连接器中不包含，需要自行编写连接器来进行适配。

4.2.3 Kafka Consumer API

优势：Kafka Consumer API 非常简单，可以使用 Consumer Groups，因此可以并行使用 Topic 。新版本的 Kafka ( V2.2.0 ) 对于偏移量的管理和提交、Balance 、幂等性等无需开发者关心。

局限：在 ETL 场景中，Kafka Connect Sink 更加合适，因为它们会避免针对外部数据源编写复杂的逻辑。

4.2.4 Kafka Connect Sink API

优势：与 Kafka Connect Source API 类似，Kafka Connect Sink API 允许利用现有的 Kafka 连接器的生态系统来执行流式 ETL，且无需编写任何代码。Kafka Connect Sink API 建立在 Kafka Consumer API 的基础之上，但是与它有所不同。

局限：如果写入的数据源没有可用的适配器，那么需要自行编写 Kafka Connect 连接器，并且调试过程会有些复杂。

4.2.5 Kafka Streams API

优势：对于流处理场景，Kafka 中附带 Kafka Streams API，并且能够编写高级 DSL ( 类似于函数式编程或者 Spark 类型的程序 ) 或偏底层的 API ( 类似于 Storm ) 。Kafka Streams API 完全隐藏了 Producer 和 Consumer 的复杂性，让开发者更加专注于流处理的逻辑实现上。同时，它还具有连接、聚合、一次性处理等功能。

局限：使用 Kafka Streams API 会让编码门槛提高，同时也可能让你业务逻辑变得复杂。

4.2.6 KSQL

优势：KSQL 不是 Kafka API 的直接组成部分，而是 Kafka Streams 之上的包装器。这里还是值得一说的，虽然 Kafka Streams 允许编写一些复杂的 Topology，但它还是需要一些实质性的编程知识，尤其是新手来说。KSQL 希望通过提供与现有的 SQL 语义类似来抽象出这种复杂性。对于开发者来说，KSQL 是非常具有诱惑力的，它使得流处理器变得轻而易举。

局限：对于复杂的业务场景，对数据进行复杂的转换操作，或一些特定的需求，可能还是需要使用 Kafka Streams 来完成。

5. Kafka 与 Kubernetes 结合是否效率更高?

5.1 介绍

Kubernetes 是 Google 开源的一个容器编排引擎，它支持自动化部署、大规模可伸缩、应用容器化管理。Kubernetes 旨在运行无状态工作负载，这些工作负载通常采用微服务架构形式，轻量级、水平扩展。而 Kafka 的本质上是一个分布式的存储介质，这意味着你在使用时必需处理状态，它比微服务更重要。尽管 Kubernetes 支持有状态的工作负载，但还是需要谨慎使用。

那么，应该在 Kubernetes 上运行 Kafka 吗?如何让 Kafka 和 Kubernetes 互相补充，以及如何避免可能遇到的“坑”?

5.2 基础指标

进程：Kafka Broker 对 CPU 很友好，TLS 的引入可能会产生一些开销。Kafka Client 如果使用加密会需要更多的 CPU，但是这并不会影响 Broker 。

内存：Kafka Broker 的 JVM 通常可以设置为 4GB-8GB 之间，但是由于 Kafka 大量使用了页面缓存，因此还是需要有足够的系统内存。在 Kubernetes 中，相应的设置容器资源限制和请求。

存储：容器中的存储是暂时的，重启后数据将会丢失，但是可以对 Kafka 数据使用空目录卷。因此，需要使用持久化存储，存储必须是非本地的，以便 Kubernetes 在重启后或重新定位后更加灵活的选择另一个节点。

网络：与大多数分布式系统一样，Kafka 性能在很大程度上取决于低网络延迟和高带宽。建议不要把所有的 Broker 放在同一个节点，因为这样会降低可用性。如果 Kubernetes 节点出现故障，那么整个 Kafka 集群都会出现故障。

5.3 性能

安装 Kafka 之前，做 POC 测试是非常重要的。这样做的好处是，在遇到有关性能瓶颈问题时，可以提供帮助。而 Kafka 附带了两个 POC 测试工具，它们分别是：kafka-producer-perf-test.sh 和 kafka-consumer-perf-test.sh 。

监控：监控 Kafka 指标是非常有必要的，能够让我们及时的掌握 Kafka 、Zookeeper 集群的健康状态，例如使用 Kafka Eagle 来监控和管理 Kafka Topic ( http://www.kafka-eagle.org/ ) 。

日志：日志是一个比较关键的部分，确保 Kafka 安装中所有的容器都记录到 stdout 和 stderr 中，并确保 Kubernetes 集群日志能集中管理，例如输送到 ElasticSearch 。

动态更新：StatefulSets 支持自动更新，RollingUpdate 策略将一次更新一个 Kafka Pod，来实现零停机维护，这也是 Kubernetes 的优势之一。

扩容：Kubernetes 可以很容易的将 Pod 缩放到一定数量的副本，这意味着可以声明性的定义所需数量的 Kafka Broker 。

备份 & 还原：Kafka 部署在 Kubernetes 中，这样 Kafka 的可用性就取决于 Kubernetes 的可用性，如果 Kubernetes 集群出现故障，那么 Kafka 的可用性就会下降，同时，也会出现数据丢失的风险，因此需要做好数据备份策略，例如 MirrorMaker，或是 S3 进行连接备份。

5.4 对于 Kubernetes 的选择

对于中小型的 Kafka 集群，将 Kafka 部署到 Kubernetes 是一个不错的选择，因为它提供了更大的灵活性、且简化了操作。如果在延迟或吞吐量方面有较高的功能性要求，独立部署的方式可能会更好。

6. 总结

本篇文章，介绍了 Kafka 应用实践与生态集成，通过参考本篇文章的内容，大家可以做出合理、有效的选择。