🌈 [Section4] 8. [Spring WebFlux] Project Reactor

📕 오늘 배운 내용!

• Reactor (Project Reactor)
• 마블 다이어그램 (Marble Diagram)
• 스케줄러 (Scheduler)
• Operators

---

✏️ Reactor (Project Reactor)

• 리액티브 스트림즈 표준 사양을 구현한 구현체 중 하나
• 리액티브한 애플리케이션으로 동작하는데 있어 핵심적인 역할을 담당하는 리액티브 프로그래밍을 위한 라이브러리

[참고 1] / [참고 2]

---

✔ Reactor 특징

[공식 홈페이지]

• Non-Blocking 통신
  ➜ 쓰레드가 차단되지 않음
  
  
• MSA(Microservice Architecture) 구조에 적합한 라이브러리
  ( MSA 기반 애플리케이션들은 요청 쓰레드가 차단되는 Blocking 통신을 사용하기에는 무리가 있기 때문에 Non-Blocking 통신을 완벽히 지원하는 라이브러리여야 하기 때문 )
  
  
• Operator로 시작해서 Operator로 끝난다 !
  
  
• Mono[0|1]와 Flux[N]이라는 두 가지 Publisher 타입 제공
  • Mono[0|1]에서 0과 1
    ➜ 0건 or 1건의 데이터를 emit 할 수 있음을 의미
  • Flux[N]에서 N
    ➜ 여러 건의 데이터를 emit할 수 있음을 의미
    
    
• Backpressure 전략
  • Subscriber의 처리 속도가 Publihser의 emit 속도를 따라가지 못할 때 적절하게 제어하는 전략
  • DROP 전략
    ➜ 버퍼가 가득 차면, 이후에 들어오는 데이터는 DROP하는 전략
  • LATEST 전략
    ➜ 버퍼가 가득 차면, 가장 최근에 emit된 데이터만 남기고 폐기하는 전략

💡 LATEST 전략이 DROP 전략이랑 비슷한데
DROP은 Downstream에서 버퍼를 모두 처리할 때 까지 들어오는 데이터들을 들어올 때 마다
하나하나 DROP하다가, 버퍼가 비면 그 때 emit된 데이터를 차례로 다시 버퍼에 넣는 전략이고

LATEST는 Downstream에서 버퍼를 모두 처리할 때 까지 들어오는 데이터들을 그대로 대기열로 모아놨다가,
버퍼가 비면 그 때 마지막으로(최근에) emit된 데이터만 남기고 나머지 대기열의 데이터들을 한번에 폐기하는 전략임

• • BUFFER 전략
    • BUFFER DROP LATEST
      ➜ 버퍼가 가득 차면, 버퍼 안에 있는 데이터 중 가장 최근에 채워진 데이터를 DROP하는 전략
    • BUFFER DROP OLDEST
      ➜ 버퍼가 가득 차면, 버퍼 안에 있는 데이터 중 가장 오래된 데이터를 DROP하는 전략

💡 만약, Publisher에서 들어오는 데이터를 끊임없이 emit하는데 Subscriber의 처리 속도가 느리다면,
대기하는 데이터가 지속적으로 쌓이는 것을 방치하게되면 오버플로우가 발생하게 되고 결국엔 시스템이 다운될 것
➜ Backpressure 전략으로 이를 방지
⠀
[참고] https://projectreactor.io/docs/core/release/reference/#reactive.backpressure

---

✏️ 마블 다이어그램 (Marble Diagram)

• 하나의 데이터가 다이어그램 상에서 시간의 흐름에 따라 어떻게 변화하는지 데이터의 흐름을 표현한 그림

➜ 리액티브 프로그래밍에서는 Operator에 따라 데이터 흐름이 다양하게 변화하는데,
이 때 복잡한 데이터의 흐름을 마블 다이어그램을 통해 좀 더 쉽게 이해 가능

• Reactor에서 제공하는 수많은 Operaotr의 내부 동작을 조금 더 쉽게 이해하고,
  해당 Operator를 적절하게 사용하는데 도움을 줌

✔️ 마블 (Marble)

• 원래 단어의 의미는 '구슬'
• 하나의 데이터를 나타냄
  ➜ 마블 다이어그램이 구슬(하나의 데이터)이 어떤 흐름으로 굴러가는지 표현한 것

[참고]

---

✔ Mono의 마블 다이어그램

( Reactor의 데이터 타입 중 하나인 Mono를 마블 다이어그램으로 표현한 것 )

• 한개의 구슬
  ➜ 0건 또는 1건의 데이터만 emit하는 Reactor 타입임을 의미
  
  
• 위 아래로 두개의 타임 라인이 있고, 모두 데이터가 흘러가는 시간의 흐름을 표현
  ( 왼 ➜ 오 로 시간이 흐름 )

① 원본 Mono(Original Mono)에서 Sequence가 시작되는 것을 타임라인으로 표현
⠀
② Mono의 Sequence에서 데이터가 emit 되는 것을 표현
⠀
③ 수직 막대 바는 Mono의 Sequence가 정상 종료되었다는 것을 의미
⠀
④ Mono에서 지원하는 어떤 Operator에서 입력으로 들어오는 구슬 모양의 데이터가 가공 처리되는 것을 표현
⠀
⑤ Operator에서 가공 처리된 데이터가 Downstream으로 전달될 때의 타임라인
⠀
⑥ Mono에서 emit된 데이터가 처리되는 과정에 에러가 발생한다면 X로 표시
❗ |는 정상 종료 / X는 에러로 인한 비정상 종료

---

✔ Flux의 마블 다이어그램

( Reactor의 데이터 타입 중 하나인 Flux를 마블 다이어그램으로 표현한 것 )

• Mono와는 달리 여러 개의 구슬
  ➜ 여러 개(0 … N)의 데이터를 emit하는 Reactor 타입임을 의미

( ⬇️ 아래 ① ~ ⑥은 그냥 표를 읽는 방법이므로 Mono와 같음 )
⠀
① 원본 Flux에서 Sequence가 시작되는 것을 타임라인으로 표현
⠀
② Flux의 Sequence에서 데이터가 emit 되는 것을 표현
⠀
③ 수직 막대 바는 Flux의 Sequence가 정상 종료되었다는 것을 의미
⠀
④ Flux에서 지원하는 어떤 Operator에서 입력으로 들어오는 구슬 모양의 데이터가 가공 처리되는 것을 표현
⠀
⑤ Operator에서 가공 처리된 데이터가 Downstream으로 전달될 때의 타임라인
⠀
⑥ Flux에서 emit된 데이터가 처리되는 과정에 에러가 발생한다면 X로 표시
❗ |는 정상 종료 / X는 에러로 인한 비정상 종료

💡 main 쓰레드 / 데몬 쓰레드
⠀⠀
✔️ main 쓰레드
➜ 주 쓰레드
⠀
✔️ 데몬 쓰레드
➜ 보조 쓰레드 ( 주 쓰레드에 영향을 받는 쓰레드 )
➜ 주 쓰레드가 종료되면 따라서 종료됨

---

✔ map() Operator의 마블 다이어그램 예시

• map() Operator
  ➜ 입력으로 들어오는 데이터를 개발자가 구현하는 동작대로 변환해서 Downstream으로 전달하는 역할
  
  
• 위 그림을 살펴보면,
  위의 sequence에서 x로 들어간 데이터가 동일한 색의 결과값으로 변환되어 Downstream으로 전달되는 것을 알 수 있음

---

✏️ 스케줄러 (Scheduler)

• 쓰레드 관리자
  ➜ 여러 쓰레드들을 손쉽게 관리
  ➜ 복잡한 멀티쓰레딩 프로세스를 단순하게 해줌
• Reactor Sequence 상에서 처리되는 동작들을 하나 이상의 쓰레드에서 동작하도록 별도의 쓰레드를 제공해 주는 것

[Scheduler 참고]

---

✔ Scheduler 전용 Operator

( 적절한 상황에 맞는 쓰레드를 추가로 생성하는 Operator )

✔️ subscribeOn() Operator

• 데이터 소스에서 데이터를 emit하는 원본 Publisher의 실행 쓰레드를 지정하는 역할
• 구독 직 후 실행되는 Operator 체인의 실행 쓰레드를 Scheduler로 지정한 쓰레드로 변경

✔️ doOnSubscribe() Operator
- 구독 발생 직 후에 트리거 되는 Operator
- 구독 직 후에 어떤 동작을 수행하고 싶다면 doOnSubscribe()에 로직을 작성하면 됨
- 이 앞에 subscribeOn() Operator 사용할 경우 doOnSubscribe()의 로직은 쓰레드가 바뀌지 않고 main에서 실행됨
- 이후 데이터를 생성하고 emit하는 단계부터는 다른 쓰레드에서 실행

• 주로 Schedulers.boundedElastic() 사용

✔️ publishOn() Operator

• 전달 받은 데이터를 가공 처리하는 Operator 앞에 추가해서 실행 쓰레드를 별도로 추가하는 역할
• publishOn() 기준 Downstream 쓰레드가 publishOn()의 Scheduler로 지정한 쓰레드로 변경

✔️ doOnNext() Operator
- doOnNext() 바로 앞에 위치한 Operator가 실행될 때, 트리거 되는 Operator

• 주로 Schedulers.parallel() 사용
  
  

💡 Operator 체인마다 실행 쓰레드를 구분해서 실행할 수 있도록 하는 이유
Spring WebFlux 기반의 애플리케이션은 적은 수의 쓰레드로 대량의 요청을 Non-Blocking 방식으로 처리할 수 있는 구조임
이 구조는 쓰레드가 복잡한 계산이 필요한 작업의 경우, 응답 지연이 발생할 수 있음 !
➜ Scheduler를 통해 별도의 쓰레드를 생성한 후, 복잡한 계산을 수행하도록 함

[Scheduler 지원 쓰레드 유형 참고]

---

✏️ Operators

• Reactor에서 가장 중요한 구성요소

✔ 상황별로 분류된 Operator 목록

[Which operator do I need? 참고]

① 새로운 Sequence를 생성(Creating)하고자 할 경우

• just()
• ⭐ fromStream()
  ➜ Java의 Stream을 입력으로 전달 받아 emit하는 Operator
• ⭐ fromIterable()
  ➜ Java의Iterable을 입력으로 전달 받아 emit하는 Operator
  ( List, Map, Set 등의 컬렉션을 fromIterable()의 파라미터로 전달 가능 )
• fromArray()
• range()
• interval()
• empty()
• never()
• defer()
• using()
• generate()
• ⭐ create()
  ➜ 프로그래밍 방식으로 Signal 이벤트를 발생시키는 Operator
  ➜ 한 번에 여러 건의 데이터를 비동기적으로 emit할 수 있는 Operator

✔️ Signal
➜ Publisher가 발생시키는 이벤트

② 기존 Sequence에서 변환 작업(Transforming)이 필요한 경우

• ⭐ map()
  ➜ 안에서 가공처리 해줌
• ⭐ flatMap()
  ➜ flatMap() 내부로 들어오는 데이터마다 하나의 새로운 Sequence가 생성됨
  ( 들어오는 Sequence는 하나인데, 내부에서 작동되는 Sequence는 여러개 )
  ➜ flatMap() 내부에서 정의하는 Sequence = Inner Sequece
  ➜ But, 각 쓰레드의 작업의 처리 순서를 보장하지 않음 ( 출력 랜덤 )
• ⭐ concat()
  ➜ 입력으로 전달하는 Publisher의 Sequence를 하나로 이어붙여서 차례대로 데이터를 emit함
  ( 순서대로 emit됨 )
• collectList()
• collectMap()
• merge()
• ⭐ zip()
  ➜ 입력으로 전달되는 여러 개의 Publisher Sequence에서 emit된 데이터를 결합하는 Operator
  ➜ 각각의 Sequence에서 emit되는 데이터 중 같은 차례(index)의 데이터들이 결합
  ( 두 Sequence의 emit 시점이 매번 다르더라도 emit 시점이 늦은 데이터가 emit될 때까지 대기 했다가 두 개의 데이터를 전달 받아 결합 )

💡 여기서의 '결합'
➜ 각 Publisher가 emit하는 데이터를 하나씩 전달 받아서 새로운 데이터를 만든 후에 Downstream으로 전달한다는 의미

• then()
• switchIfEmpty()
  ➜ 들어오는 데이터가 없으면 내부에서 새 Sequence를 만들어서 그 Sequence의 결과 데이터를 리턴할 수 있고,
  들어오는 데이터가 있으면 그 데이터를 리턴
• and()
• when()

③ Sequence 내부의 동작을 확인(Peeking)하고자 할 경우

• doOnSubscribe
• ⭐ doOnNext()
  ➜ 데이터 emit 시 트리거 되어 부수 효과(side-effect)를 추가할 수 있는 Operator
  ➜ 리턴값이 없음
  ➜ 주로 로깅(로그를 기록 또는 출력하는 작업)에 사용하지만,
  데이터를 emit하면서 필요한 추가 작업이 있다면 doOnNext()에서 처리

✔️ 부수 효과 (side-effect)
➜ 어떤 동작을 실행하되 리턴 값이 없는 것

• doOnError()
• doOnCancel()
• doFirst()
• doOnRequest()
• doOnTerminate()
• doAfterTerminate()
• doOnEach()
• doFinally()
• ⭐ log()
  ➜ Publisher에서 발생하는 Signal 이벤트 마다 로그로 출력해주는 역할
  ➜ 원하는 만큼 추가 가능

④ Sequence에서 데이터 필터링(Filtering)이 필요한 경우

• ⭐ filter()
  ➜ 들어오는 데이터를 어떤 조건에 맞춰서 필터링
• take()
  ➜ 파라미터에 처리하고싶은 데이터의 수가 들어감
• ignoreElements()
• distinct()
• ⭐ take()
• next()
• skip()
• sample()
• single()

⑤ 에러를 처리(Handling errors)하고자 할 경우

• ⭐ error()
  ➜ Reactor Sequence 상에서 의도적으로 예외를 던져서 onError Signal 이벤트 발생시키는데 사용
  ( throw 와 같음 )
• ⭐ timeout()
  ➜ 입력으로 주어진 시간동안 emit되는 데이터가 없으면 onError Signal 이벤트 발생 시킴
  ➜ retry() Operator와 함께 사용하는 경우가 많음
  ( 일정한 시간 안에 안오면 다시 재시도 해봐라 ! )
• onErrorReturn()
• onErrorResume()
• onErrorMap()
• doFinally()
• ⭐ retry()
  ➜ Sequence 상에서 에러가 발생할 경우, 입력으로 주어진 숫자만큼 재구독해서 Sequence를 다시 시작
  ➜ 파라미터에 재시도할 수 있는 횟수가 들어감
  ➜ timeout() Operator와 함께 사용하는 경우가 많음

---

😜 실습

• projects - be-template-reactor
  ( 구현 실습보단 예시에 가까움 )

---

🌈 느낀점

사실 뭔가 Stream이랑 비슷해서 이해는 그래도 쉽긴 했는데
Mono와 Flux를 그래서 언제 어떻게 써야 하는지는 아직 잘 모르겠다 ! ㅎㅎ
그리고 Operator 너무 많아..
나중에 마저 공부해야겠닷

 

( 본 게시물은 2022/12/01에 작성한 글을 옮긴 글입니다. 원문은 요기에 있음! )

728x90