本博客系列中的其他部分

第1部分：Spring Cloud Stream Kafka 应用中的事务简介

第2部分：Spring Cloud Stream Kafka 应用中的生产者发起的事务

第3部分：在 Spring Cloud Stream 中与外部事务管理器同步

第4部分：使用 Spring Cloud Stream 和 Apache Kafka 的事务回滚策略

在我们之前本系列讨论中，已经对 Spring Cloud Stream Kafka 应用中事务的工作原理进行了基本分析，现在我们终于来到了问题的核心：**精确一次语义**，这是流应用中一个经常被讨论和需要的特性。在本系列博客的这一部分中，我们将探讨如何通过 Apache Kafka 事务在 Spring Cloud Stream 应用中实现精确一次语义。前面几节关于事务工作原理的知识，使得理解 Spring Cloud Stream Kafka 应用如何实现精确一次语义变得相对容易。

需要特别注意的是，除了我们在本系列博客前面文章中已经看到的代码之外，我们不需要编写任何新的代码来实现**精确一次语义**。这篇博客阐明了在 Spring Cloud Stream Kafka 应用中充分支持精确一次语义所需的一些期望。

在分布式计算中，精确一次语义很难实现。我们不会深入探讨所有技术细节来解释为什么这是一项如此困难的任务。有兴趣了解精确一次语义的底层原理以及为什么它在分布式系统中难以实现的读者，可以参考相关领域的更广泛文献。这篇来自 Confluent 的博客是了解这些技术挑战以及 Apache Kafka 为实现这些挑战而采用的解决方案的良好起点。

尽管我们不会深入探讨细节，但了解 Apache Kafka 提供的不同交付保证还是很有价值的。主要有三种主要的交付保证：

• 至少一次语义
• 最多一次语义
• 精确一次语义

在**至少一次**的交付语义中，应用程序可能会接收数据一次或多次，但保证至少接收一次。在**最多一次**语义的交付保证中，应用程序可能会接收数据零次或一次，这意味着存在数据丢失的可能性。另一方面，**精确一次**语义保证，顾名思义，只交付一次。根据应用程序的使用场景，您可以选择使用其中任何一种保证。默认情况下，Apache Kafka 提供至少一次的交付保证，这意味着记录可能会被多次交付。如果您的应用程序能够处理重复记录或无记录的后果，那么使用非精确一次的保证是可以的。相反，如果您处理的是关键任务数据，例如金融系统或医疗数据，则必须保证精确一次的交付和处理，以避免严重后果。由于像 Apache Kafka 这样的系统具有分布式特性，因此由于许多活动部件的特性，通常很难实现精确一次语义。

Spring Cloud Stream Kafka 和精确一次语义

我们在本系列博客前面的文章中看到了许多不同的场景。Apache Kafka 中的精确一次语义解决了**读-处理-写**(或**消费-转换-生产**)应用。有时会对我们究竟在做什么“一次”感到困惑。是初始消费、数据处理还是最终的生产部分？Apache Kafka 保证了整个**读->处理-写**序列的精确一次语义。在此序列中，读和处理部分始终是**至少一次** - 例如，如果由于任何原因处理或写的一部分失败。当您依赖精确一次交付时，事务非常关键，以便最终成功发布数据或回滚。一个潜在的副作用是初始消费和处理可能会发生多次。例如，如果事务回滚，则不会更新消费者偏移量，并且下一次轮询（如果它是 Spring Cloud Stream 中的重试或应用程序重新启动时）会重新交付相同的记录并再次处理。因此，在消费和处理（转换）部分的保证是至少一次，这是一个关键的理解点。任何以read_committed隔离级别运行的下游消费者都将只接收来自上游处理器的消息正好一次。因此，必须理解，在精确一次交付的世界中，处理器和下游消费者必须协调才能从精确一次语义中受益。任何以read_uncommitted隔离级别运行的生产主题的消费者可能会看到重复数据。

另一个需要牢记的点是，由于记录的消费和处理可能会发生多次，因此应用程序代码需要遵循幂等模式，这主要是在您的代码与外部系统（例如数据库）交互时需要考虑的问题。在这种情况下，应用程序需要确保用户代码没有副作用。

让我们重新审视之前看到的简单消费-处理-生产循环的代码。

@Bean
public Consumer<PersonEvent> process(TxCode txCode) {
   return txCode::run;
}

@Component
class TxCode {

   @Transactional
   void run(PersonEvent pe) {
       Person person = new Person();
       person.setName(pe.getName());

       Person savedPerson = repository.save(person);

       PersonEvent event = new PersonEvent();
       event.setName(savedPerson.getName());
       event.setType("PersonSaved");
       streamBridge.send("process-out-0", event);
   }
}

如前所述，要使此应用程序具有事务性，我们必须使用适当的值提供spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix配置属性。在 Spring Cloud Stream 中，提供此属性是使上述代码段完全能够实现精确一次交付所需的全部操作。整个端到端流程在事务边界内运行（尽管我们在上面的示例中有两个事务）。我们有一个外部 Kafka 事务，当容器调用侦听器时在容器中启动，以及由事务拦截器启动的另一个 JPA 事务。当StreamBridge发送发生时，将使用来自初始 Kafka 事务的相同事务资源，但它不会在控制权返回到容器后才提交。当方法退出时，JPA 事务将提交。假设这里出了问题，数据库操作抛出异常。在这种情况下，JPA 不会提交，它将回滚，并且异常传播回侦听器容器，此时 Kafka 事务也将回滚。另一方面，如果 JPA 操作成功，但 Kafka 发布失败并抛出异常，则 JPA 不会提交而是回滚，并且异常将传播到侦听器。

在上面的代码中，如果我们没有与外部事务管理器同步，而只是发布到 Kafka，那么我们不需要使用@Transactional注解，我们甚至可以将代码内联到txCode方法中作为消费者lambda的一部分

@Bean
public Consumer<PersonEvent> process() {
   return pe -> {
	  Person person = new Person();
       person.setName(pe.getName());
       PersonEvent event = new PersonEvent();
       event.setName(person.getName());
       event.setType("PersonSaved");
       streamBridge.send("process-out-0", event);

   }
}

在这种情况下，我们只有当容器调用侦听器时由容器启动的 Kafka 事务。当代码通过StreamBridgesend 方法发布记录时，KafkaTemplate使用来自初始事务的相同事务生产者工厂。

这两种情况下的情况都是我们完全处于事务状态，并且事务的最终发布仅执行一次。具有read_committed隔离级别的下游消费者应该只消费一次。

Kafka Streams 和精确一次语义

在本系列中，到目前为止，我们还没有讨论 Kafka Streams。具有讽刺意味的是，最初，Kafka Streams 应用是 Apache Kafka 添加事务支持和精确一次语义的原因，但我们还没有讨论过它。原因是，在 Kafka Streams 应用中实现精确一次语义非常简单，几乎是不言而喻的。就像他们所说的，它只是一个配置旋钮。要了解 Kafka Streams 中的精确一次语义的更多信息，请参阅这篇来自 Confluent 的博客

与基于普通 Kafka 客户端的应用程序一样，在 Kafka Streams 的情况下，当您在**消费-处理-生产**模式中生成最终输出时，精确一次保证就会发挥作用，这意味着只要下游消费者使用read_committed隔离级别，它们就会正好消费一次。

Kafka Streams 配置属性processing.guarantee用于启用 Kafka Streams 应用程序中的精确一次语义。您可以在 Spring Cloud Stream 中通过设置spring.cloud.stream.kafka.streams.binder.configuration.processing.guarantee属性来设置它。您需要将值设置为exactly_once。默认情况下，Kafka Streams 使用at_least_once的值。

在有状态的 Kafka Streams 应用程序中，通常会发生以下三个主要活动：

1. 初始消费记录
2. 通过变更日志主题更新状态存储。
3. 生成数据

模式是接收并处理记录。在此过程中，任何状态信息都会具体化到状态存储中，本质上是更新特定的变更日志主题。最后，输出记录将发布到另一个 Kafka 主题。如果您注意到了这种模式，它看起来类似于我们已经看到的许多场景，除了状态存储部分。当将processing.guarantee设置为exactly_once时，Kafka Streams 保证如果在这些活动期间发生异常或应用程序崩溃，整个单元将原子回滚，就像什么也没有发生一样。应用程序重启后，处理器会再次消费记录，处理它，最后发布数据。由于此发布在后台以事务方式发生，因此隔离级别为read_committed的下游消费者在它最终发布之前不会消费该记录，从而处理实现事务性所需的所有操作（例如提交已消费记录的偏移量等），从而保证精确一次交付。

Kafka Streams 的精确一次交付保证适用于从 Kafka 相关活动角度来看的记录的端到端消费、处理和发布。当存在外部系统时，它不提供此保证。例如，假设您的代码与外部系统（例如数据库插入或更新操作）存在交互。在这种情况下，应用程序需要自行决定如何参与事务。Spring 的事务支持在这种情况下再次派上用场。我们不想在这里重复代码。但是，正如我们在此系列中多次看到的那样，您可以将与数据库交互的代码封装到一个单独的方法中，使用@Transactional注解对其进行注释，并提供适当的事务管理器，例如我们已经见过的 JPA 事务管理器。当此类方法抛出异常时，JPA 事务将回滚，并且异常会传播到 Kafka Streams 处理器代码，最终传播回 Kafka Streams 框架本身，然后回滚原始 Kafka 事务。这里再次值得重复的是，必须理解在流拓扑中从处理器调用的这些操作必须编码为处理幂等性，因为“精确一次”仅适用于整个过程，而不适用于序列中单独的读取和处理。

结论

正如我们在本文开始时提到的，**精确一次交付**语义在分布式计算中是一个复杂的话题。但是，借助 Kafka 本身提供的实现精确一次语义的解决方案以及 Spring 在 Spring for Apache Kafka 和 Spring Cloud Stream 框架中的支持，在 Spring Cloud Stream Kafka 应用程序中实现精确一次交付语义相对容易。