Kafka是一个分布式、可扩展、弹性和容错的事件流平台，使您能够实时处理大量数据。Kafka Streams是一个库，通过构建Kafka生产者和消费者库，并利用Kafka的本地功能提供数据并行性、分布式协调、容错性和操作简单性，简化了应用程序开发。

架构

Kafka体系结构由两个主要层组成：

• 存储层：负责存储和复制应用程序生成和使用的事件。存储层基于分布式提交日志的抽象，分布式提交日志是一种以持久和容错的方式存储记录序列的数据结构。Kafka中的记录是一个键值对，它还包含时间戳和可选的头。记录可以代表任何类型的事件，如客户订单、付款、点击网站或传感器读数
• 计算层：负责在事件到达或离开系统时处理和转换事件。计算层由四个核心组件组成：生产者API、消费者API、流API和连接器API。这些组件允许您与存储层交互，并对事件执行各种操作

Kafka主题是共享一个共同主题或类别的记录流的逻辑名称。例如，您可以有一个名为“/orders”的主题，其中包含与系统中的订单相关的所有事件。一个主题可以划分为多个分区，这些分区是Kafka中并行性和可扩展性的单位。每个分区都是一个有序且不可变的记录序列，这些记录被分配给集群中的一个或多个代理（服务器）。一个分区可以有一个引导代理和零个或多个从引导代理复制数据以实现容错。

生产者API允许您将事件写入Kafka中的一个或多个主题。您可以指定各种参数，如键、值、分区、时间戳、标头、压缩类型和每个事件的确认级别。生产者API还处理事件的批处理、缓冲、序列化、错误处理和负载平衡。

消费者API允许您读取Kafka中一个或多个主题的事件。您可以指定各种参数，如组ID、偏移量、分区分配策略、反序列化类型、轮询间隔和每个使用者的提交行为。使用者API还处理事件的重新平衡、偏移管理、错误处理和并发控制。

流API允许您对Kafka中的事件执行有状态流处理。您可以使用构建在生产者和消费者API之上的Java库来定义复杂的转换、聚合、联接、窗口、状态存储和对事件的交互式查询。流API还处理流处理应用程序的可扩展性、容错性、一次性语义和应用程序重置。

连接器API允许您将Kafka与数据库、云服务或遗留系统等外部系统集成。您可以使用源连接器将外部系统中的数据摄取到Kafka主题中，也可以使用接收器连接器将Kafka话题中的数据导出到外部系统中。连接器API还处理配置管理。

流

Kafka Streams是对生产者和消费者的抽象，它使您能够专注于处理您的Kafka数据，而不必担心低级细节。您可以用Java或Scala编写代码，创建一个JAR文件，然后启动独立的应用程序，将记录流式传输到Kafka主题。

Kafka Streams为处理数据提供了两个主要的抽象：KStream和KTable。KStream表示一个无限制的记录流，可以使用无状态操作（如映射、过滤、连接和聚合）对其进行转换。KTable表示记录的变更日志流，可以按键查询这些记录，并通过计数、分组、窗口化和联接等有状态操作进行更新。

Kafka Streams还支持交互式查询，允许您从外部客户端或服务查询应用程序的状态。您可以使用交互式查询来公开KTables的最新值或状态存储的内容。

设计可扩展kafka体系结构的最佳实践

需要考虑的主要方面

• 分区：分区是实现Kafka中可扩展性和并行性的关键。一个主题被划分为一个或多个分区，这些分区分布在多个代理中。每个分区都有一个引导程序和零个或多个跟随程序，它们复制数据以实现容错。生产者根据记录键或自定义分区器将记录写入分区。使用者通过形成使用者组并将分区分配给组成员，并行地从分区中读取记录
• 复制：复制是Kafka中确保数据可用性和持久性的机制。每个分区都可以有一个复制因子，该因子指定在不同的代理上维护多少个数据副本。分区的引导者处理所有的读写请求，而跟随者被动地复制来自引导者的数据。如果领导者失败，其中一名追随者将自动当选为新的领导者
• 序列化：序列化是将数据从一种格式转换为另一种格式以进行传输或存储的过程。Kafka使用字节数组作为记录的数据类型，因此您需要在将数据发送到Kafka之前对其进行序列化，并在从Kafka收到数据后对其进行反序列化。Kafka为字符串、整数、字节等常见数据类型提供了内置的序列化器和反序列化器。您还可以为JSON、Avro、Protobuf等复杂数据类型使用自定义序列化器和反串行器。
• 压缩：压缩是通过消除冗余或使用编码方案来减小数据大小的技术。压缩可以通过减少网络带宽使用、磁盘空间使用和CPU使用来提高Kafka的性能和效率。Kafka支持多种压缩编解码器，如gzip、snappy、lz4、zstd等。您可以在生产者级别或主题级别配置压缩
• 保留：保留是决定数据在被删除或压缩之前在Kafka中保留多长时间的策略。保留可以基于时间或大小限制。基于时间的保留会删除超过指定持续时间的记录。当分区的总大小超过指定的限制时，基于大小的保留将删除记录。您还可以使用日志压缩只保留每个记录键的最新值

最佳实践

• 有意义的记录键：记录键对于确定如何在Kafka中对数据进行分区和处理非常重要。您应该使用反映数据语义和处理逻辑的有意义的记录键。例如，如果要计算每个客户的订单数量，则应使用客户ID作为记录密钥。避免使用空键或随机键，因为它们可能会导致数据分布不均匀或不必要的重新分区
• 避免创建过多或过少的主题：过多的主题会增加管理元数据和代理连接的开销。主题过少会导致数据失真和消费者之间的争论。一个好的经验法则是每个逻辑数据类型或域有一个主题
• 选择一个平衡吞吐量和延迟的分区大小：更大的分区大小可以通过允许更多记录一起批处理来提高吞吐量。然而，它也会延迟向消费者交付记录，从而增加延迟。较小的分区大小可以通过更快地传递记录来减少延迟。然而，它也可以通过降低批处理效率来降低吞吐量。一个好的经验法则是分区大小在1MB到10MB之间
• 选择与所需并行级别匹配的分区数：分区数决定了有多少消费者可以并行使用主题中的数据。如果您的分区比使用者多，则某些分区将处于空闲状态。如果你的消费者比分区多，那么一些消费者就会饿死。一个好的经验法则是至少拥有与预期消费者的最大数量一样多的分区
• 选择一个符合可用性和持久性要求的复制因子：复制因子决定了在不同的代理上维护每个分区的副本数量。更高的复制因子可以在一个复制副本出现故障时允许消费者切换到另一个复制复制副本，从而提高可用性。它还可以通过降低代理崩溃时数据丢失的风险来提高耐用性。但是，更高的复制系数也会增加磁盘空间使用率和网络流量。一个好的经验法则是复制因子在2到4之间。
• 选择适当的状态存储：状态存储是存储流应用程序状态的本地数据库，如KTables或自定义聚合。Kafka Streams提供两种类型的状态存储：RocksDB和内存。RocksDB是一个持久的键值存储，支持快速高效地访问大量数据。内存中状态存储速度更快，但不那么耐用，而且更占用内存。您应该选择适合您的性能和可靠性要求的状态存储类型
• 调整您的配置参数：Kafka Streams公开了许多配置参数，允许您自定义流应用程序的行为和性能。您应该根据您的用例和环境来调整这些参数。例如，您可以调整缓冲区大小、批处理大小、提交间隔、缓存大小、轮询超时等，以优化应用程序的吞吐量和延迟。您还可以启用度量和日志记录来监视应用程序并对其进行故障排除
• 测试和基准测试您的应用程序：测试和基准是确保流媒体应用程序的质量和性能的重要步骤。您应该使用实际的数据和负载场景来测试您的应用程序，并测量关键指标，如吞吐量、延迟、资源利用率、错误率等。您还应该将结果与您的期望和要求进行比较，并确定需要解决的任何瓶颈或问题
• 保护您的Kafka集群和数据：您应该使用SSL、SASL或Kerberos等机制为您的Kavka集群启用身份验证、授权和加密。您还应该使用ACL、配额、复制和备份等功能来保护您的数据免受未经授权的访问、篡改或丢失

常见问题及解决方案

数据重复

使用Kafka内置的数据冗余和恢复功能，确保数据只存储一次。您还可以使用KafkaConnect将Kafka与其他系统集成，避免数据重复。

数据丢失或群集故障

您可以配置您的Kafka集群进行容错。您可以使用复制来确保您的数据存储在多个代理上。您还可以使用负载均衡器在代理之间分配流量。

数据倾斜

当数据在分区之间分布不均匀时，可能会发生数据偏斜，这会影响Kafka集群的性能。为了解决这个问题，您可以使用分区策略，在分区之间均匀分布数据，例如基于键的分区或循环分区。

缓慢的再平衡时间

您可以利用缓存来减少需要重新平衡的数据量。您还可以监控主题的数据量，以确保Kafka的性能不受影响。

网络拥塞

通过使用高速网络接口优化网络配置，将网络配置为低延迟，并使用压缩来减少传输的数据量。

冗余数据存储

仅将Kafka用于短期存储数据，并根据您的特定需求将数据迁移到关系数据库或非关系数据库。您还可以将Kafka配置为使用HDFS或blob存储以获得额外的永久性。

延迟或无序事件

您可以在Kafka Streams中使用事件时间语义和窗口操作来处理延迟或无序事件。事件时间语义意味着您使用嵌入事件中的时间戳而不是处理时间来确定事件的顺序。通过窗口操作，可以将事件分组为固定或动态时间间隔，并对其应用聚合或联接。您还可以为处理延迟到达或更新的窗口指定宽限期和保留期。

模式演变或数据格式更改

您可以使用模式注册表和序列化框架，如Avro、Protobuf或JSONSchema来处理模式演变或数据格式更改。模式注册表是一种存储和管理数据模式的服务。序列化框架允许您通过架构兼容性检查和转换对数据进行序列化和反序列化。您可以使用自定义序列化程序和反序列化程序将模式注册表和序列化框架与Kafka Streams集成。

安全漏洞

您可以使用加密、身份验证和授权来保护您的Kafka集群。您还可以监视Kafka集群的安全问题，并在必要时采取纠正措施。

监测不足

您可以实施一个全面的监控解决方案，跟踪吞吐量、延迟和错误率等关键指标。您还可以设置警报，以便在出现性能问题时通知您，并在必要时采取纠正措施。

实施不当

它可能导致数据处理效率低下和技术债务。为了避免这种情况，您可以在设计和实现Kafka架构时遵循最佳实践。您还可以监视Kafka集群的性能和可用性问题，并在必要时采取纠正措施。

最后一句话

为高性能流应用程序设计和实现可扩展的Kafka架构需要仔细规划和实现。Kafka事件驱动架构、Kafka Streams和Apache Pulsar分布式消息和流媒体平台是提高Kafka性能的一些解决方案。

在之前的一篇博客文章中，我们介绍了Apache Kafka®的一次语义。这篇文章介绍了各种消息传递语义，介绍了幂等生成器、事务和Kafka流的一次处理语义。现在，我们将继续上一节的内容，深入探讨Apache Kafka中的事务。该文档的目标是让读者熟悉有效使用Apache Kafka中的事务API所需的主要概念。

我们将讨论设计事务API的主要用例、Kafka的事务语义、用于Java客户端的事务API的细节、实现的有趣方面，以及在使用API时的重要注意事项。

这篇博客文章并不是关于使用事务细节的教程，我们也不会深入讨论设计细节。相反，我们将在适当的地方链接到JavaDocs或设计文档，以供希望深入研究的读者使用。

我们希望读者熟悉基本的Kafka概念，比如主题、分区、日志偏移量，以及代理和客户在基于Kafka的应用程序中的角色。熟悉Java的Kafka客户机也会有所帮助。

为什么交易?

我们在Kafka中设计的事务主要用于那些显示“读-进程-写”模式的应用程序，其中的读和写来自于异步数据流，比如Kafka主题。这种应用程序通常称为流处理应用程序。

第一代流处理应用程序可以容忍不准确的处理。例如，使用web页面印象流并生成每个web页面的视图聚合计数的应用程序可以容忍计数中的一些错误。

然而，随着这些应用程序的流行，对具有更强语义的流处理应用程序的需求也在增长。例如，一些金融机构使用流处理应用程序来处理用户帐户上的借方和贷方。在这些情况下，不能容忍处理过程中的错误:我们需要准确地一次处理所有消息，没有例外。

更正式地说，如果流处理应用程序使用消息a并生成消息B，使得B = F(a)，那么仅一次处理就意味着如果且仅当成功生成B时才使用a，反之亦然。

使用配置为至少一次传递语义的普通Kafka生产者和消费者，流处理应用程序可能会在以下方面失去一次处理语义:

1. 由于内部重试，生产者.send()可能导致消息B的重复写入。这是由幂等生产者解决的，并不是本文其余部分的重点。
2. 我们可能会重新处理输入消息A，导致将重复的B消息写入输出，这违反了一次处理语义。如果流处理应用程序在写入B之后但在将A标记为已使用之前崩溃，则可能发生重新处理。因此，当它恢复时，它将再次消耗A并再次写入B，从而导致重复。
3. 最后，在分布式环境中，应用程序会崩溃，甚至更糟!-暂时失去与系统其余部分的连接。通常，会自动启动新实例来替换那些被认为丢失的实例。通过这个过程，我们可能会有多个实例处理相同的输入主题，并写入相同的输出主题，从而导致输出重复，并违反一次处理语义。我们称之为“僵尸实例”问题。

我们在Kafka中设计了事务api来解决第二个和第三个问题。事务通过使这些周期成为原子性的，并通过促进僵死的隔离，从而在读写周期中实现精确的一次处理。

事务性语义

原子多分区写道

事务允许对多个Kafka主题和分区进行原子写入。事务中包含的所有消息都将被成功写入，或者一个也不写入。例如，处理过程中的错误可能导致事务中止，在这种情况下，来自事务的任何消息都不会被使用者读取。现在我们来看看它是如何实现原子读写周期的。

首先，让我们考虑原子读写周期的含义。简而言之,这意味着如果一个应用程序使用一个消息的抵消X topic-partition tp0,和写消息B topic-partition tp1在消息上做一些处理,B = F (a),然后read-process-write周期是a和B原子只有在消息被认为成功地消耗和发表在一起,要么一无所有。

现在，只有当消息A的偏移量X标记为已使用时，才会认为它是从主题分区tp0使用的。将偏移量标记为已使用的偏移量称为提交偏移量。在Kafka中，我们通过写入内部Kafka主题offsets主题来记录偏移量提交。仅当消息的偏移量提交到偏移量主题时，才认为该消息已被消耗。

因此从一个偏移量提交只是另一个写一个卡夫卡的话题,因为消息被认为是只有当其抵消消费承诺,原子还写跨多个主题和分区使原子read-process-write周期:提交的抵消X的补偿主题写的消息B tp1将单个事务的一部分,因此原子。

僵尸击剑（Zombie fencing）

我们通过要求为每个事务生产者分配一个称为transaction .id的惟一标识符来解决zombie实例的问题。这用于跨流程重新启动标识相同的生产者实例。

API要求事务生产者的第一个操作应该是显式注册其事务。使用Kafka集群的id。当它这样做时，Kafka代理使用给定的事务检查打开的事务。id并完成它们。它还增加与transaction .id关联的epoch。epoch是存储在每个transaction .id中的内部元数据。

一旦epoch被碰撞，任何具有相同事务的生产者。身份证和旧时代被认为是僵尸，被隔离。来自这些生产者的未来事务写将被拒绝。

读事务消息

现在，让我们将注意力转向在读取作为事务的一部分写入的消息时提供的保证。

Kafka使用者只会在事务被提交时才会向应用程序提交事务消息。换句话说，使用者不会交付作为开放事务一部分的事务性消息，也不会交付作为中止事务一部分的消息。

值得注意的是，上面的保证没有达到原子读取。特别是，当使用Kafka使用者来消费来自主题的消息时，应用程序将不知道这些消息是否作为事务的一部分写入，因此它们不知道事务何时开始或结束。进一步说,一个给定的消费者不保证订阅所有分区事务的一部分,它没有发现这个方法,这就很难保证所有的信息是一个事务的一部分最终会被一个消费者。

简而言之:Kafka保证使用者最终只交付非事务性消息或提交的事务性消息。它将从打开的事务中保留消息，并从中止的事务中过滤出消息。

Java中的事务API

事务特性主要是一个服务器端和协议级特性，任何支持它的客户端库都可以使用它。用Java编写的“读-处理-写”应用程序，使用Kafka的事务API，看起来应该是这样的:

第1-5行通过指定事务设置生产者。配置id并将其注册到initTransactions API。inittransactions()返回后，由具有相同事务的生产者的另一个实例启动的任何事务。id会被关闭和隔离。

第7-10行指定KafkaConsumer应该只读取非事务性消息，或者从它的输入主题中提交事务性消息。流处理应用程序通常在多个读写阶段处理其数据，每个阶段使用前一阶段的输出作为其输入。通过指定read_committed模式，我们可以在所有阶段只执行一次处理。

第14-21行演示了读写循环的核心:我们使用一些记录，启动一个事务，处理使用的记录，将处理过的记录写入输出主题，将使用的偏移量发送到偏移量主题，最后提交事务。根据上面提到的保证，我们知道偏移量和输出记录将作为一个原子单元提交。

事务是如何工作的

在本节中，我们将简要概述上述事务api引入的新组件和新数据流。为了更详尽地讨论这个主题，您可以阅读原始设计文档，或者观看介绍事务的Kafka峰会演讲。

下面内容的目标是在调试使用事务的应用程序时，或者在尝试调优事务以获得更好的性能时，提供一个心智模型。

事务协调器和事务日志

Kafka 0.11.0中的transactions API引入的组件是事务协调器和上图右侧的事务日志。

事务协调器是在每个Kafka代理中运行的模块。事务日志是一个内部kafka主题。每个协调器在事务日志中拥有一些分区子集。其代理为其领导的分区。

每一个事务。id通过一个简单的哈希函数映射到事务日志的特定分区。这意味着只有一个协调器拥有给定的transaction .id。

通过这种方式，我们利用Kafka的rock solid复制协议和leader选择过程来确保事务协调器总是可用的，并且所有事务状态都被持久地存储。

值得注意的是，事务日志只存储事务的最新状态，而不是事务中的实际消息。消息仅存储在实际的主题分区中。事务可以处于“进行中”、“准备提交”和“完成”等不同状态。存储在事务日志中的就是这种状态和相关的元数据。

数据流

在较高的层次上，数据流可以分为四种不同的类型。

A:生产者和事务协调者的交互

执行事务时，生产者向事务协调器发出以下请求:

initTransactions API注册一个事务。id与协调器。此时，协调器将使用该事务关闭任何挂起的事务。id和碰撞的时代，以栅栏出僵尸。每个生产者会话只发生一次。

当生产者在事务中第一次将数据发送到一个分区时，该分区首先向协调器注册。

当应用程序调用commitTransaction或abortTransaction时，将向协调器发送一个请求，以开始两阶段提交协议。

B:协调器和事务日志的交互

随着事务的进展，生产者发送上述请求来更新协调器上事务的状态。事务协调器将其拥有的每个事务的状态保存在内存中，并将该状态写入事务日志(以三种方式复制，因此是持久的)。

事务协调器是从事务日志中读写的惟一组件。如果给定的代理失败，则将选出一个新的协调器作为死代理拥有的事务日志分区的leader，它将从传入分区读取消息，以便为这些分区中的事务重建其内存状态。

C:生产者写数据到目标主题分区

在向协调器注册了事务中的新分区之后，生产者将数据正常地发送到实际的分区。这是同一个生产者。发送流，但是要进行一些额外的验证以确保生产者不受保护。

D:主题分区交互的协调器

在生产者发起提交(或中止)之后，协调器开始两阶段提交协议。

在第一阶段，协调器将其内部状态更新为“prepare_commit”，并在事务日志中更新此状态。一旦完成了这一步，就可以保证在任何情况下提交事务。

然后协调器开始第2阶段，将事务提交标记写入作为事务一部分的主题分区。

这些事务标记不公开给应用程序，而是由处于read_committed模式的使用者使用，以过滤掉中止的事务中的消息，并且不返回作为打开事务一部分的消息(即，在日志中但没有与之关联的事务标记的。

一旦写入了标记，事务协调器将事务标记为“完成”，并且生产者可以启动下一个事务。

实践中处理交易

既然我们已经理解了事务的语义以及它们是如何工作的，那么我们就将注意力转向编写利用事务的应用程序的实践方面。

如何选择一个transaction .id

事务。id在保护僵尸方面起着重要作用。但是保持一个标识符在不同的生产者会话之间是一致的，并且适当地隔离僵尸是有点棘手的。

正确隔离“僵尸”的关键是确保对于给定的transaction .id，读写周期中的输入主题和分区总是相同的。如果这不是真的，那么一些消息可能会通过事务提供的围栏泄漏。

例如，在一个分布式流处理应用程序中，假设主题分区tp0最初是由transactional处理的。T0 id。如果在以后的某个时候，它可以映射到另一个具有transactional的生产者。id T1，在T0和T1之间没有栅栏。因此，可以对来自tp0的消息进行重新处理，这违反了一次处理的保证。

实际上，必须存储输入分区和事务之间的映射。外部存储中的id，或者对其进行一些静态编码。Kafka Streams选择后一种方法来解决这个问题。

事务如何执行，以及如何调优它们

事务生产者的性能

让我们将注意力转向事务如何执行。

首先，事务只导致适度的写放大。增加的写是由于:

1. 对于每个事务，我们都有额外的rpc向协调器注册分区。这些是成批的，因此我们的rpc比事务中的分区要少。
2. 在完成事务时，必须将一个事务标记写入参与事务的每个分区。同样，事务协调器在单个RPC中批量处理为同一代理绑定的所有标记，因此我们在那里保存RPC开销。但是我们不能避免对事务中的每个分区进行一次额外的写操作。
3. 最后，我们将状态更改写入事务日志。这包括对添加到事务中的每批分区的写操作、“prepare_commit”状态和“complete_commit”状态。

我们可以看到，开销与作为事务一部分写入的消息的数量无关。因此，提高吞吐量的关键是在每个事务中包含更多的消息。

实际上，对于在最大吞吐量下生成1KB记录的生产者，每100ms提交一条消息只会导致吞吐量降低3%。较小的消息或较短的事务提交间隔将导致更严重的降级。

增加事务持续时间的主要代价是增加了端到端延迟。请记住，读取事务性消息的使用者不会交付作为开放事务一部分的消息。因此，提交间隔的时间越长，应用程序的等待时间就越长，从而增加了端到端延迟。

事务消费者的性能

事务性消费者比生产者简单得多，因为它所需要做的就是:

1. 筛选属于中止的事务的消息。
2. 不返回作为开放事务一部分的事务消息。

因此，当以read_committed模式读取事务消息时，事务使用者的吞吐量没有下降。这样做的主要原因是，我们在读取事务性消息时保持零副本读取。

而且，使用者不需要任何缓冲来等待事务完成。相反，代理不允许它提前进行补偿，其中包括打开的事务。

因此，消费者是极其轻量级和高效的。有兴趣的读者可以在本文档中了解消费者设计的细节。

进一步的阅读

我们刚刚触及了Apache Kafka中事务的皮毛。幸运的是，几乎所有的设计细节都记录在网上。有关文件如下:

1. 最初的Kafka KIP:它提供了关于数据流的详细信息和公共接口的概述，特别是随事务而来的配置选项。
2. 原始设计文档:不适合胆小的人，这是权威的地方——源代码外!-了解如何处理每个事务RPC，如何维护事务日志，如何清除事务数据，等等。
3. KafkaProducer javadocs:这是一个学习如何使用新api的好地方。页面开头的示例以及send方法的文档都是很好的起点。

结论

在这篇文章中，我们了解了Apache Kafka中事务API的关键设计目标，理解了事务API的语义，并对API的实际工作方式有了更深入的了解。

如果我们考虑一个读-进程-写循环，这篇文章主要讨论了读和写路径，处理本身就是一个黑盒。事实上，在处理阶段可以做很多事情，这使得仅使用事务api无法保证一次处理。例如，如果处理对其他存储系统有副作用，这里介绍的api不足以保证只进行一次处理。

Kafka Streams框架使用这里描述的事务api向上移动价值链，并为各种流处理应用程序提供一次处理，甚至包括那些在处理期间更新某些额外状态存储的应用程序。

将来的一篇博客文章将讨论Kafka流如何提供一次处理语义，以及如何编写利用它的应用程序。

最后，对于那些渴望了解上述api实现细节的人，我们将在另一篇后续博客文章中介绍一些更有趣的解决方案。

原文：https://www.confluent.io/blog/transactions-apache-kafka/

本文：https://pub.intelligentx.net/node/829

讨论：请加入知识星球【首席架构师圈】或者飞聊小组【首席架构师智库】