07 Blocking to Reactive

이 글은 tech.io/playgrounds/929/reactive-programming-with-reactor-3/BlockingToReactive 에 대한 문제풀이이다.

 

 

Reactive 코드 속에서 레거시하고, Reactive하지 않은 코드는 어떻게 처리해야할까?

예를들어, JDBC connection과 같은 Blocking 코드가 있고 이를 Reactive pipeline에 통합해한다면 어떻게해야할까?

가장 좋은 방법은, blocking이 진행되는 부분을 Schedulers를 통해 별도의 스레드에서 실행되도록 고립시키는 것이다.

JDBC의 예에서는 fromIterable 팰터리 메서드를 사용할 수 있다.

 

subscribeOn

이 메서드는 제공된 스케줄러에서 시작부터 시퀀스를 분리해서 실행할수 있도록한다.

이 트릭을 사용한 첫번째 예제를 통해, repository의 모든 사용자를 천천히 읽어보자.

Flux<User> blockingRepositoryToFlux(BlockingRepository<User> repository) {
    return Flux.defer(() -> Flux.fromIterable(repository.findAll())).subscribeOn(Schedulers.elastic());
} 

defer메서드로 repository함수 호출을 감싸, 지연된 instance를 Flux에 등록할 수 있다.

 

 

publishOn

database에 저장하는것과 같은 느린 subscriber에서는, 시퀀스의 좀더 작은 부분에 publishOn을 사용하여 별도로 처리할 수 있다.

Mono<Void> fluxToBlockingRepository(Flux<User> flux, BlockingRepository<User> repository) {
    return flux.publishOn(Schedulers.elastic()).doOnNext(repository::save).then();
} 

위 함수가 Reactive Streams의 pipeline에서 실행되면, 이 시퀀스만 별도의 스케줄러를 사용하여 실행되게 된다.

 

 

 

 

 

참고

tech.io/playgrounds/929/reactive-programming-with-reactor-3/BlockingToReactive

 

wiki.sys4u.co.kr/pages/viewpage.action?pageId=7766994#id-%EC%97%B0%EC%8A%B5%EB%AC%B8%EC%A0%9C%EB%A1%9C%EB%B0%B0%EC%9B%8C%EB%B3%B4%EB%8A%94Reactor-11.BlockingtoReactive

http://wiki.sys4u.co.kr/pages/viewpage.action?pageId=7766994#id-%EC%97%B0%EC%8A%B5%EB%AC%B8%EC%A0%9C%EB%A1%9C%EB%B0%B0%EC%9B%8C%EB%B3%B4%EB%8A%94Reactor-11.BlockingtoReactive

luvstudy.tistory.com/100

Reactor 언제 어떤 Operator를 써야 할까?