Spring Webflux / Reactor release calling thread

Question

So I understand that under the hood Spring WebFlux and Reactor uses netty for nio, now I would like to release the calling thread to free up the resources in order to handle more requests. Is the below simple code releasing the calling thread?

@GetMapping("/foo")
public Mono<Void> bar() {

  someService.veryLongSyncOperation();

  return Mono.empty();
}

I did not wrap the service call in a Flux/Mono, I just want to first verify that the calling thread is released while the service does its long work. Is this enough to achieve calling thread release? If so is there a way to test this?

I was thinking that the framework sees the return type and that's enough for it to know that it must release the calling thread.

Alexander Pankin · Accepted Answer · 2019-11-16T12:24:04.657

4

You could use .subscribeOn(Schedulers.elastic()) as it is mentioned in the reactor reference guide

@GetMapping("/foo")
public Mono<Void> bar() {
    return Mono.fromCallable(() -> someService.veryLongSyncOperation())
            .subscribeOn(Schedulers.elastic())
            .then();
}

each subscription will happen on a dedicated single-threaded worker from Schedulers.elastic().

Upd: There is Schedulers.boundedElastic() scheduler now. I'd recommend using it by default.

edited Nov 16 '19 at 12:24

answered Mar 09 '19 at 20:21

Alexander Pankin

3,787
1
13
23

Does this only happen inside a Controller method? I have a method within the application which needs to be executed as a new thread and the main thread should just return – Amol Kshirsagar Feb 08 '22 at 12:21

score 1 · Answer 2 · answered Mar 09 '19 at 19:43

No. In this case you are calling the long running process in the Netty IO thread. The simplest way I can think of to this is to create a Mono sink and run the long op in a new thread (or perhaps via a thread pool). When the operation completes successfully, you call sink.success(), and if it fails, you call sink.error(x) passing the thrown exception.

@GetMapping("/foo")
public Mono<Void> bar() {
    return Mono.create(sink -> {
        new Thread(() -> {
            try {
                someService.veryLongSyncOperation();
                sink.success();
            } catch (Exception ex) {
                sink.error(ex);
            }
        }).start();
    });  
}

The calling thread returns immediately after setting up the flow, and WebFlux will subscribe to the returned Mono which will trigger the thread to run in the new thread.

Spring Webflux / Reactor release calling thread

2 Answers2

Linked