为何Flux.buffer().subscribe()不会立即返回？

在使用Project Reactor和Spring WebFlux时，发现订阅经过buffer处理的Flux时，subscribe()方法不会立即返回，要等到调用FluxSink.complete()后才返回。虽然当前场景需要阻塞等待完成，但惊讶这是默认行为，担心操作有误。

简化代码 #

Flux<MyPojo> flux = Flux.create(sink -> {

    MyPojo data = nextData();
    while (null != data) {

        sink.next(data);
        // 注：实际代码中此处需补充 data = nextData()，否则会触发无限循环

    }

    System.out.println("Finished reading data");
    sink.complete();
    System.out.println("Completed sink");

});

System.out.println("Subscribing...");

flux.buffer(500).subscribe((List<MyPojo> outputList) -> {

    System.out.println(String.format("Going to process %d items...", outputList.size()));
    processSomeOutput(outputList);
    System.out.println("    Processed some output");

});

System.out.println("Returned from subscribe");

实际输出 #

Subscribing...
Going to process 500 items...
    Processed some output
Going to process 500 items...
    Processed some output
Finished reading data
Going to process 99 items...
    Processed some output
Completed sink
Returned from subscribe

预期输出 #

Subscribing...
Returned from subscribe
Going to process 500 items...
    Processed some output
Going to process 500 items...
    Processed some output
Finished reading data
Going to process 99 items...
    Processed some output
Completed sink

原因解析 #

核心问题出在Flux的线程执行模型上：

• Flux.create中的数据生成逻辑是在订阅线程（即调用subscribe()的当前线程）同步执行的。当你调用subscribe()时，Reactor会直接在当前线程启动上游的数据生成循环，直到所有数据推送完成、sink.complete()调用完毕，才会结束subscribe()方法。
• buffer(500)只是攒够指定数量的数据后再向下游推送，但因为上游是同步推送，每攒够500条数据，订阅者的onNext回调会立刻在当前线程执行，整个流程完全同步阻塞，导致subscribe()无法提前返回。

实现subscribe()立即返回的方法 #

如果希望subscribe()调用后立刻返回，需要把上游的数据生成逻辑放到异步线程中执行，通过subscribeOn指定调度器即可：

Flux<MyPojo> flux = Flux.create(sink -> {
    MyPojo data = nextData();
    while (null != data) {
        sink.next(data);
        data = nextData(); // 补充更新数据，避免无限循环
    }
    System.out.println("Finished reading data");
    sink.complete();
    System.out.println("Completed sink");
})
// 指定异步调度器，让数据生成逻辑在单独线程执行
.subscribeOn(Schedulers.boundedElastic());

修改后，Flux.create的逻辑会在boundedElastic线程池的线程中异步执行，当前线程调用subscribe()后会立刻返回，后续的数据处理都在异步线程中进行，输出顺序就会符合你的预期。

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