为什么我的Project Reactor代码会阻塞？如何实现非阻塞执行？

问题描述 #

我是Reactive编程新手，了解到它在大数据检索时不会阻塞代码。为此我编写了如下代码：

import reactor.core.publisher.Flux;

public class ProjectReactorExamples {
    public static void main(String[] args) {
        System.out.println("started " +Thread.currentThread().getName());       
        Flux.range(1,10).subscribe(data -> print(data));
        System.out.println("Finished");
      }
    
    private static void print(long data) {      
        System.out.println("Printing "+data+"with the thread "+ Thread.currentThread().getName());     
    }
}

我期望的输出顺序是：

started main
Finished
Printing 1 with the thread ...
Printing 2 with the thread ...
...
Printing 10 with the thread ...

但实际输出总是顺序执行，Finished总是在subscribe执行完成后才显示，看起来存在阻塞。请问如何改写代码，让Finished作为第二行输出，订阅的数据在最后？

问题原因 #

你当前的代码里，Flux.range(1,10)是同步生成元素的，默认会在调用subscribe()的主线程上执行所有订阅逻辑（包括遍历1-10、调用print方法），所以主线程会被阻塞，直到所有元素处理完成才会执行后续的System.out.println("Finished")，这就是输出顺序不符合预期的核心原因。

解决方案 #

要实现异步执行订阅逻辑，让主线程先输出Finished，你需要把Flux的订阅和元素处理逻辑放到后台线程中执行，同时确保主线程不会立刻退出（否则后台守护线程会被终止，看不到print的输出）。这里推荐两种实用方法：

方法1：使用subscribeOn指定后台线程池 #

subscribeOn可以让Flux的整个订阅过程（包括元素生成、下游处理）切换到指定的线程池执行，主线程会立刻继续执行后续代码。

修改后的代码：

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

public class ProjectReactorExamples {
    public static void main(String[] args) throws InterruptedException {
        System.out.println("started " + Thread.currentThread().getName());       
        // 使用subscribeOn将订阅逻辑切换到boundedElastic线程池
        Flux.range(1,10)
            .subscribeOn(Schedulers.boundedElastic())
            .subscribe(data -> print(data));
        System.out.println("Finished");
        // 主线程短暂休眠，等待后台线程执行完成（生产环境可改用CountDownLatch等优雅同步方式）
        Thread.sleep(1000);
      }
    
    private static void print(long data) {      
        System.out.println("Printing " + data + " with the thread " + Thread.currentThread().getName());     
    }
}

方法2：用publishOn仅让下游处理异步 #

如果你只想让print方法在后台线程执行，而元素生成逻辑保留在主线程，可以使用publishOn：

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

public class ProjectReactorExamples {
    public static void main(String[] args) throws InterruptedException {
        System.out.println("started " + Thread.currentThread().getName());       
        Flux.range(1,10)
            .publishOn(Schedulers.boundedElastic()) // 下游print操作切换到后台线程
            .subscribe(data -> print(data));
        System.out.println("Finished");
        Thread.sleep(1000);
      }
    
    private static void print(long data) {      
        System.out.println("Printing " + data + " with the thread " + Thread.currentThread().getName());     
    }
}

关键说明 #

• Schedulers.boundedElastic()是Reactor推荐用于IO密集型任务的线程池，会自动管理线程数量，避免资源耗尽。
• 代码中的Thread.sleep(1000)是为了让主线程等待后台线程执行完成，实际生产环境中可以用CountDownLatch、CompletableFuture等更灵活的同步机制，避免硬编码休眠时间。
• 执行后你会看到输出顺序完全符合预期：主线程先输出started main和Finished，随后后台线程输出1-10的打印内容。

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