「Kafka技术」Apache 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实现细节的人，我们将在另一篇后续博客文章中介绍一些更有趣的解决方案。

谢谢大家关注，转发，点赞和点在看。