How the extractors in Reactor blocking (in terms of performance)

user_17111269 0
Viewed 199

I've noticed that it's not recommended to use such extractors Mono.toFuture(), Flux.collectList() since they will block the flow.

I'm not very sure the 'blocking' is in which way. Like it in the code below, I know Flux.collectList() will wait for all the item finishing, will it having like a certain thread keep waiting or it's just the last thread that finishes at the last do the .collectList() thing?

It has been metion that Mono.toFuture() will block too, will it return a 'future' immediately (and the future will be usable when onNext() or onComplete() occurred), or it will return until the onNext() or the onComplete() occurred?

var m = Flux.range(0, 100)
        .parallel()
        .runOn(Schedulers.boundedElastic())
        .map(i -> Mono.fromFuture(
                Mono.just(i).map(n -> {
                    try {
                        var s = (long) (Math.random() * 100);
                        Thread.sleep(s);
                        System.out.println(Thread.currentThread() + "after " + s + "ms awaking: " + n);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return n;
                }).toFuture())
        )
        .doOnNext(o -> System.out.println(Thread.currentThread() + "before sequential"))
        .sequential();

var mm = Flux.merge(m)
        .doOnNext(o -> System.out.println(Thread.currentThread() + "before collecting"))
        .collectList()
        .doOnNext(o -> System.out.println(Thread.currentThread() + "before map"))
        .map(list -> list.stream().map(i -> i).collect(Collectors.toList()))
        .publishOn(Schedulers.single())
        .toFuture();
project-reactor java
1 Answers

Your assumptions aren't quite correct.

Mono.toFuture() isn't blocking at all - it simply returns a CompleteableFuture, which you can either block (if you call its get() method) or execute asynchronously (if you use any of its async methods, like thenApply(), thenCompose() etc.) You break out of the reactor context and so forfeit things like backpressure, but you don't immediately have to block.

It's possible you're thinking of (very) old versions of reactor where I believe there was a toFuture() variant that returned a Future, rather than a CompleteableFuture - and while that wasn't blocking either, it put you in a context where you had to then block, as Future has no async component. So while the method call itself wasn't blocking, that was then the only choice you had.

Contrary to popular belief Flux.collectList() also isn't blocking - it specifically returns a Mono<List<T>>, that is a non-blocking publisher that will emit a single element, which is a list of everything that's in that flux. You can call block() on this publisher of course, and that operation would be blocking - but calling collectList() by itself is no more blocking than any other operator.

That being said, it certainly can cause problems. Due to the nature of what it's doing (collecting all elements from a flux into a single list in memory), it may not be ideal:

  • You might have to wait a long time for the list to be emitted, with no feedback on how many elements it contains, or if it's being populated at all;
  • You might run out of memory if the number of elements, or size of elements in the flux is particularly large;
  • You can't output any intermediate state as elements are added, so you forfeit things like streaming JSON support.

That doesn't make it blocking however, it just means there's a different set of potential issues you need to weigh up before deciding whether it's an operator that's worth using in your partiuclar scenario.

user_5514060
Related