Spring Kafka/Spring Cloud Stream 如何保证涉及数据库和 Kafka 的事务性/原子性？

Question

Spring Kafka，因此Spring Cloud Stream，允许我们创建事务生产者和处理器。我们可以在其中一个示例项目中看到该功能的实际应用：https://github.com/spring-cloud/spring-cloud-stream-samples/tree/master/transaction-kafka-samples:

@Transactional
    @StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
    public PersonEvent process(PersonEvent data) {
        logger.info("Received event={}", data);
        Person person = new Person();
        person.setName(data.getName());

        if(shouldFail.get()) {
            shouldFail.set(false);
            throw new RuntimeException("Simulated network error");
        } else {
            //We fail every other request as a test
            shouldFail.set(true);
        }
        logger.info("Saving person={}", person);

        Person savedPerson = repository.save(person);

        PersonEvent event = new PersonEvent();
        event.setName(savedPerson.getName());
        event.setType("PersonSaved");
        logger.info("Sent event={}", event);
        return event;
    }

在这段摘录中，有一个从 Kafka 主题读取，在数据库中写入和另一个对另一个 Kafka 主题的写入，所有这些都是事务性的。

我想知道并想回答的是，这在技术上是如何实现和实施的。

由于数据源和 Kafka 不参与 XA 事务（2 阶段提交），那么实现如何保证本地事务可以从 Kafka 读取、提交到数据库并以事务方式将所有这些写入 Kafka？

Answer 1

没有保证，只有在 Kafka 内部。

Spring 提供了事务同步，所以提交很接近，但是数据库可以提交而 Kafka 没有。所以你必须处理重复的可能性。

当直接使用 spring-kafka 时，正确的方法不是使用@Transactional，而是在侦听器容器中使用ChainedKafkaTransactionManager。

见Transaction Synchronization。

另请参阅Distributed transactions in Spring, with and without XA 和“Best Efforts 1PC 模式”作为背景。

但是，对于 Stream，不支持链式事务管理器，因此需要 @Transactional（使用 DB 事务管理器）。这将为链式 tx 管理器提供类似的结果，数据库首先提交，就在 Kafka 之前。