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提供了事务同步，因此提交非常紧密，但DB可以提交，而Kafka不能。所以你必须处理重复的可能性。

当直接使用spring kafka时，正确的方法不是使用事务，而是在侦听器容器中使用链式KafkatTransactionManager。

请参见事务同步。

另请参阅Spring中的分布式事务，使用和不使用XA以及“最佳努力1PC模式”作为背景。

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