Mono.just() is still blocking in my use case when the emitted element is pending

Question

I have two test that use sleep() to simulate an api call which takes time to process, and test if Mono.just() make it non-blocking.

In my first test, I emitted a String directly, but made it blocking in map().

    @Test
    public void testMono() throws InterruptedException {

        for (int i = 0; i < 10; i++) {
            Mono.just("Hello World")
                    .map(s -> {
                        System.out.println("Thread:" + Thread.currentThread().getName() + " " + s);
                        try {
                            sleep(2000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        return s.length();
                    })
                    .map(integer -> {
                        try {
                            sleep(3000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        return integer.toString();
                    })
                    .subscribeOn(Schedulers.parallel())

                    .subscribe(System.out::println);

            System.out.println("Loop: " + i + " " + "after MONO");
        }

        sleep(10000);
    }

And the result is expected as non-blocking, since all the outputs System.out.println("Loop: " + i + " " + "after MONO"); show up together at the same time.

However, in the second test, I replace the emitted element from "Hello world" to a blocking getString() method. The purpose of put sleep() in getString() is that, I'd like to simulate the scenario that it's taking time to get the emitted element. And the result is now blocking since the output shows up one by one after it finished receiving the emitted element.

    @Test
    public void testMono2() throws InterruptedException {

        for (int i = 0; i < 10; i++) {
            Mono.just(getString())
                    .map(s -> {
                        System.out.println("Thread:" + Thread.currentThread().getName() + " " + s);
                        try {
                            sleep(2000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        return s.length();
                    })
                    .map(integer -> {
                        try {
                            sleep(3000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        return integer.toString();
                    })
                    .subscribeOn(Schedulers.parallel())

                    .subscribe(System.out::println);

            System.out.println("Loop: " + i + " " + "after MONO");
        }

        sleep(10000);
    }

    public String getString(){
        try {
            sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "Hello world";
    }

And in the third test, it's also acting as non-blocking.

    int count = 0;

    @Test
    public void testFluxSink() throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            System.out.println("Start Loop: " + i);
            int curLoop = count++;
            Flux.create(fluxSink -> {
                try {
                    System.out.println("Loop: " + curLoop + " " + " start sleeping");
                    sleep(5000);
                    System.out.println("Loop: " + curLoop + " " + " end sleeping");
                    fluxSink.next(onNext());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).subscribeOn(Schedulers.parallel()).subscribe(s -> {
                System.out.println(Thread.currentThread());
                System.out.println("Loop: " + curLoop + " " + "completed receiving" + " " + s);
            });
        }
        sleep(100000);
    }

    public String onNext() throws InterruptedException {
        sleep(5000);
        return "onNext";
    }

Wondering if I misunderstand the concept of reactor, and would like to know how did I use it in the wrong way for the 2nd test?

Answer 1

First of all, it is crucial to understand that Reactor does not perform any magical conversion of blocking calls to non-blocking ones. To make an end-to-end non-blocking application, you need to use non-blocking drivers, NIO, etc. If you wrap your blocking code to the Mono or Flux, the thread is still will be blocked.

As for your second example, with sleep in the getString method. There are two points to pay attention at:

The first one: you use Mono.just. The important thing is that it is usually used to provide a known value immediately. It means that the getString calculation is happening not in the scope of the reactor chain, but in the main thread during the reactor chain assembling phase. That's why you see "sequential" behavior. If you replace it with Mono.fromCallable, then the calculation will be in the reactor chain (parallel scheduler threads), and you will see the same behavior as in the 1 or 3 examples.

The second one: It's important to note, that wrapping your getString method to the Mono.fromCallable does not make your code non-blocking. Your sleep still halts the threads of the parallel scheduler. In your production code, your getString method will likely be some database call or other service call that should be done via a non-blocking driver or NIO-based lib like Netty. To emulate it use delayElement instead of sleep, it works in a non-blocking way. Here is an example:

public static void main(String[] args) throws InterruptedException {
    for (int i = 0; i < 3; i++) {
        getString()
            .map(s -> {
                System.out.println(Instant.now() + " Thread:" + Thread.currentThread().getName() + " " + s);
                return s.length();
            })
            .subscribeOn(Schedulers.parallel())
            .subscribe();
        System.out.println(Instant.now() + " Loop: " + i + " " + "after MONO");
    }

    TimeUnit.SECONDS.sleep(10);
}

public static Mono<String> getString() {
    return Mono.delay(Duration.ofSeconds(3))
        .map(it -> "hello world");
}

It prints

2023-05-17T20:31:12.099464Z Loop: 0 after MONO
2023-05-17T20:31:12.112953Z Loop: 1 after MONO
2023-05-17T20:31:12.113145Z Loop: 2 after MONO
2023-05-17T20:31:15.105750Z Thread:parallel-2 hello world
2023-05-17T20:31:15.117944Z Thread:parallel-5 hello world
2023-05-17T20:31:15.117944Z Thread:parallel-6 hello world