。

相反#xff0c;我们将在适当的地方链接到JavaDocs或设计文档#xff0c;以供希望深入研究的读者使用。

为什么交易?

我们在Kafka中设计的事务主要用于那些显示“读-进程-写”模式的应用…这篇博客文章并不是关于使用事务细节的教程我们也不会深入讨论设计细节。

相反我们将在适当的地方链接到JavaDocs或设计文档以供希望深入研究的读者使用。

为什么交易?

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

这种应用程序通常称为流处理应用程序。

第一代流处理应用程序可以容忍不准确的处理。

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

然而随着这些应用程序的流行对具有更强语义的流处理应用程序的需求也在增长。

例如一些金融机构使用流处理应用程序来处理用户帐户上的借方和贷方。

在这些情况下不能容忍处理过程中的错误:我们需要准确地一次处理所有消息没有例外。

F(a)那么仅一次处理就意味着如果且仅当成功生成B时才使用a反之亦然。

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

由于内部重试生产者.send()可能导致消息B的重复写入。

这是由幂等生产者解决的并不是本文其余部分的重点。

我们可能会重新处理输入消息A导致将重复的B消息写入输出这违反了一次处理语义。

如果流处理应用程序在写入B之后但在将A标记为已使用之前崩溃则可能发生重新处理。

因此当它恢复时它将再次消耗A并再次写入B从而导致重复。

最后在分布式环境中应用程序会崩溃甚至更糟!-暂时失去与系统其余部分的连接。

通常会自动启动新实例来替换那些被认为丢失的实例。

通过这个过程我们可能会有多个实例处理相同的输入主题并写入相同的输出主题从而导致输出重复并违反一次处理语义。

我们称之为“僵尸实例”问题。

我们在Kafka中设计了事务api来解决第二个和第三个问题。

事务通过使这些周期成为原子性的并通过促进僵死的隔离从而在读写周期中实现精确的一次处理。

事务性语义

事务允许对多个Kafka主题和分区进行原子写入。

事务中包含的所有消息都将被成功写入或者一个也不写入。

例如处理过程中的错误可能导致事务中止在这种情况下来自事务的任何消息都不会被使用者读取。

现在我们来看看它是如何实现原子读写周期的。

首先让我们考虑原子读写周期的含义。

简而言之,这意味着如果一个应用程序使用一个消息的抵消X

topic-partition

(a),然后read-process-write周期是a和B原子只有在消息被认为成功地消耗和发表在一起,要么一无所有。

现在只有当消息A的偏移量X标记为已使用时才会认为它是从主题分区tp0使用的。

将偏移量标记为已使用的偏移量称为提交偏移量。

在Kafka中我们通过写入内部Kafka主题offsets主题来记录偏移量提交。

仅当消息的偏移量提交到偏移量主题时才认为该消息已被消耗。

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

僵尸击剑Zombie

我们通过要求为每个事务生产者分配一个称为transaction

.id的惟一标识符来解决zombie实例的问题。

这用于跨流程重新启动标识相同的生产者实例。

API要求事务生产者的第一个操作应该是显式注册其事务。

使用Kafka集群的id。

当它这样做时Kafka代理使用给定的事务检查打开的事务。

id并完成它们。

它还增加与transaction

.id关联的epoch。

epoch是存储在每个transaction

一旦epoch被碰撞任何具有相同事务的生产者。

身份证和旧时代被认为是僵尸被隔离。

来自这些生产者的未来事务写将被拒绝。

读事务消息

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

Kafka使用者只会在事务被提交时才会向应用程序提交事务消息。

换句话说使用者不会交付作为开放事务一部分的事务性消息也不会交付作为中止事务一部分的消息。

值得注意的是上面的保证没有达到原子读取。

特别是当使用Kafka使用者来消费来自主题的消息时应用程序将不知道这些消息是否作为事务的一部分写入因此它们不知道事务何时开始或结束。

进一步说,一个给定的消费者不保证订阅所有分区事务的一部分,它没有发现这个方法,这就很难保证所有的信息是一个事务的一部分最终会被一个消费者。

简而言之:Kafka保证使用者最终只交付非事务性消息或提交的事务性消息。

它将从打开的事务中保留消息并从中止的事务中过滤出消息。

事务特性主要是一个服务器端和协议级特性任何支持它的客户端库都可以使用它。

用Java编写的“读-处理-写”应用程序使用Kafka的事务API看起来应该是这样的:

第1-5行通过指定事务设置生产者。

配置id并将其注册到initTransactions

API。

inittransactions()返回后由具有相同事务的生产者的另一个实例启动的任何事务。

id会被关闭和隔离。

第7-10行指定KafkaConsumer应该只读取非事务性消息或者从它的输入主题中提交事务性消息。

流处理应用程序通常在多个读写阶段处理其数据每个阶段使用前一阶段的输出作为其输入。

通过指定read_committed模式我们可以在所有阶段只执行一次处理。

第14-21行演示了读写循环的核心:我们使用一些记录启动一个事务处理使用的记录将处理过的记录写入输出主题将使用的偏移量发送到偏移量主题最后提交事务。

根据上面提到的保证我们知道偏移量和输出记录将作为一个原子单元提交。

在本节中我们将简要概述上述事务api引入的新组件和新数据流。

为了更详尽地讨论这个主题您可以阅读原始设计文档或者观看介绍事务的Kafka峰会演讲。

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

事务协调器和事务日志

事务协调器是在每个Kafka代理中运行的模块。

事务日志是一个内部kafka主题。

每个协调器在事务日志中拥有一些分区子集。

其代理为其领导的分区。

每一个事务。

id通过一个简单的哈希函数映射到事务日志的特定分区。

这意味着只有一个协调器拥有给定的transaction

solid复制协议和leader选择过程来确保事务协调器总是可用的并且所有事务状态都被持久地存储。

值得注意的是事务日志只存储事务的最新状态而不是事务中的实际消息。

消息仅存储在实际的主题分区中。

事务可以处于“进行中”、“准备提交”和“完成”等不同状态。

存储在事务日志中的就是这种状态和相关的元数据。

数据流

API注册一个事务。

id与协调器。

此时协调器将使用该事务关闭任何挂起的事务。

id和碰撞的时代以栅栏出僵尸。

每个生产者会话只发生一次。

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

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

随着事务的进展生产者发送上述请求来更新协调器上事务的状态。

事务协调器将其拥有的每个事务的状态保存在内存中并将该状态写入事务日志(以三种方式复制因此是持久的)。

事务协调器是从事务日志中读写的惟一组件。

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

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

在向协调器注册了事务中的新分区之后生产者将数据正常地发送到实际的分区。

这是同一个生产者。

发送流但是要进行一些额外的验证以确保生产者不受保护。

在第一阶段协调器将其内部状态更新为“prepare_commit”并在事务日志中更新此状态。

一旦完成了这一步就可以保证在任何情况下提交事务。

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

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

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

实践中处理交易

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

如何选择一个transaction

事务。

id在保护僵尸方面起着重要作用。

但是保持一个标识符在不同的生产者会话之间是一致的并且适当地隔离僵尸是有点棘手的。

.id读写周期中的输入主题和分区总是相同的。

如果这不是真的那么一些消息可能会通过事务提供的围栏泄漏。

例如在一个分布式流处理应用程序中假设主题分区tp0最初是由transactional处理的。

T0

id。

如果在以后的某个时候它可以映射到另一个具有transactional的生产者。

id

T1在T0和T1之间没有栅栏。

因此可以对来自tp0的消息进行重新处理这违反了一次处理的保证。

实际上必须存储输入分区和事务之间的映射。

外部存储中的id或者对其进行一些静态编码。

Kafka

对于每个事务我们都有额外的rpc向协调器注册分区。

这些是成批的因此我们的rpc比事务中的分区要少。

在完成事务时必须将一个事务标记写入参与事务的每个分区。

同样事务协调器在单个RPC中批量处理为同一代理绑定的所有标记因此我们在那里保存RPC开销。

但是我们不能避免对事务中的每个分区进行一次额外的写操作。

最后我们将状态更改写入事务日志。

这包括对添加到事务中的每批分区的写操作、“prepare_commit”状态和“complete_commit”状态。

我们可以看到开销与作为事务一部分写入的消息的数量无关。

因此提高吞吐量的关键是在每个事务中包含更多的消息。

实际上对于在最大吞吐量下生成1KB记录的生产者每100ms提交一条消息只会导致吞吐量降低3%。

较小的消息或较短的事务提交间隔将导致更严重的降级。

增加事务持续时间的主要代价是增加了端到端延迟。

请记住读取事务性消息的使用者不会交付作为开放事务一部分的消息。

因此提交间隔的时间越长应用程序的等待时间就越长从而增加了端到端延迟。

筛选属于中止的事务的消息。

不返回作为开放事务一部分的事务消息。

因此当以read_committed模式读取事务消息时事务使用者的吞吐量没有下降。

这样做的主要原因是我们在读取事务性消息时保持零副本读取。

而且使用者不需要任何缓冲来等待事务完成。

相反代理不允许它提前进行补偿其中包括打开的事务。

因此消费者是极其轻量级和高效的。

有兴趣的读者可以在本文档中了解消费者设计的细节。

Kafka中事务的皮毛。

幸运的是几乎所有的设计细节都记录在网上。

有关文件如下:

KIP:它提供了关于数据流的详细信息和公共接口的概述特别是随事务而来的配置选项。

原始设计文档:不适合胆小的人这是权威的地方——源代码外!-了解如何处理每个事务RPC如何维护事务日志如何清除事务数据等等。

KafkaProducer

javadocs:这是一个学习如何使用新api的好地方。

页面开头的示例以及send方法的文档都是很好的起点。

Kafka中事务API的关键设计目标理解了事务API的语义并对API的实际工作方式有了更深入的了解。

如果我们考虑一个读-进程-写循环这篇文章主要讨论了读和写路径处理本身就是一个黑盒。

事实上在处理阶段可以做很多事情这使得仅使用事务api无法保证一次处理。

例如如果处理对其他存储系统有副作用这里介绍的api不足以保证只进行一次处理。

Kafka

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

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

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