Search code examples
javaspring-webfluxreactive-programmingproject-reactor

Is it possible to start Mono's in parallel and aggregate the result

I know it is possible to chain Mono's, for ex,...

Mono<String> resultAMono = loadA();
Mono<String> resultBMono = resultA.flatMap(resultA -> loadB());

This will chain and resultBMono will run when resultAMono returns....

So my question is, is it possible to start 2 Mono's in parallel and when both returns continue with another Mono?

I think it will look something like this...

Mono<String> resultAMono = loadA();
Mono<String> resuktBMono = loadB();
Mono<Tuple2<Stirng, String> tupleMono = Mono.zip(resultAMono, resultBMono);

but I have no idea this will run in Parallel or what can I do this to run in parallel...

Thx for answers....

Solution

  • 2 semantics, 1 way to make them run in parallel

    The two options I present below both need some additional tuning to make A and B Mono run in parallel: namely, each Mono should use subscribeOn(Scheduler) to get out of the common thread from where they're merged.

    If you only care about the completion of A and B

    Use when to listen for A and B completion and then to continue with a completely different Mono:

    Mono.when(monoAwithSubscribeOn, monoBwithSubscribeOn)
    .then(Mono.just("A and B finished, I don't know their value"));

    If you care about A and B values

    Use zip + map/flatMap depending on what you want to do with the result.

    Mono.zip(monoAwithSubscribeOn, monoBwithSubscribeOn)
    .map(tuple2 -> new Foo(tuple2.getT1(), tuple2.getT2(), "bar");

    or

    Mono.zip(monoAwithSubscribeOn, monoBwithSubscribeOn)
    .flatMap(tuple2 -> fetchMoreDataAsMono(tuple2.getT1(), tuple2.getT2()));

    then will ignore the previous data, so it wouldn't make much sense to use zip before it.

    also, zip will result in an empty Mono if one of A or B is empty! Use switchIfEmpty/defaultIfEmpty to protect against that case.