[WebFlux] Reactive Streams

Reactive Programming

✳️ 전통적인 콜백 기반 API나 Future와 같은 JDK의 비동기 코드 작성 방식에 비해 더 유연하고 효율적

1. 비동기 처리의 간소화 : 비동기 작업을 스트림의 형태로 처리하므로 데이터의 흐름을 통해 각 단계를 연결하고 데이터가 준비되는 즉시 처리 가능

2. 선언적 API 제공 : Reactor, RxJava와 같은 라이브러리는 map, filter, reduce와 같은 함수형 프로그래밍 스타일의 연산자를 제공하여 데이터 변환하고 조작하는 과정을 간결하게 표현

3. backPressure 지원 : 데이터 소비자가 데이터 생산자의 속도를 제어할 수 있도록 함 

4. 복잡성 감소와 가독성 향상 : 연산자 체이닝을 통해 코드의 가독성 높이고 각 단계의 작업을 분리하여 개발자가 코드를 이해하고 유지보수하기 쉽게 함

 

• stream (스트림)
  • 데이터의 연속적인 흐름
• event-driven (이벤트 기반)
  • 외부 환경이나 사용자 입력과 같은 이벤트에 반응하여 동작
  • 이벤트가 발생할 때마다 그에 따라 비동기적으로 처리 
• Async (비동기성)
  • 여러 작업이 동시에 실행될 수 있으며 결과는 작업이 완료될 때까지 기다리지 않고 이벤트나 콜백을 통해 처리됨

---

✳️ 자바에서 제공하는 비동기 프로그래밍 vs Reactor 코드 

userService.getFavorites(userId, new Callback<List<String>>() { 
  public void onSuccess(List<String> list) { 
    if (list.isEmpty()) { 
      suggestionService.getSuggestions(new Callback<List<Favorite>>() {
        public void onSuccess(List<Favorite> list) { 
          UiUtils.submitOnUiThread(() -> { 
            list.stream()
                .limit(5)
                .forEach(uiList::show); 
            });
        }

        public void onError(Throwable error) { 
          UiUtils.errorPopup(error);
        }
      });
    } else {
      list.stream() 
          .limit(5)
          .forEach(favId -> favoriteService.getDetails(favId, 
            new Callback<Favorite>() {
              public void onSuccess(Favorite details) {
                UiUtils.submitOnUiThread(() -> uiList.show(details));
              }

              public void onError(Throwable error) {
                UiUtils.errorPopup(error);
              }
            }
          ));
    }
  }

  public void onError(Throwable error) {
    UiUtils.errorPopup(error);
  }
});

요약하면 다음과 같다

1. 비동기 작업을 여러 단계로 수행하면서, 사용자의 즐겨찾기 목록을 먼저 가져오고, 즐겨찾기가 없는 경우 추천 항목을 제공하는 구조
2. 각 단계에서 성공과 오류 처리를 위한 콜백 제공

👉 콜백 지옥에 빠질 수 있다
중첩된 콜백 함수들이 여러 단계로 구성되어 있고, 코드가 점점 깊어지면서 가독성 또한 떨어지고 유지보수가 어려워진다.

 

userService.getFavorites(userId) 
           .flatMap(favoriteService::getDetails) 
           .switchIfEmpty(suggestionService.getSuggestions()) 
           .take(5) 
           .publishOn(UiUtils.uiThreadScheduler()) 
           .subscribe(uiList::show, UiUtils::errorPopup);

Reactor 코드로 나타내면 위와 같다. 

1. flatMap : 여러 즐겨찾기 ID에 대한 요청을 처리하고 그 결과를 하나의 스트림으로 반환
2. switchIfEmpty : 스트림이 비어있을 경우, 즉 즐겨찾기 목록이 비어있을 경우 호출되는 메서드
3. take(5) : 스트림에서 최대 5개의 항목만 가져오도록 제한
4. publishOn : 다음 단계의 실행을 UI 스레드에서 수행하도록 지시 (특정 스케줄러에서 실행하도록 지정)
5. subscribe : 스트림 구독

 

 

 

Reactive Streams

비동기적이고 데이터를 효율적으로 처리할 수 있는 스트림 처리를 위한 표준 정의 

• Publisher 
  • 데이터 스트림을 생성하고 데이터를 발행하는 역할 
  • onSubscribe, onNext, onError, onComplete 메서드를 통해 데이터 전송 
    • onSubscribe : subscription 객체 전달
    • onNext : subscriber는 데이터를 받고 처리 
    • onComplete : 모든 아이템이 전달 완료되어 더 이상 전달할 데이터가 없을 때 호출 -> publisher와 subscriber 연결 종료
    • onError : 데이터 스트림 처리 중 오류 발생시 호출
  • subscribe 함수 제공해서 publisher에 다수의 subscriber 등록 지원
  • ColdPublisher 
    • subscriber가 구독할 때마다 새로운 데이터 스트림 생성
  • HotPublisher
    • 데이터 스트림이 구독자에게 독립적으로 생성되지 않고, 스트림이 이미 활성화되어 있는 상태에서 subscriber가 스트림에 접근함
    • 모든 subscriber가 스트림 공유하며 현재 상태나 중간 결과를 받음

 

• Subscriber
  • 데이터 스트림을 소비하고 처리하는 역할
  • subscribe하는 시점에 publisher로부터 subscription을 받을 수 있는 인자 제공

 

• Subscription
  • Publisher와 Subscriber 간의 연결
  • 데이터 흐름 제어 

 

---

✳️ 구현 라이브러리

• Project reactor
  • Mono와 Flux publisher 제공 
  • Flux : 0개 이상의 요소를 비동기적으로 처리하는 데이터 스트림 👉 다수의 요소 처리 ≒ List<T>
  • Mono : 최대 1개의 요소를 비동기적으로 처리하는 데이터 스트림 👉 단일 값이나 빈 값 처리 ≒ Optional<T>
  • subscribeOn() : 다른 스레드에서 스트림이 처리되도록 함 

🚨Flux에서 하나의 값만 넘겨주면 되는데 Mono를 사용하는 이유..? 

1. 명확한 의도 표현 = 반환되는 값이 단일 항목임
2. Mono<Void>로 특정 사건이 완료되는 시점을 가리킬 수 있음 
3. onNext 이후 바로 onComplete를 호출하면 되기 때문에 구현이 더 간단 

• RxJava

 

 

 

Reactor

1. Reactive Streams를 구현한 비동기 데이터 스트림 처리 지원
2. Spring WebFlux에서 메인으로 사용
3. backpressure를 제공하여 안정성 높임
4. 다양한 연산자로 다양한 연산들을 조합하여 가독성 증대 

• subscribe : subscribe를 하지 않으면 아무일도 생기지 X 

 

 

 

참고자료 

https://d2.naver.com/helloworld/2771091

https://techblog.woowahan.com/12903/

https://tech.io/playgrounds/929/reactive-programming-with-reactor-3/Intro

https://projectreactor.io/docs/core/release/reference/#getting-started-introducing-reactor