路由器子流的Spring集成事务

Question

我有一个名为“creationChannel”的频道，它使用 MongoMessageStore 进行备份，如下所示：

  @Bean
  ChannelMessageStore messageStore() {
    return new MongoDbChannelMessageStore(mongoDatabaseFactory);
  }

  @Bean
  PollableChannel creationChannel(ChannelMessageStore messageStore) {
    return MessageChannels.queue("creationChannel", messageStore, "create").get();
  }

我想在我的流程中使用它，但我想确定，如果“createOrderHandler”工作正常，来自那里的消息将是只读的（同样适用于“updateOrderHandler”，但使用不同的通道)。

...some code here...
        .<HeadersElement, OperationType>route(
            payload -> route(payload),
            spec -> spec
                .transactional(transactionHandleMessageAdvice)
                .subFlowMapping(
                    OperationType.New,
                    sf -> sf
                        .channel("creationChannel")
                        .log(Level.DEBUG, "Creation of a new order", Message::getPayload)
                        .transform(Mapper::mapCreate)
                        .handle(createOrderHandler,
                            handlerSpec -> handlerSpec.advice(retryOperationsInterceptor))
                )
                .subFlowMapping(
                    OperationType.Update,
                    sf -> sf
                        .channel("updateChannel")
                        .log(Level.DEBUG, "Update for existing order", Message::getPayload)
                        .transform(Mapper::mapUpdate)
                        .handle(updateOrderHandler,
                            handlerSpec -> handlerSpec.advice(retryOperationsInterceptor))
                )
        )
...some code here...

我尝试像这样配置“transactionHandleMessageAdvice”：

  @Bean
  TransactionHandleMessageAdvice transactionHandleMessageAdvice(MongoTransactionManager transactionManager) {
    return new TransactionHandleMessageAdvice(transactionManager, new Properties());
  }

但在处理程序因异常而失败后，消息仍在从数据库中删除。

也许我应该为子流配置 Poller 并以某种方式使用 MongoTransactionManager 配置它？

Answer 1

也许我应该为子流配置 Poller 并以某种方式使用 MongoTransactionManager 配置它？

这是正确的假设。只要您有线程在流中移动（例如您的PollableChannel creationChannel），当前事务就会在消息放入存储时提交。当前线程中没有任何事情发生，因此，您使用该 .transactional(transactionHandleMessageAdvice) 开始的当前事务。

要使读取事务化，您确实必须在.transform(Mapper::mapCreate) 端点上配置Poller。因此，来自该队列通道的每个轮询都将是事务性的，直到您再次转移到不同的线程。

没有办法（而且绝对不能）让整个异步流事务化，因为事务与ThreadLocal 绑定，并且在调用堆栈返回到事务发起者的那一刻，它被提交或回滚.使用异步逻辑，我们只是打算从生产者端“发送并忘记”，让消费者在数据准备好时处理数据。这不是交易的设计目的。