How to switch back from blocking scheduler to previous scheduler using netty-reactor?

user_2864183 0
Viewed 269

How to switch back from blocking scheduler (blocking-pool) to previous scheduler (reactor-http-nio) using Spring Webflux + Netty + Reactor?

The code:

@RequiredArgsConstructor
@Service
@Slf4j
public class BookService {

    private final IBookRepo bookRepo;

    private final BlockingPoolConfig blockingPoolConfig;

    public Mono<Optional<Book>> getBook(Long id) {
        log.debug("getBook() - id: {}", id);
        return asyncCallable(() -> {
            log.trace("getBook() - invoking bookRepo.findById(id) ...");
            return bookRepo.findById(id);
        });
    }

    protected <S> Mono<S> asyncCallable(Callable<S> callable) {
        return Mono.fromCallable(callable)
                .subscribeOn(blockingPoolConfig.blockingScheduler()); 
    }
}

@RestController
@RequiredArgsConstructor
@Slf4j
public class BookController {

    private final BookService bookService;

    @GetMapping("/book/{id}")
    public Mono<Book> get(@PathVariable Long id) {
        log.debug("get() - id: {}", id);
        return bookService.getBook(id)
                .publishOn(Schedulers.parallel())  //publishOn(... ?)
                .map(optionalBook -> {
                    return optionalBook.map(book -> {
                        log.debug("get() result: {}", book);
                        return book;
                    }).orElseThrow(() -> {
                        log.debug("book with id: {} is not found.", id);
                        return new ResponseStatusException(HttpStatus.NOT_FOUND, "Book not found");
                    });
                });
    }

@Configuration
@Slf4j
public class BlockingPoolConfig {

    @Value("${spring.datasource.maximumPoolSize:8}")
    private int connectionPoolSize = 1;

    @Scope("singleton")
    @Bean
    public Scheduler blockingScheduler() {
        Scheduler scheduler = Schedulers.newBoundedElastic(connectionPoolSize, connectionPoolSize, "blocking-pool");
        return scheduler;
    }
}

Above i'm using publishOn(Schedulers.parallel()), but this one creates new thread pool (parallel). Instead of this I prefer to switch reactor-http-nio thread pool.

Log of actual result:

19:17:45.290 [reactor-http-nio-2       ] DEBUG t.a.p.controller.BookController    - get() - id: 1
19:17:45.291 [reactor-http-nio-2       ] DEBUG t.a.p.service.BookService          - getBook() - id: 1
19:17:45.316 [blocking-pool-1          ] TRACE t.a.p.service.BookService          - getBook() - invoking bookRepo.findById(id) ...
19:17:45.427 [parallel-2               ] DEBUG t.a.p.controller.BookController    - get() result: Book(id=1, title=Abc)

Log of expected result:

19:17:45.290 [reactor-http-nio-2       ] DEBUG t.a.p.controller.BookController    - get() - id: 1
19:17:45.291 [reactor-http-nio-2       ] DEBUG t.a.p.service.BookService          - getBook() - id: 1
19:17:45.316 [blocking-pool-1          ] TRACE t.a.p.service.BookService          - getBook() - invoking bookRepo.findById(id) ...
19:17:45.427 [reactor-http-nio-2       ] DEBUG t.a.p.controller.BookController    - get() result: Book(id=1, title=Abc)
reactor-netty project-reactor spring-webflux spring-boot
1 Answers

This is currently not possible, because A) these HTTP threads are not controlled by a Reactor Scheduler, but by the underlying Netty event loop itself, and B) there's no generic way in Java to "return execution to an (arbitrary) thread" if that thread doesn't have an Executor/ExecutorService associated with it.

For reactor-netty, once you've switched out of the HTTP threads there should be little reason to want to switch back to the Netty threads anyway. It will be done naturally by reactor-netty once the response is sent.

Assuming blocking pool is something like Schedulers.boundedElastic(), you might indeed want to go to Schedulers.parallel() to limit the life of the blocking threads, and that's a perfectly fine solution.

user_11134860
Related