如何在Project Reactor的Flux中并行执行元素处理函数？

你遇到的核心问题是：虽然用了Mono.zip，但两个阻塞操作默认在同一个线程执行，导致串行。要让setAge和setName真正并行，需要给每个耗时操作指定独立的调度器，避免阻塞线程。

问题原因分析 #

你的setAge方法里用了Thread.sleep(10000)这种阻塞操作，而默认情况下Mono.fromSupplier会在当前订阅线程执行。如果两个Mono都在同一个线程执行，就会出现先跑完setAge再跑setName的串行情况，完全达不到并行效果。

解决方案 #

1. 给阻塞方法添加调度器

修改setAge和setName方法，用subscribeOn(Schedulers.boundedElastic())将阻塞操作放到弹性线程池执行——这个调度器专门用来处理阻塞IO或耗时计算，不会占用其他业务线程：

private static Mono<Dog> setAge(Dog dog) {
    return Mono.fromSupplier(() -> {
        log.info("Setting age "+dog);
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        dog.age = 10;
        return dog;
    }).subscribeOn(Schedulers.boundedElastic()); // 关键：指定阻塞操作的调度器
}

// setName方法做同样修改
private static Mono<Dog> setName(Dog dog) {
    return Mono.fromSupplier(() -> {
        log.info("Setting name "+dog);
        try {
            Thread.sleep(5000); // 模拟耗时操作
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        dog.name = "Buddy";
        return dog;
    }).subscribeOn(Schedulers.boundedElastic());
}

2. 调整Flux处理逻辑

保持Mono.zip的结构（它本身就会并行订阅两个Mono），同时简化错误处理逻辑，确保两个操作完成后返回原对象：

Flux.fromIterable(mapToResult())
    .parallel(10)
    .runOn(Schedulers.parallel())
    .flatMap(result -> 
        Mono.zip(
            setAge(result)
                .onErrorResume(err -> {
                    err.printStackTrace();
                    return Mono.just(result);
                }),
            setName(result)
                .onErrorResume(err -> {
                    err.printStackTrace();
                    return Mono.just(result);
                })
        ).map(tuple -> result) // 两个操作都修改了同一个Dog实例，直接返回即可
    )
    .subscribe();

替代方案：用Mono.when简化逻辑

如果不需要获取两个操作的返回值，只需要等待两者完成，可以用Mono.when替代Mono.zip，代码更简洁：

Flux.fromIterable(mapToResult())
    .parallel(10)
    .runOn(Schedulers.parallel())
    .flatMap(result -> 
        Mono.when(
            setAge(result).onErrorResume(err -> {
                err.printStackTrace();
                return Mono.empty();
            }),
            setName(result).onErrorResume(err -> {
                err.printStackTrace();
                return Mono.empty();
            })
        ).thenReturn(result)
    )
    .subscribe();

注意事项 #

• 因为两个操作修改的是同一个Dog实例，要确保两个方法的字段修改没有线程安全冲突（比如不要同时修改同一个字段）。
• Schedulers.boundedElastic()是处理阻塞操作的首选，不要用Schedulers.parallel()——后者是给CPU密集型操作准备的，阻塞操作会耗尽它的线程池。

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