积分流中的反应Spring积分接入通量

Question

我已经看到了在IntegrationFlow中间访问通量的问题，我想知道为什么我用以下方式成功地在通量中编写逻辑：

public void writeToSolr(IntegrationFlowDefinition<?> flowDefinition) {
    flowDefinition
        .bridge(e -> e.reactive(flux -> a ->
           flux.log("write to solr")
               .flatMap(writeToSolr)
               .subscribe()));
}

首先，我想知道为什么我从未将错误抛出控制台，但在调试时我看到了错误。我还想知道这是如何工作的，为什么我需要a变量（它总是产生NullPointerException，即使流可以继续并正常工作）。当我省略变量时：

public void writeToSolr(IntegrationFlowDefinition<?> flowDefinition) {
    flowDefinition
        .bridge(e -> e.reactive(flux ->
           flux.log("write to solr")
               .flatMap(writeToSolr)
               .subscribe()));
}

我在lambda表达式中得到了一个异常错误的返回类型：无法将Dispossible转换为Publisher




Answer 1

您在该e.reactive（）中编写的逻辑不正确。请参阅留档以了解endpoint配置程序选项的用途：

/**
 * Make the consumer endpoint as reactive independently of an input channel and
 * apply the provided function into the {@link Flux#transform(Function)} operator.
 * @param reactiveCustomizer the function to transform {@link Flux} for the input channel.
 * @return the spec
 * @since 5.5
 */
public S reactive(Function<? super Flux<Message<?>>, ? extends Publisher<Message<?>>> reactiveCustomizer) {

https://docs.spring.io/spring-integration/docs/current/reference/html/dsl.html#java-dsl反应

在此配置器中，您无法执行诸如flatMap（）之类的“活动”操作。您完全消除了流的目的。必须在下游处理程序中完成flatMap。这样的桥梁没有任何意义。它只是将当前的流动状态转变为无功状态。当handle（）与类似的reactive（）配置程序将执行相同的操作，但也将应用此处理程序的一个用途-其句柄部分。

。subscribe（）在那里完全不正确。让框架在Spring集成的基础上处理提供的反应流！这就是为什么你用flux来误导自己-

writeSolr可以这样使用：

flowDefinition
    .channel(c -> c.flux())
    .handle(new ReactiveMessageHandlerAdapter((message) -> writeToSolr(message.getPayload())))

我想我们将修改endpoint的reactive（），只公开那些仅用于配置的Flux运算符。其余的不在当前endpoint配置中：必须在目标处理程序方法逻辑中完成。

此外，我认为我们可以引入handleReactive（ReactiveMessageHandler）作为IntegrationFlow的终端操作符，以简化ReactiveMessageHandlerAdapter的用法。