如何实现Spring WebClient Flux的Fire-and-Forget并保留原Flux

需求目标 #

针对Flux<MyPojo>中的每个元素，以**Fire-and-Forget（即发即弃）**的方式向第三方Web服务发送携带JSON payload的HTTP请求，同时返回原Flux供后续处理。

背景信息 #

• 第三方Web服务器处理单个请求固定耗时5秒，且无法控制该服务器；
• 第三方服务器仅支持单个请求调用，无批量/列表API；
• 第三方服务器可靠性极高，允许无限制调用；
• 无需关注请求响应结果，不想等待5秒的处理耗时。

已尝试方案 #

1. 使用flatMap的实现：

private Flux<MyPojo> sendMyPojoFireAndForget(Flux<MyPojo> myPojoFlux) {
    return myPojoFlux.flatMap(oneMyPojo -> webClient.post().uri("http://example.com/path").bodyValue(oneMyPojo).exchangeToMono(clientResponse -> Mono.just(oneMyPojo)));
}

问题：并非真正的Fire-and-Forget，程序仍会等待响应返回。

2. 直接调用subscribe()的实现：

webClient.post()
            .uri("http://example.com/path")
            .bodyValue(oneMyPojo)
            .retrieve()
            .bodyToMono(Void.class)
            .subscribe();

问题：会丢失原Flux<MyPojo>，最终得到Flux<Void>，无法继续处理原数据。

核心疑问 #

如何在对Flux每个元素执行Fire-and-Forget的HTTP调用同时，完整保留原Flux用于下游处理？

用doOnNext操作符就能完美解决这个问题——它是专门处理副作用逻辑的操作符，既能触发HTTP请求，又不会修改或阻塞原Flux的数据流。

基础实现代码 #

private Flux<MyPojo> sendMyPojoFireAndForget(Flux<MyPojo> myPojoFlux) {
    return myPojoFlux.doOnNext(oneMyPojo -> 
        webClient.post()
            .uri("http://example.com/path")
            .bodyValue(oneMyPojo)
            .retrieve()
            .bodyToMono(Void.class)
            // 手动订阅触发异步请求，原Flux不会等待请求完成
            .subscribe(
                // 成功回调：因为是即发即弃，这里可以留空或仅打日志
                () -> {},
                // 异常回调：建议保留，用于排查偶发错误
                error -> log.error("发送MyPojo请求失败: {}", error.getMessage(), error)
            )
    );
}

关键细节说明 #

• doOnNext的作用：它只会在每个元素流过时执行HTTP请求逻辑，原元素会直接向下游传递，完全不会等待第三方服务器的5秒处理时间，真正实现Fire-and-Forget。
• subscribe()的作用：在doOnNext内部调用subscribe()会触发WebClient请求的异步执行，请求会在后台线程处理，不影响原数据流的流转。
• 异常处理：虽然第三方服务器可靠性高，但添加异常回调能避免静默失败，方便后续排查问题。

可选优化：控制并发请求数 #

如果担心瞬间发起过多请求占用本地资源，可以给请求指定专门的线程池，控制并发量：

private Flux<MyPojo> sendMyPojoFireAndForget(Flux<MyPojo> myPojoFlux) {
    // 创建一个专门处理即发即弃请求的线程池，限制并发数
    Scheduler fireAndForgetScheduler = Schedulers.newBoundedElastic(10, 100, "fire-and-forget-pool");
    
    return myPojoFlux.doOnNext(oneMyPojo -> 
        webClient.post()
            .uri("http://example.com/path")
            .bodyValue(oneMyPojo)
            .retrieve()
            .bodyToMono(Void.class)
            .subscribeOn(fireAndForgetScheduler)
            .subscribe(
                () -> {},
                error -> log.error("发送MyPojo请求失败: {}", error.getMessage(), error)
            )
    );
}

这里设置了最多10个并发请求，你可以根据自身服务的资源情况调整参数。

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