如何在Spring Integration中订阅响应式单向处理器返回的Mono<Void>

Mono的自动订阅逻辑 #

Spring Integration 5.0及以上版本原生支持响应式组件，当你在.handle()中传入返回Publisher（包括Mono<Void>）的实现，或是使用ReactiveMessageHandler类型的适配器（比如MongoDb.reactiveOutboundChannelAdapter）时，框架会自动完成对Publisher的订阅，无需手动操作。你看到MongoDB成功写入数据就是自动订阅生效的结果，订阅过程会绑定到当前消息的处理生命周期，异常会默认向上传递到消息对应的错误通道。

订阅完成信号、获取操作回执的实现 #

你之前认为返回Mono<Void>的处理器一定是流终点属于常见误解，你可以通过响应式handle的重载方法，在操作完成后生成回执消息向下游传递，示例代码如下：

@Bean
public IntegrationFlow flow(AdapterThatReturnsMonoVoid adapter) {
    return f -> f
     // 原有转换、路由等逻辑
     .handleReactive((message, context) -> 
         // 执行原有响应式操作，返回Mono<Void>
         adapter.handleMessage(message)
             // 操作完成后生成回执消息，可自定义payload和headers
             .then(Mono.just(MessageBuilder
                 .withPayload("操作执行完成")
                 .copyHeaders(message.getHeaders())
                 .build()))
     )
     // 下游可直接订阅回执消息做后续处理
     .channel("operationAckChannel")
     .handle(yourAckService::processAck);
}

如果你使用的是MongoDB等官方提供的响应式出站适配器，也可以用同样的逻辑包装：

@Bean
public IntegrationFlow mongoFlow(ReactiveMongoTemplate mongoTemplate) {
    return f -> f
        .handleReactive((message, context) -> 
            mongoTemplate.save(message.getPayload(), "your_collection")
                .then(Mono.just(MessageBuilder
                    .withPayload("MongoDB写入完成")
                    .setHeader("originalPayload", message.getPayload())
                    .build()))
        )
        .channel("mongoOperationAckChannel");
}

如果你不希望修改主流程的原有逻辑，也可以通过wireTap旁路监听操作结果，在旁路流中处理回执逻辑，不会对主流程的执行产生影响。

内容的提问来源于stack exchange，提问作者Gonzalo