我已經看到了關于在 an 中間訪問通量IntegrationFlow的問題，我想知道為什么我通過以下方式成功地在通量內部撰寫邏輯：
public void writeToSolr(IntegrationFlowDefinition<?> flowDefinition) {
flowDefinition
.bridge(e -> e.reactive(flux -> a ->
flux.log("write to solr")
.flatMap(writeToSolr)
.subscribe()));
}
我首先想知道為什么我從來沒有向控制臺拋出錯誤，但是在除錯時我看到了錯誤。我還想知道它是如何作業的，為什么我需要這個a變數（它總是產生NullPointerException，即使流程可以繼續并正常作業）。當我省略a變數時：
public void writeToSolr(IntegrationFlowDefinition<?> flowDefinition) {
flowDefinition
.bridge(e -> e.reactive(flux ->
flux.log("write to solr")
.flatMap(writeToSolr)
.subscribe()));
}
我得到一個例外Bad return type in lambda expression: Disposable cannot be converted to Publisher<Message<?>>- 例如，由于型別問題，代碼無法編譯代碼。
uj5u.com熱心網友回復：
你寫的邏輯e.reactive()不正確。請參閱該端點配置器選項的檔案以了解其用途：
/**
* 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-reactive
您不能像flatMap()在此配置器中那樣執行那些“活動”操作。你完全消除了流程的目的。這flatMap必須在下游的處理程式中完成。這樣bridge沒有任何意義。它只是將當前的流動狀態變成反應性的。當handle()具有相似的reactive()配置器會做同樣的事情時，也會應用這個處理程式的一個目的——它的句柄部分。
在.subscribe()那里做是不正確的。讓框架在 Spring Integration 之上處理提供的回應流！這就是你用那個誤導自己的原因flux -> a ->。它確實編譯并讓您運行，因為它不是編譯或配置部分。當您已經發送訊息時，它確實是在運行時評估的回呼。
writeToSolr可以這樣使用：
flowDefinition
.channel(c -> c.flux())
.handle(new ReactiveMessageHandlerAdapter((message) -> writeToSolr(message.getPayload())))
我認為我們將修改reactive()端點以僅公開那些Flux僅用于配置的運算子。其余的不在當前端點配置中：必須在目標處理程式方法邏輯中完成。
另外我認為我們可以引入一個handleReactive(ReactiveMessageHandler)作為終端運算子的IntegrationFlow來簡化那個ReactiveMessageHandlerAdapter用法。
轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/451112.html