如何使用Scala开发Apache Kafka

本篇内容介绍了“如何使用Scala开发Apache Kafka”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

Apache Kafka是一个广受欢迎的分布式流媒体平台，New Relic、Uber以及Square等数千家公司都在使用它构建可扩展、高吞吐量、可靠的实时流媒体系统。例如，New Relic的Kafka集群每秒处理超过1500万条消息，总数据速率接近1 Tbps。

Kafka在应用程序开发人员和数据科学家中非常受欢迎，因为它极大简化了数据流的处理过程。但是，Kafka在Scala上实践会比较复杂。如果消费者无法跟上数据流，并且消息在他们看到之前就消失了，那么具有自动数据保留限制的高吞吐量发布/订阅模式并没有多大用。同样，如果托管数据流的系统无法扩展以满足需求或者不可靠，也没有什么用。

为了降低这种复杂性，作者将可能的问题分为4大类共20条，以方便用户理解：

• Partitions（分区）

• Consumers（消费者）

• Producers（生产者）

• Brokers

Kafka是一种高效分布式消息传递系统，可提供内置数据冗余和弹性，同时保留高吞吐量和可扩展性。它包括自动数据保留限制，使其非常适合将数据视为流的应用程序，并且还支持对键值对映射建模的“压缩”流。

了解最佳实践之前，你需要熟悉一些关键术语：

• Message消息：Kafka中的记录或数据单元。每条消息都有一个键（key）和一个值（value），以及可选标题。

• 生产者：生产者向Kafka的topic发布消息。生产者决定要发布哪个topic分区，可以随机（循环）或使用基于消息密钥的分区算法。

• Broker：Kafka在分布式系统或集群中运行，集群中的每个节点都称为broker。

• Topic：Topic是发布数据记录或消息的类别。消费者订阅topic以读取写入其中的数据。

• Topic partition：topic分为多个分区，每个消息都有一个偏移量。每个分区通常至少复制一或两次。每个分区都有一个leader和至少一个副本（数据副本），这些副本存在于follower身上，可以防止broker失败。集群中的所有broker都是leader和follower，但是代理最多只有一个topic partition副本，leader用于所有读写操作。

• 偏移：为分区内的每条消息分配一个偏移量，这是一个单调递增整数，用作分区内消息的唯一标识符。

• 消费者：消费者通过订阅 topic partition读取Kafka主题的消息，消费应用程序，并处理消息以完成所需工作。

• Consumer group：消费者可以组织成消费者群组，分配topic partition以平衡组中所有使用者。在消费者群组中，所有消费者都在负载均衡模式下工作。换句话说，组中每个消费者都将看到每条消息。如果一个消费者离开，则将该分区分配给该组中的其他消费者，这个过程称为再平衡。如果组中的消费者多于分区，则一些消费者将闲置。如果组中的消费者少于分区，则某些消费者将使用来自多个分区的消息。

• Lag：当消费者无法从分区中读取消息，消费者就会出现Lag，表示为分区顶部后的偏移数。从Lag状态恢复所需的时间取决于消费者每秒消耗消息的速度：

time = messages / (consume rate per second - produce rate per second)

第一部分：使用分区的最佳实践！

在分区部分，我们需要了解分区的数据速率，以确保拥有正确的保留空间。分区的数据速率是生成数据的速率。换句话说，它是平均消息大小乘以每秒消息数。数据速率决定了给定时间内所需的保留空间（以字节为单位）。如果不知道数据速率，则无法正确计算满足基本保留目标所需的空间大小。数据速率指定了单个消费者需要支持的最低性能而保证不会出现Lag。

除非有其他架构需求，否则在写入topic时使用随机分区。当进行大规模操作时，分区之间的数据速率不均可能难以管理。需要注意以下三方面：

1、首先，“热点”（更高吞吐量）分区的消费者必须处理比消费者群组中其他消费者更多的消息，这可能导致处理和网络瓶颈。

2、其次，必须为具有最高数据速率的分区调整topic保留空间大小，这可能会导致topic中其他分区的磁盘使用量增加。

3、最后，在分区领导方面实现最佳平衡比简单地扩展到所有 brokers更复杂。“热点”分区的份量可能是同一topic中另一分区的10倍。

第二部分：使用消费者最佳实践！

如果消费者运行的Kafka版本低于0.10，请升级。在0.8.x版本中，消费者使用Apache ZooKeeper进行消费者群组协调，并且许多已知错误可能导致长期运行的平衡甚至是重新平衡算法的失败（我们称之为“重新平衡风暴”）。在重新平衡期间，将一个或多个分区分配给使用者群组中的每个使用者。在再平衡中，分区所有权在消费者中不断变通，阻止任何消费者在消费方面取得实际进展。

4、调整消费者套接字缓冲区以进行高速获取。在Kafka 0.10.x中，参数为isreceive.buffer.bytes，默认为64kB。在Kafka 0.8.x中，参数是socket.receive.buffer.bytes，默认为100kB。对于高吞吐量环境，这两个默认值都太小，特别是如果brocker和消费者之间的网络带宽延迟大于局域网（LAN）。对于延迟为1毫秒或更长的高带宽网络（10 Gbps或更高），请考虑将套接字缓冲区设置为8或16 MB。如果内存不足，请考虑1 MB，也可以使用值-1，这样底层操作系统可以根据网络条件调整缓冲区大小。但是，对于需要启动“热点”消费者的系统而言，自动调整的速度可能或比较慢。

5、设计高吞吐量消费者，以便在有保证的情况下实施背压，最好只消耗可以有效处理的东西，而不是消耗太多，以至于过程停止，退出消费者群组。 消费者应该使用固定大小的缓冲区（参见Disruptor模式），如果在Java虚拟机（JVM）中运行，最好是在堆外使用。固定大小的缓冲区将阻止消费者将大量数据拖到堆上，JVM花费所有时间来执行垃圾收集而不是做你想让它处理的工作——处理消息。

6、在JVM上运行消费者时，请注意垃圾回收可能对消费者产生的影响。例如，垃圾收集较长时间暂停可能导致ZooKeeper会话或者消费者组失去平衡。对于brocker来说也是如此，如果垃圾收集暂停时间过长，则可能会从集群中退出。

第三部分：使用生产者最佳实践！

7、配置生产者等待确认。 这就是生产者如何知道消息实际已经发送到brocker上的分区。在Kafka 0.10.x中，设置为acks; 在0.8.x中，它是request.required.acks。Kafka通过复制提供容错功能，因此单个节点的故障或分区leader的更改不会影响可用性。如果将生产者配置为没有ack（也称为“fire and forget”），则消息可能会无声地丢失。

8、配置生产者重试次数。默认值为3，通常太低。正确的值取决于需求，对于无法容忍数据丢失的应用程序，请考虑Integer.MAX_VALUE（实际上是无穷大），这可以防止leader分区的brocker无法立即响应生产请求。

9、对于高吞吐量生产者，调整缓冲区大小，特别是buffer.memory和batch.size（以字节为单位）。由于batch.size是按分区设置的，因此生产者性能和内存使用量可与topic中的分区数相关联。这里的值取决于几个因素：生产者数据速率（消息的大小和数量），生成的分区数以及可用的内存量。请记住，较大的缓冲区并不总是好的，如果生产者由于某种原因而停顿（例如，一个领导者通过确认响应较慢），在堆上缓存更多数据可能会导致更多垃圾收集。

10、制定应用程序跟踪指标，例如生成的消息数，平均生成的消息大小和消耗的消息数。

第四部分：brocker最佳实践！

11、Topic需要brocker的内存和CPU资源，日志压缩需要brocker上的堆（内存）和CPU周期才能成功完成，并且失败的日志压缩会使brocker处于无限增长的分区风险中。你可以在brocker上使用tunelog.cleaner.dedupe.buffer.size和log.cleaner.threads，但请记住，这些值会影响brocker上的堆使用情况。如果brocker抛出OutOfMemoryError异常，它将关闭并可能丢失数据。缓冲区大小和线程数将取决于要清理的主题分区数量以及这些分区中消息的数据速率和密钥大小。从Kafka 0.10.2.1版本开始，监视日志清理程序日志文件以查找ERROR条目是检测日志清理程序线程问题的最可靠方法。

12、监控brocker的网络吞吐量。确保使用发送（TX）和接收（RX），磁盘I/O，磁盘空间和CPU使用率来执行此操作。容量规划是维护集群性能的关键部分。

13、在集群中的brocker之间分配分区leader，其需要大量的网络I/O资源。例如，当使用复制因子3运行时，leader必须接收分区数据，并同步传递给所有副本，再传输给想要使用该数据的消费者。因此，在这个例子中，作为领导者，在使用网络I/O方面至少是follower的四倍，leader必须从磁盘读取，follower只需要写。

14、不要忽略监视brocker的同步副本（ISR）缩减，重复不足的分区和不受欢迎的lesder。这些是集群中潜在问题的迹象。例如，单个分区的频繁ISR收缩可能表明该分区的数据速率超过了leader为消费者和副本线程提供服务的能力。

15、根据需要修改Apache Log4j属性。Kafka代理日志记录可能会占用过多磁盘空间。但是，不要完全放弃日志记录，brocker日志可能是在事件发生后重建事件序列的最佳方式，有时也是唯一方式。

16、禁用topic自动创建有关的明确策略，定期清理未使用的topic。例如，如果x天没有看到任何消息，请考虑topic失效并将其从集群中删除，这样可以避免在集群中创建必须管理的其他元数据。

17、对于持续的高吞吐量代理，请提供足够的内存以避免从磁盘系统读取，应尽可能直接从操作系统的文件系统缓存中提供分区数据。但是，这意味着必须确保消费者能够跟上，滞后的消费者将迫使brocker从磁盘读取。

18、对于具有高吞吐量服务级别目标（SLO）的大型集群，请考虑将topic隔离到brocker子集。如何确定要隔离的topic取决于业务需求，例如，如果有多个使用相同集群的联机事务处理（OLTP）系统，则将每个系统的topic隔离到brocker的不同子集以帮助限制事件的潜在爆炸半径。

19、使用较新topic消息格式的旧客户端（反之亦然）会在brocker客户端转换格式时对brocker程序施加额外负担，尽可能避免这种情况。

20、不要认为在本地台式机上测试brocker代表在实际生产环境中的性能。使用复制因子1对分区的环回接口进行测试是与大多数生产环境完全不同的拓扑。通过环回可以忽略网络延迟，并且在不涉及复制时，接收leader确认所需的时间可能会有很大差异。

“如何使用Scala开发Apache Kafka”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注亿速云网站，小编将为大家输出更多高质量的实用文章！