Schedulers.boundedElastic 似乎使用相同的线程进行处理

Question

通过查看 API，我的理解是使用 Schedulers.boundedElastic() 或 Schedulers.newBoundedElastic(3, 10, "MyThreadGroup"); 等变体；或 Schedulers.fromExecutor(executor) 允许在多个线程中处理 IO 操作。

但是使用以下示例代码进行的模拟似乎表明单个线程/同一线程正在 flatMap 中执行工作

Flux.range(0, 100)
                .flatMap(i -> {
                    try {
                        // IO operation
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("Mapping for " + i + " is done by thread " + Thread.currentThread().getName());
                    return Flux.just(i);
                })
                .subscribeOn(Schedulers.boundedElastic())
                .subscribe();

Thread.sleep(10000); // main thread

//This yields the following

Mapping for 0 is done by thread boundedElastic-1
Mapping for 1 is done by thread boundedElastic-1
Mapping for 2 is done by thread boundedElastic-1
Mapping for 3 is done by thread boundedElastic-1 ...

上面的输出向我表明相同的线程正在 flatMap 中运行。当在订阅多个 IO 时调用 flatMap 时，是否有一种方法可以让多个线程来处理项目？我期待看到boundedElastic-1、boundedElastic-2 ...。

Answer 1

1.非阻塞IO的并发（首选）

如果您有机会使用非阻塞 IO（例如 Spring WebClient），那么您无需担心线程或调度程序，并且可以立即获得并发性：

Flux.range(0, 100)
        .flatMap(i -> Mono.delay(Duration.ofMillis(500)) // e.g.: reactive webclient call
                .doOnNext(x -> System.out.println("Mapping for " + i + " is done by thread " + Thread.currentThread()
                        .getName())))
        .subscribe();

2. 阻塞IO的并发

如果可以选择的话，最好避免阻塞 IO。如果您无法避免它，您只需对代码进行轻微修改并应用于subscribeOn内部Mono：

Flux.range(0, 100)
        .flatMap(i -> Mono.fromRunnable(() -> {
            try {
                // IO operation
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("Mapping for " + i + " is done by thread " + Thread.currentThread().getName());
        }).subscribeOn(Schedulers.boundedElastic()))
        .subscribe();