单CPU环境下Spring WebFlux线程模型问题咨询

你的核心问题在于**map操作会在订阅者所在的线程（此处为reactor-http-nio-1）上同步执行**，而单CPU环境下Spring WebFlux的IO线程池只有1个线程。你的heavyComputation是CPU密集型、单条耗时5秒的任务，会完全占用这个唯一的线程，导致所有Flux中的元素只能串行处理，无法实现你期望的并发。

虽然heavyComputation没有IO阻塞，但它是纯CPU计算任务，会持续占用线程资源，让WebFlux的IO线程无法处理其他任务，自然就没有并发空间。

要实现CPU密集型任务的并发处理，你需要将计算任务切换到专门的线程池执行，避免阻塞WebFlux的IO线程。推荐两种方式：

方式1：使用publishOn切换线程上下文 #

publishOn可以指定后续操作（比如map）在指定的调度器线程池中执行。对于CPU密集型任务，优先使用Schedulers.parallel()（线程数默认等于CPU核心数），如果单CPU下需要更多并发（利用时间片轮转），可以自定义线程池。

代码示例 #

import reactor.core.scheduler.Schedulers;

@GetMapping(value = "/upload-flux", consumes = MediaType.MULTIPART_FORM_DATA_VALUE, produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> question(@RequestPart("question") Flux<String> stringFlux) {
    return stringFlux
            // 切换到CPU密集型线程池执行后续map操作
            .publishOn(Schedulers.parallel())
            .map(this::heavyComputation);
}

private String heavyComputation(String oneString) {
    // 耗时的内存解密、对象转换、哈希校验、加密等操作
    return encryptedString;
}

如果单CPU下需要更高的并发度（比如允许4个任务交替执行），可以自定义线程池：

import java.util.concurrent.Executors;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

// 自定义CPU密集型线程池，单CPU环境下可适当设置线程数
private static final Scheduler CPU_INTENSIVE_SCHEDULER = Schedulers.fromExecutor(
        Executors.newFixedThreadPool(4)
);

@GetMapping(value = "/upload-flux", consumes = MediaType.MULTIPART_FORM_DATA_VALUE, produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> question(@RequestPart("question") Flux<String> stringFlux) {
    return stringFlux
            .publishOn(CPU_INTENSIVE_SCHEDULER)
            .map(this::heavyComputation);
}

方式2：使用flatMap结合Mono.fromCallable异步执行 #

flatMap可以同时订阅多个Mono，实现并发处理，配合Mono.fromCallable将同步计算任务包装为异步操作，并指定调度器。这种方式还能通过flatMap的第二个参数控制并发数，避免线程池过载。

代码示例 #

@GetMapping(value = "/upload-flux", consumes = MediaType.MULTIPART_FORM_DATA_VALUE, produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> question(@RequestPart("question") Flux<String> stringFlux) {
    return stringFlux
            .flatMap(oneString -> 
                Mono.fromCallable(() -> heavyComputation(oneString))
                    .subscribeOn(Schedulers.parallel())
            , 4); // 控制并发数为4，可根据单CPU性能调整
}

• 不要在WebFlux的IO线程（reactor-http-nio-*）上执行CPU密集型任务，否则会阻塞整个服务的请求处理。
• CPU密集型任务优先使用Schedulers.parallel()或自定义固定线程池，避免使用Schedulers.boundedElastic()（该调度器针对IO密集型任务，线程数会动态扩容，不适合CPU密集场景）。
• 单CPU环境下的“并发”是通过线程时间片轮转实现的，无法做到真正的并行，但能避免任务串行等待，提升整体处理效率。

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