Reactor中如何切换执行线程？代码示例疑问求解

先聊聊你代码里的问题：你现在直接调用Flux.fromIterable(data).subscribe(...)，Reactor默认会在调用subscribe的当前线程（也就是处理HTTP请求的Spring Web线程）上执行所有流操作，所以所有printStatement的调用都会复用同一个线程，自然线程ID不会变化。

要实现线程切换，Reactor提供了publishOn和subscribeOn两个核心方法，配合Schedulers调度器就能指定线程池执行任务。下面给你具体的解决方案：

核心方案：用调度器切换线程 #

针对你的场景（需要在异步线程中执行阻塞的printStatement方法），最适合的调度器是Schedulers.boundedElastic()——它专门用来处理阻塞/IO密集型任务，会根据需求动态创建线程（默认上限1000），避免阻塞主线程或其他关键线程池。

修改后的代码示例 #

@RequestMapping(value = "/try", method = RequestMethod.GET)
@ResponseBody
public String demo(){
    List<String>data=new ArrayList<>();
    data.add("A1");
    data.add("A2");
    data.add("A3");
    data.add("A4");
    
    // 使用publishOn指定后续操作的执行线程池
    Flux.fromIterable(data)
        .publishOn(Schedulers.boundedElastic())
        .subscribe(this::printStatement);
        
    return "done";
}

public void printStatement(String s){
    long i;
    for(i=0;i<1000000000;i++) {}
    LOGGER.info(s+"------"+Thread.currentThread().getId());
}

修改后，每个printStatement的执行都会被提交到boundedElastic线程池的线程中，你会看到不同的线程ID（线程池会复用空闲线程，但大概率每个元素的执行线程不同）。

区分publishOn和subscribeOn #

很多人会混淆这两个方法，这里简单梳理下区别：

• subscribeOn：指定整个流的订阅触发线程，也就是流的源头（比如fromIterable）和所有上游操作的执行线程，整个流只会受一次subscribeOn影响（多次调用只有第一个生效）。
  你也可以这样写，效果和上面类似：

  Flux.fromIterable(data)
      .subscribeOn(Schedulers.boundedElastic())
      .subscribe(this::printStatement);
  

• publishOn：指定下游操作的执行线程，也就是说在publishOn之后的所有操作都会切换到指定线程池执行，你可以多次调用publishOn来切换不同阶段的线程。

注意事项 #

• 不要用Schedulers.parallel()处理阻塞任务：这个调度器是给CPU密集型任务设计的，线程数固定为CPU核心数，阻塞操作会占满线程池，影响其他任务的执行。
• 你的代码中subscribe是异步执行的：返回"done"的时候，printStatement可能还在后台运行，这是正常的异步行为；如果需要等待所有任务完成再返回，建议用Mono/Flux作为接口返回值，让Spring Web自动处理异步逻辑，避免阻塞主线程。

内容的提问来源于stack exchange，提问作者Rakshith R Pai