如何在使用subscribe的异步响应式WebClient请求中返回Flux且不使用block

你遇到的编译错误本质是响应式编程的用法错误：

• subscribe是触发流订阅的终端操作，返回值为Disposable类型，不允许返回自定义的流对象，也不应该在业务处理链中间调用
• flatMap算子要求你返回Publisher类型的响应式流对象，你需要把HTTP调用、数据转换的整个逻辑串成完整的响应式链，而不是手动订阅中间流

直接移除subscribe调用，将数据转换逻辑拼接在HTTP调用返回的流之后作为后续处理环节，再将整个流作为flatMap的返回值即可，修改后代码如下：

public Flux<TargeData> getData(Flux<Message<EventInput>> message) 
{
    // 建议将WebClient单例注入，不要每次请求都创建实例，此处保留原有逻辑仅做用法修正
    WebClient client = WebClient.create();
    return message
        .flatMap(it -> {
            Event event = objectMapper.convertValue(it.getPayload(), Event.class);
            String eventType = event.getHeader().getEventType();
            if (DISTRIBUTOR.equals(eventType)) {
                String callBackURL = event.getHeader().getCallbackEnpoint();
                return client.get()
                    .uri(callBackURL)
                    .headers(httpHeaders -> {
                        httpHeaders.setContentType(MediaType.APPLICATION_JSON);
                        httpHeaders.setAccept(List.of(MediaType.APPLICATION_JSON));
                    })
                    .exchangeToFlux(response -> {
                        if (response.statusCode().equals(HttpStatus.OK)) {
                            System.out.println("Response is OK");
                            return response.bodyToFlux(NodeInput.class);
                        }
                        return Flux.empty();
                    })
                    // 直接在流上做类型转换，无需手动订阅
                    .map(nodeInput -> objectMapper.convertValue(nodeInput, SourceData.class))
                    // 适配transform方法返回的Iterable类型，展开为多元素流
                    .flatMapIterable(source -> this.TransformImpl.transform(source));
            }
            return Flux.empty();
        });
}

• 不要在方法内每次创建WebClient实例，Spring Boot已经自动装配了WebClient.Builder，注入后构建单例WebClient复用即可，减少连接创建开销
• 可以补充HTTP调用异常、状态码非200、类型转换失败的降级处理逻辑，避免异常直接中断整个流
• 生产环境建议使用日志框架代替System.out打印，方便问题排查

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