事务
本节描述 Spring for Apache Kafka 如何支持事务。
概述
0.11.0.0 客户端库增加了对事务的支持。Spring for Apache Kafka 通过以下方式增加支持：
KafkaTransactionManager：与常规 Spring 事务支持（@Transactional、TransactionTemplate等）一起使用- 事务性
KafkaMessageListenerContainer - 使用
KafkaTemplate的本地事务 - 与其他事务管理器的事务同步
通过为 DefaultKafkaProducerFactory 提供一个 transactionIdPrefix 来启用事务。在这种情况下，工厂不再管理单个共享的 Producer，而是维护一个事务性生产者缓存。当用户对某个生产者调用 close() 时，它会被返回到缓存以供复用，而不是真正被关闭。每个生产者的 transactional.id 属性为 transactionIdPrefix + n，其中 n 从 0 开始，并为每个新生产者递增。在早期版本的 Spring for Apache Kafka 中，对于由基于记录的监听器的监听容器启动的事务，transactional.id 的生成方式有所不同，以支持隔离僵尸生产者，但随着 3.0 起 EOSMode.V2 成为唯一选项，这已不再必要。对于运行多个实例的应用，transactionIdPrefix 必须在每个实例上唯一。
另请参阅 恰好一次语义。
另请参阅 transactionIdPrefix。
使用 Spring Boot 时，只需设置 spring.kafka.producer.transaction-id-prefix 属性即可——Spring Boot 会自动配置一个 KafkaTransactionManager bean 并将其装配到监听容器中。
从 2.5.8 版本开始，你现在可以在生产者工厂上配置 maxAge 属性。当使用可能闲置达到代理 transactional.id.expiration.ms 的事务性生产者时，这会很有用。对于当前的 kafka-clients，这可能在没有再均衡的情况下导致 ProducerFencedException。通过将 maxAge 设置为小于 transactional.id.expiration.ms，如果某个生产者超过其最大年龄，工厂将刷新该生产者。
使用 KafkaTransactionManager
KafkaTransactionManager 是 Spring Framework 的 PlatformTransactionManager 的一个实现。在其构造函数中向它提供生产者工厂的引用。如果你提供自定义生产者工厂，它必须支持事务。请参阅 ProducerFactory.transactionCapable()。
你可以将 KafkaTransactionManager 与常规 Spring 事务支持（@Transactional、TransactionTemplate 等）一起使用。如果事务处于活动状态，那么在该事务范围内执行的任何 KafkaTemplate 操作都会使用该事务的 Producer。管理器根据成功或失败提交或回滚事务。你必须将 KafkaTemplate 配置为使用与事务管理器相同的 ProducerFactory。
事务同步
本节指仅生产者事务（即并非由监听容器启动的事务）；有关容器启动事务时链接事务的信息，请参阅 使用消费者发起的事务。
如果你想向 Kafka 发送记录并执行一些数据库更新，你可以使用常规 Spring 事务管理，例如配合 DataSourceTransactionManager。
@Transactional
public void process(List<Thing> things) {
things.forEach(thing -> this.kafkaTemplate.send("topic", thing));
updateDb(things);
}
@Transactional 注解的拦截器开启事务，KafkaTemplate 会将一个事务与该事务管理器同步；每次发送都会参与该事务。当方法退出时，数据库事务先提交，随后是 Kafka 事务。如果你希望按相反顺序执行提交（先 Kafka），请使用嵌套的 @Transactional 方法，外层方法配置为使用 DataSourceTransactionManager，内层方法配置为使用 KafkaTransactionManager。
有关以 Kafka 优先或数据库优先配置同步 JDBC 与 Kafka 事务的应用示例，请参阅 Kafka 事务与其他事务管理器配合的示例。
从 2.5.17、2.6.12、2.7.9 和 2.8.0 版本开始，如果在主事务已提交之后，同步事务上的提交失败，该异常将被抛给调用者。此前，这会被静默忽略（以 debug 级别记录）。应用程序应在必要时采取补救措施，以补偿已提交的主事务。
使用消费者发起的事务
ChainedKafkaTransactionManager 从 2.7 版本起已被弃用；有关更多信息，请参阅其父类 ChainedTransactionManager 的 JavaDocs。相反，应在容器中使用 KafkaTransactionManager 来开启 Kafka 事务，并用 @Transactional 标注监听器方法以开启另一个事务。
有关链接 JDBC 与 Kafka 事务的示例应用，请参阅 Kafka 事务与其他事务管理器配合的示例。
KafkaTemplate 本地事务
你可以使用 KafkaTemplate 在本地事务中执行一系列操作。以下示例展示了如何做到这一点：
boolean result = template.executeInTransaction(t -> {
t.sendDefault("thing1", "thing2");
t.sendDefault("cat", "hat");
return true;
});
回调中的参数是模板本身（this）。如果回调正常退出，事务被提交。如果抛出异常，事务被回滚。
如果正在处理一个 KafkaTransactionManager（或同步）事务，它不会被使用。相反，会使用一个新的 "嵌套" 事务。
TransactionIdPrefix
使用 EOSMode.V2（又称 BETA，这是唯一支持的模式）时，即使对于消费者发起的事务，也不再需要使用相同的 transactional.id；事实上，与生产者发起的事务一样，它在每个实例上必须唯一。此属性在每个应用程序实例上必须具有不同的值。
TransactionIdSuffix 固定
从 3.2 开始，引入了一个新的 TransactionIdSuffixStrategy 接口来管理 transactional.id 后缀。默认实现是 DefaultTransactionIdSuffixStrategy，当将 maxCache 设置为大于零时，可以在特定范围内复用 transactional.id，否则后缀将通过递增计数器动态生成。当请求一个事务性生产者而 transactional.id 全部被占用时，抛出 NoProducerAvailableException。随后用户可以使用一个配置为重试该异常的 RetryTemplate，并配合适当配置的退避策略。
public static class Config {
@Bean
public ProducerFactory<String, String> myProducerFactory() {
Map<String, Object> configs = producerConfigs();
configs.put(ProducerConfig.CLIENT_ID_CONFIG, "myClientId");
...
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(configs);
...
TransactionIdSuffixStrategy ss = new DefaultTransactionIdSuffixStrategy(5);
pf.setTransactionIdSuffixStrategy(ss);
return pf;
}
}
将 maxCache 设置为 5 时，transactional.id 为 my.txid.+{0-4}。
将 KafkaTransactionManager 与 ConcurrentMessageListenerContainer 一起使用并启用 maxCache 时，必须将 maxCache 设置为大于或等于 concurrency 的值。如果某个 MessageListenerContainer 无法获取 transactional.id 后缀，它将抛出 NoProducerAvailableException。在 ConcurrentMessageListenerContainer 中使用嵌套事务时，必须调整 maxCache 设置以应对增加的嵌套事务数量。
KafkaTemplate 事务性与非事务性发布
通常，当 KafkaTemplate 是事务性的（配置了支持事务的生产者工厂）时，事务是必需的。事务可以由 TransactionTemplate、@Transactional 方法、调用 executeInTransaction，或在配置了 KafkaTransactionManager 时由监听容器启动。任何在事务范围之外使用该模板的尝试都会导致模板抛出 IllegalStateException。从 2.4.3 版本开始，你可以将模板的 allowNonTransactional 属性设置为 true。在这种情况下，模板将通过调用 ProducerFactory 的 createNonTransactionalProducer() 方法，允许操作在没有事务的情况下运行；该生产者将像平常一样被缓存或线程绑定以供复用。请参阅 使用 DefaultKafkaProducerFactory。
批量监听器的事务
当在使用事务时监听器失败，AfterRollbackProcessor 会被调用，以便在回滚发生后采取某些动作。对记录监听器使用默认的 AfterRollbackProcessor 时，会执行 seek，使失败的记录被重新投递。然而对于批量监听器，整个批都会被重新投递，因为框架不知道批中哪条记录失败。更多信息请参阅 回滚后处理器。
使用批量监听器时，2.4.2 版本引入了一种替代机制来处理处理批时的失败：BatchToRecordAdapter。当一个 batchListener 设置为 true 的容器工厂配置了 BatchToRecordAdapter 时，监听器一次以一条记录被调用。这使得可以在批内进行错误处理，同时仍能根据异常类型停止处理整个批。默认提供了一个 BatchToRecordAdapter，它可以配置标准的 ConsumerRecordRecoverer，例如 DeadLetterPublishingRecoverer。以下测试用例配置片段说明了如何使用此特性：
public static class TestListener {
final List<String> values = new ArrayList<>();
@KafkaListener(id = "batchRecordAdapter", topics = "test")
public void listen(String data) {
values.add(data);
if ("bar".equals(data)) {
throw new RuntimeException("reject partial");
}
}
}
@Configuration
@EnableKafka
public static class Config {
ConsumerRecord<?, ?> failed;
@Bean
public TestListener test() {
return new TestListener();
}
@Bean
public ConsumerFactory<?, ?> consumerFactory() {
return mock(ConsumerFactory.class);
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory());
factory.setBatchListener(true);
factory.setBatchToRecordAdapter(new DefaultBatchToRecordAdapter<>((record, ex) -> {
this.failed = record;
}));
return factory;
}
}