Spring WebFlux Reactor：高效组合多源非阻塞数据流

// 根据Foo获取一系列Bar
Flux<Bar> getBarsByFoo(Foo foo);

// 根据Bar获取一个More
Mono<More> getMoreByBar(Bar bar);

// 组合Bar、Foo和More成一个Combined对象
Combined getCombinedFrom(Bar bar, Foo foo, More more);

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

// 假设Foo, Bar, More, Combined是已定义的POJO类
class Foo {}
class Bar {}
class More {}
class Combined {}

public class ReactiveCombiner {

    // 模拟的非阻塞服务方法
    private Flux<Bar> getBarsByFoo(Foo foo) {
        System.out.println("Fetching Bars for Foo: " + foo.hashCode());
        // 实际应用中会是数据库查询或外部服务调用
        return Flux.just(new Bar(), new Bar()).delayElements(java.time.Duration.ofMillis(50));
    }

    private Mono<More> getMoreByBar(Bar bar) {
        System.out.println("Fetching More for Bar: " + bar.hashCode());
        // 实际应用中会是数据库查询或外部服务调用
        return Mono.just(new More()).delayElement(java.time.Duration.ofMillis(30));
    }

    private Combined getCombinedFrom(Foo foo, Bar bar, More more) {
        System.out.println("Combining Foo: " + foo.hashCode() + ", Bar: " + bar.hashCode() + ", More: " + more.hashCode());
        return new Combined();
    }

    /**
     * 根据Foo对象，组合生成Flux<Combined>
     *
     * @param foo 输入的Foo对象
     * @return 包含Combined对象的Flux流
     */
    public Flux<Combined> getCombinedByFoo(Foo foo) {
        // 1. 获取Bars的Flux流
        Flux<Bar> bars = getBarsByFoo(foo);

        // 2. 对每个Bar，使用flatMap进行进一步的非阻塞操作
        Flux<Combined> result = bars.flatMap(bar -> {
            // 对于每个Bar，获取对应的More对象（返回Mono<More>）
            Mono<More> nextMore = getMoreByBar(bar);

            // 当More对象可用时，将其与原始的Foo和Bar组合成Combined
            Mono<Combined> nextCombined = nextMore.map(more -> getCombinedFrom(foo, bar, more));

            // 返回这个Mono<Combined>，flatMap会将其扁平化到主Flux流中
            return nextCombined;
        });

        return result;
    }

    // 如果Foo本身也是通过Mono获取的
    public Flux<Combined> getCombinedFromMonoFoo(Mono<Foo> monoFoo) {
        return monoFoo.flatMapMany(this::getCombinedByFoo);
    }

    public static void main(String[] args) {
        ReactiveCombiner combiner = new ReactiveCombiner();
        Foo myFoo = new Foo();

        System.out.println("--- Starting combination for a single Foo ---");
        combiner.getCombinedByFoo(myFoo)
                .doOnNext(c -> System.out.println("Received Combined object: " + c.hashCode()))
                .blockLast(); // 阻塞等待所有结果，仅用于示例

        System.out.println("\n--- Starting combination for a Mono<Foo> ---");
        Mono<Foo> monoMyFoo = Mono.just(new Foo());
        combiner.getCombinedFromMonoFoo(monoMyFoo)
                .doOnNext(c -> System.out.println("Received Combined object from Mono<Foo>: " + c.hashCode()))
                .blockLast(); // 阻塞等待所有结果，仅用于示例
    }
}

// 限制同时进行的getMoreByBar(bar)操作最多为4个
Flux<Combined> result = bars.flatMap(bar -> {
    Mono<More> nextMore = getMoreByBar(bar);
    Mono<Combined> nextCombined = nextMore.map(more -> getCombinedFrom(foo, bar, more));
    return nextCombined;
}, 4); // 指定并发度为4

public Flux<Combined> getCombinedByFooSequentially(Foo foo) {
    Flux<Bar> bars = getBarsByFoo(foo);

    // 使用concatMap确保每个Bar的处理是顺序的
    Flux<Combined> result = bars.concatMap(bar -> {
        Mono<More> nextMore = getMoreByBar(bar);
        Mono<Combined> nextCombined = nextMore.map(more -> getCombinedFrom(foo, bar, more));
        return nextCombined;
    });

    return result;
}