CODEWITHSUNDEEP
spring-webfluxproject-reactor
Ben
Ben

Reputation: 361

Swapping between Schedulers in Reactor

I have a blocking workload that I want to execute on the bounded elastic scheduler. After this work is done, a lot of work that could be executed on the parallel scheduler follows, but it will automatically continue to run on the thread from the bounded elastic scheduler.

When is it "correct" to drop the previous scheduler you set earlier in the chain? Is there ever a reason to do so if it's not strictly necessary, because of thread starvation, for example?

I can switch the scheduler of a chain by "breaking" the existing chain with flatMap, then, switchIfEmpty, and probably a few more methods. Example:

public Mono<Void> mulpleSchedulers() {
    return blockingWorkload()
            .doOnSuccess(result -> log.info("Thread {}", Thread.currentThread().getName())) // Thread boundedElastic-1
            .subscribeOn(Schedulers.boundedElastic())
            .flatMap(result -> Mono.just(result)
                    .subscribeOn(Schedulers.parallel()))
            .doOnSuccess(result -> log.info("Thread {}", Thread.currentThread().getName())); // Thread parallel-1
}

Upvotes: 1

Views: 1986

Answers (1)

kerbermeister
kerbermeister

Reputation: 4251

Generally it is not a bad practice to switch the execution to another scheduler throughout your reactive chain.

To switch the execution to another scheduler in the middle of your chain you can use publishOn() operator. Then any subsequent operator call will be run on the supplied scheduler worker to such publishOn().

Take a look at 4.5.1 from reference

However, you should clearly know why do you do that. Is there any reason for this?

If you want to run some long computational process (some CPU-bound work), then it is recommended to execute it on Schedulers.parallel()

For making blocking calls it is recommended to use Schedulers.boundedElastic()

However, usually for making blocking calls we use subscribeOn(Schedulers.boundedElastic()) on "blocking publisher" directly. Wrapping blocking calls in reactor

Upvotes: 2

Related Questions