Spring Webflux：启动后台Mono长任务并立即返回HTTP响应

你遇到的问题非常典型：既要快速返回HTTP响应，又要让长时任务在后台异步执行，同时还要契合响应式编程的最佳实践，避免显式调用subscribe()带来的警告和潜在问题。

核心思路 #

响应式流是冷流，只有被订阅时才会触发执行。我们的目标是：

1. 快速生成并返回201 Created响应
2. 在后台触发长时任务的执行，不阻塞响应的返回
3. 避免在WebFlux的IO线程上执行长时任务，防止线程资源耗尽

优雅的实现方式 #

这里推荐使用doOnSuccess操作符触发后台任务，同时利用Schedulers.boundedElastic()将长时任务调度到专门的线程池，并妥善处理任务中的错误（避免静默失败）：

import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import reactor.core.scheduler.Schedulers
import org.springframework.http.ResponseEntity
import org.springframework.web.bind.annotation.PutMapping
import org.springframework.web.bind.annotation.PathVariable
import org.springframework.web.server.ServerHttpRequest

@PutMapping("/{jobId}")
fun startNewJob(@PathVariable("jobId") jobId: String, request: ServerHttpRequest): Mono<ResponseEntity<Unit>> {
    // 提前生成Job的URI（假设传入的jobId就是最终Job的ID，无需等待任务完成）
    val jobUri = generateJobUri(request, jobId)
    val response = ResponseEntity.created(jobUri).build<Unit>()
    val logger = LoggerFactory.getLogger(javaClass)

    return Mono.just(response)
        .doOnSuccess {
            // 在专门的弹性线程池上执行长时任务，避免阻塞WebFlux的IO线程
            startNewJob(jobId)
                .subscribeOn(Schedulers.boundedElastic())
                .subscribe(
                    { completedJob -> 
                        // 可选：任务成功后的回调，比如记录日志
                        logger.info("Long-running job {} completed successfully", completedJob.id)
                    },
                    { error ->
                        // 务必处理任务错误，防止异常丢失
                        logger.error("Long-running job {} failed", jobId, error)
                    }
                )
        }
}

关键细节说明 #

1. 为什么用doOnSuccess？
   这个操作符会在响应流成功发出信号后执行副作用逻辑，刚好符合我们"返回响应后再启动后台任务"的需求。

2. 为什么用Schedulers.boundedElastic()？
   WebFlux默认的IO线程专为非阻塞操作设计，不适合执行长时阻塞任务。boundedElastic()会创建有界线程池，专门处理这类任务，避免线程资源耗尽。

3. 为什么还是要调用subscribe()？
   因为startNewJob(jobId)返回的是冷流，只有订阅才会触发执行。在这个场景下，我们不需要将后台任务的结果纳入响应流，所以显式订阅是必要的。IntelliJ的警告只是提醒你注意非阻塞环境中的订阅操作，但只要我们正确处理错误并使用合适的调度器，这种写法是符合响应式实践的。

4. 关于Job URI的生成
   如果你的startNewJob会生成新的Job ID（而非使用传入的jobId），可以先生成临时URI，或者在后台任务完成后通过WebSocket、Webhook等方式通知客户端。通常推荐在请求时就确定Job的唯一标识，这样可以直接返回有效的Location头。

替代方案：避免显式subscribe() #

如果你确实想完全避免显式调用subscribe()，可以将后台任务与响应流并行组合，确保响应流不会等待后台任务完成：

@PutMapping("/{jobId}")
fun startNewJob(@PathVariable("jobId") jobId: String, request: ServerHttpRequest): Mono<ResponseEntity<Unit>> {
    val jobUri = generateJobUri(request, jobId)
    val response = ResponseEntity.created(jobUri).build<Unit>()
    val logger = LoggerFactory.getLogger(javaClass)

    // 后台任务流：调度到弹性线程池，处理错误后忽略结果
    val backgroundJob = startNewJob(jobId)
        .subscribeOn(Schedulers.boundedElastic())
        .doOnError { logger.error("Job {} failed", jobId, it) }
        .then()

    // 并行执行响应流和后台任务流，立即返回响应
    return Mono.just(response)
        .flatMap { backgroundJob.thenReturn(it) }
}

这个写法通过flatMap和thenReturn隐式订阅了后台任务流，没有显式调用subscribe()，同时响应会立即返回，后台任务在后台继续执行。

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