Reactor에서 말하는 스케쥴러는 @Scheduled 애너테이션을 이용해 사용하는 그 스케쥴러랑은 결이 다르다. 후자는 등록된 쓰레드풀을 돌리는 이름붙은 별도 쓰레드풀 관리자로 설명이 충분한데, 후자는 비슷한 결에 몇가지 기능이 더 있음.

Reactor에서 스케쥴러가 작업을 받으면 실제 작업을 실행하는 실행 단위는 Scheduler.createWorker() 로 생성하는 Worker 임. 개념적으로는 이 친구를 별도 쓰레드로 취급할수도 있는데, 사실 항상 쓰레드일 필요는 없다. 작업을 실행하는 쓰레드를 전혀 바꾸지 않는 워커 구현체도 있기 때문. 정확하게 이해하려면, 작업이 실행될 컨텍스트를 결정하기 위한 추상화라고 생각하면 됨.

그럼 이름을 왜 헷갈리게 스케쥴러라고 지은걸까? 그건 이 인터페이스가 워커를 지정해주는것 뿐만 아니라 시간을 기반으로 한 기능까지 담당하기 때문이다.

인터페이스를 보자.

interface Scheduler extends Disposable {
    Disposable schedule(Runnable task);                                   // 즉시 실행
    Disposable schedule(Runnable task, long delay, TimeUnit unit);        // 지연 실행
    Disposable schedulePeriodically(Runnable task, long initialDelay, long period, TimeUnit unit); // 주기 실행
    long now(TimeUnit unit);                                              // 시계(clock)
    Worker createWorker();
}

단순 워커를 생성해서 반환하는 것 뿐만 아니라.. delay 등의 인자를 받거나, schedulePeriodically , now 등의 메서드가 있음. Mono.delay() , Flux.interval() , delayElement() 같은 시간관련된 메서드들이 결국 이 Scheduler 를 베이스로 하고 있음.

비슷한 케이스를 생각해보면 JDK의 ExecutorService 가 있다. 특히 ScheduledExecutorService 구현체와 기능이 비슷하다고 생각할 수 있음. 다만 다음의 차이가 있다.

  1. ExecutorService는 여러 태스크를 받고 그걸 여러 쓰레드에서 동시에 잘 실행시키켜주기면 하면 ok지만, Scheduler 는 반드시 한 subscription에서 나오는 여러 신호(onNext, onComplete)들을 겹치지 않고 순차적으로 실행해야 만 한다. 때문에 작업을 보통 FIFO순서로 처리함.
  2. now() 로 이 스케쥴러가 생각하고 있는 현재 시각을 밖으로 노출하고, 그 시간을 개발자가 바꿔 낄 수 있도록 했음. 그래서 테스트 등에서 시간을 빨리감기할 수도 있다.
  3. ScheduledExecutorService 는 진짜 쓰레드 풀을 감싼 녀석임. 하지만 Scheduler 는 인터페이스에 불과해서 쓰레드를 쓰지 않는 구현도 있을 수 있다. 실제로 Scheduler.immediate() 는 제출된 Runnable을 현재 쓰레드에서 그대로 즉시 실행한다. 이를 두고 실행컨텍스트가 별도로 없다고 표현하는데, 이때 ‘실행 컨텍스트’ 라는 개념이 쓰레드 풀보다 더 넓은 개념으로 사용됨.

라곤 하지만… 결국 블로킹하지 않기 위해 별도 쓰레드를 워커라고 부르면서 도입할수 밖에 없는 것 처럼 보인다. 그런데 이 워커가 경량쓰레드가 아니라면… 그냥 가상쓰레드 쓰는게 낫지 않나?

사용 예시코드

    fun create(request: ServerRequest): Mono<ServerResponse> {
        return request.bodyToMono<AdmMailRequest>()
            .flatMap { req -> Mono.fromCallable { admMailService.create(req) }
							         .subscribeOn(Schedulers.boundedElastic()) }
            .flatMap { response -> ServerResponse.status(HttpStatus.CREATED).bodyValue(response) }
    }

BoundedElastic

이름이 참 특이해서 처음에 보면 무슨 ES랑 관련이 있나 싶은데, 핵심 사용법은 쓰레드 풀Executor임. parallel() 이 코어수만큼 고정된 쓰레드를 다룬다면, 이 녀석은 수요에 맞게 워커를 동적으로 만들고 회수하는 것 차이.

생각해보면 ‣ 의 ThreadpoolExecutor 랑 다르게 이해할 필요가 있나 싶기도 함. 오히려 네이밍은 너무 구리다.

이녀석은 이런식으로 동작한다.