反应Spring集成相当于项目Reactor。doOnNext（）

Question

作为项目Reactor的用户，也想使用Spring集成，我想执行以下操作，这将以这样的方式工作：

flux.buffer(duration)
    .doOnNext(bulkWriteToCockroach())
    .doOnNext(bulkWriteToPulsar());

一开始，我认为解决方案是执行以下错误代码：

IntegrationFlow.from(myflow)
    .aggregate(myTimeBasedAggregation())
    .handle(bulkWriteToCockroach())
    .handle(bulkWriteToPulsar());

当然，由于outputChannel问题，它无法工作。我想知道如何一个接一个地执行操作（例如，在CockroachDB写入完成之前不要继续脉冲星写入，如果第一次操作失败，请停止这些消息的流）。

我正在考虑使用Spring集成事务支持，但我担心它在Retor中的使用。

我还看到有一种叫做网关（）的东西，但是我无法找到一个基于JavaDSL的使用它的可用示例。

Answer 1

查看其JavaDocs：

/**
 * Specify whether failures for one or more of the handlers should be
 * ignored. By default this is false meaning that an Exception
 * will be thrown whenever a handler fails. To override this and suppress
 * Exceptions, set the value to true.
 * @param ignoreFailures true if failures should be ignored.
 */
public void setIgnoreFailures(boolean ignoreFailures) {

使用JavaDSL，您应该查看：

/**
 * The {@link org.springframework.integration.channel.PublishSubscribeChannel} {@link #channel}
 * method specific implementation to allow the use of the 'subflow' subscriber capability.
 * @param publishSubscribeChannelConfigurer the {@link Consumer} to specify
 * {@link PublishSubscribeSpec} options including 'subflow' definition.
 * @return the current {@link BaseIntegrationFlowDefinition}.
 */
public B publishSubscribeChannel(Consumer<PublishSubscribeSpec> publishSubscribeChannelConfigurer) {

其示例可能如下所示：

                .publishSubscribeChannel(s -> s
                        .subscribe(f -> f
                                .handle(bulkWriteToCockroach()))
                        .subscribe(f -> f
                                .handle(bulkWriteToPulsar())));

请参阅文档中的更多内容：https://docs.spring.io/spring-integration/docs/current/reference/html/dsl.html#java-dsl子流

使现代化

网关（）在docs：https://docs.spring.io/spring-integration/docs/current/reference/html/dsl.html#java-dsl-gateway中解释