0

我无法让 spring-cloud-stream-binder-kafka 用于以下用例：

启动@transaction（Rest 控制器）数据库更新/插入发送 Kafka 消息

在事务提交之前，消费者（使用@EnableBinding 和@StreamListener 配置）能够读取记录。此使用者已经配置了 read_committed 隔离级别。

我不确定这是一个问题还是我这边的任何配置。

尝试配置 bean，ChainedTransactionManager，但有一些其他问题。

4

1 回答 1

0

以下对我有用

@Bean(name = "chainedTransactionManager")
  @Primary
  public PlatformTransactionManager chainedTransactionManager(JpaTransactionManager jpaTM,
      BinderFactory binders) {
    Binder<MessageChannel,?,?> binder = binders.getBinder("kafka", MessageChannel.class);
    if (binder instanceof KafkaMessageChannelBinder) {
      ProducerFactory<byte[], byte[]> pf =
          ((KafkaMessageChannelBinder) binder).getTransactionalProducerFactory();
      KafkaTransactionManager<byte[], byte[]> ktm = new KafkaTransactionManager<>(pf);
      ktm.setTransactionSynchronization(
          AbstractPlatformTransactionManager.SYNCHRONIZATION_ON_ACTUAL_TRANSACTION);
      return new ChainedKafkaTransactionManager<Object, Object>(jpaTM, ktm);
    } else {
      return jpaTM;
    }
  }

可以在https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/729获取更多详细信息

于 2020-03-23T05:11:34.047 回答