如何将阻塞式代码转为响应式风格？WebFlux性能问题求助

问题核心 #

你的代码存在几个关键问题，导致WebFlux下IO线程被阻塞、性能暴跌：

1. 阻塞操作（Thread.sleep）直接在WebFlux的IO线程执行，占用核心处理线程，无法应对高并发。
2. subscribeOn放置位置错误，没有作用到阻塞逻辑上，线程调度完全失效。
3. 手动调用subscribe()脱离了响应式框架的线程管理逻辑，破坏了流的编排。

修正方案 #

关键原则 #

• 所有阻塞IO操作必须包装到**Schedulers.boundedElastic()**线程池执行，这是Reactor专门为阻塞操作设计的线程池，避免占用IO线程。
• 用响应式操作符替代手动subscribe()，让框架统一管理流的执行和线程调度。
• 合理选择操作符：无需等待结果用doOnNext+fromRunnable，需要等待结果用flatMap+fromCallable。

修正后代码 #

import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.util.ArrayList;
import java.util.List;

public class ReactiveExample {

    public Mono<List<String>> a() {
        // 同步生成列表（CPU密集操作，无需切换线程）
        List<String> strings = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            strings.add("hello " + i);
        }
        Mono<List<String>> resultMono = Mono.just(strings);

        // 处理b()：fire-and-forget场景，无需等待完成
        resultMono = resultMono.doOnNext(ignored -> 
            Mono.fromRunnable(this::b)
                .subscribeOn(Schedulers.boundedElastic())
                .subscribe()
        );

        // 处理c()：需要等待执行完成，用flatMap串联流
        resultMono = resultMono.flatMap(list -> 
            Mono.fromCallable(this::c)
                .subscribeOn(Schedulers.boundedElastic())
                .thenReturn(list) // 执行完c后返回原列表
        );

        return resultMono;
    }

    // 模拟耗时阻塞操作
    public void b() {
        try {
            Thread.sleep(100);
        } catch (InterruptedException err) {
            throw new RuntimeException(err);
        }
    }

    // 模拟外部HTTP请求的阻塞操作
    public int c() {
        try {
            Thread.sleep(300);
        } catch (InterruptedException err) {
            throw new RuntimeException(err);
        }
        return 1;
    }
}

关键细节说明 #

• 线程池选择：Schedulers.boundedElastic()会根据负载动态创建线程，空闲线程会自动回收，避免线程泄漏；CPU密集操作优先用Schedulers.parallel()。
• 操作符差异：
  • doOnNext是副作用操作，不改变流的内容，适合执行不需要等待的异步任务（比如日志、通知）。
  • flatMap用于串联异步操作，会等待内部Mono完成后再继续流的执行，适合需要依赖返回值的场景。
• subscribeOn的正确用法：必须将subscribeOn绑定到包含阻塞操作的Mono上，它只会影响当前Mono的上游执行线程，放在整个流的末尾无效。

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