Spring cloud stream kafka在生产者端的交易

Question

我们有一个使用Kafka的Spring-Cloud-Stream应用程序。要求是在生产者端，消息列表需要放在事务的主题中。同一应用中的消息没有消费者。当我使用spring启动事务时。云流动Kafka。粘合剂交易事务id前缀，我面临的错误是调度程序没有订阅服务器，并且从主题获得的分区总数少于配置的事务。应用程序无法在事务模式下获取主题的分区。你能告诉我是否遗漏了什么吗。明天我会发布详细的日志。

谢啦

Answer 1

您需要显示您的代码和配置以及您正在使用的版本。

文档中讨论了仅生产商交易。

通过设置spring启用事务。云流动Kafka。粘合剂交易transactionIdPrefix为非空值，例如tx-。在处理器应用程序中使用时，使用者启动事务；在使用者线程上发送的任何记录都参与同一事务。当监听器正常退出时，监听器容器将向事务发送偏移量并提交它。公共生产者工厂用于使用spring配置的所有生产者绑定。云流动Kafka。粘合剂交易生产商。*财产；忽略单个绑定Kafka生产者属性。

如果希望在源应用程序中使用事务，或从某个任意线程中使用仅生产者事务（例如@Scheduled方法），则必须获取对事务生产者工厂的引用，并使用它定义KafkaTransactionManager bean。

java prettyprint-override">@Bean
public PlatformTransactionManager transactionManager(BinderFactory binders) {
    ProducerFactory<byte[], byte[]> pf = ((KafkaMessageChannelBinder) binders.getBinder(null,
            MessageChannel.class)).getTransactionalProducerFactory();
    return new KafkaTransactionManager<>(pf);
}

注意，我们使用BinderFactory获得了对binder的引用；当只配置了一个绑定器时，在第一个参数中使用null。如果配置了多个活页夹，请使用活页夹名称获取参考。一旦我们有了对绑定器的引用，我们就可以获得对ProducerFactory的引用并创建一个事务管理器。

然后，您只需要普通的Spring事务支持，例如TransactionTemplate或@Transactional，例如：

public static class Sender {

    @Transactional
    public void doInTransaction(MessageChannel output, List<String> stuffToSend) {
        stuffToSend.forEach(stuff -> output.send(new GenericMessage<>(stuff)));
    }

}

如果希望将仅生产者事务与其他事务管理器中的事务同步，请使用ChainedTransactionManager。