3. Reactive Streams - part 2(with 토비의 봄 TV 스프링 리액티브 프로그래밍 요약)

[dreamchaser3]

참고 블로그

https://jongmin92.github.io/

[gaepury]

디버깅 하기

• applicationContext에서(main 메서드) Hooks.onOperatorDebug()

[gaepury]

publishon vs subscribeon (by 유정님)

[jekalmin]

다음주 숙제:

• 3편(Reactive Streams - Schedulers): https://www.youtube.com/watch?v=Wlqu1xvZCak

[gaepury]

토비의 봄 TV 6회 스프링 리액티브 프로그래밍 전체 볼것들 List

• 1편(Reactive Streams): https://www.youtube.com/watch?v=8fenTR3KOJo
• 2편(Reactive Streams - Operators): https://www.youtube.com/watch?v=DChIxy9g19o
• 3편(Reactive Streams - Schedulers): https://www.youtube.com/watch?v=Wlqu1xvZCak
• 8편(WebFlux): https://www.youtube.com/watch?v=ScH7NZU_zvk
• 9편(Mono의 동작방식과 block()): https://www.youtube.com/watch?v=LK6NRV8tZBM
• 10편( Flux의 특징과 활용방법): https://www.youtube.com/watch?v=bc4wTgA_2Xk

[gaepury]

토비의 봄 TV 6회 스프링 리액티브 프로그래밍 1편(Reactive Streams) 요약

https://www.youtube.com/watch?v=8fenTR3KOJo

iterable <—> observable(상대성)

• Iterable: pull 방식
• observable: push

  옵저버 패턴의 부족한 점

• Completed
• Error

  Reactive Streams

• Publisher: Obervable 과 대응
• Subscriber: Observer와 대응
• Publisher.subscribe(Subscriber)
  • observable.addObserver와 대응
• Subscription
  • publisher, subscriber 중계 역할
    • backpressure(publisher, subscriber 속도차 조절)
  • 전체 flow
    • 
  • 최초 request(Subscription 메서드)는 onSubcribe에서 해야함(Subscriber)
  • Publisher를 병렬적으로 처리하고싶다고 멀티 스레드 환경처럼 하는것은 불가능

[gaepury]

토비의 봄 TV 6회 스프링 리액티브 프로그래밍 2편(Reactive Streams - Operators) 요약

https://www.youtube.com/watch?v=DChIxy9g19o

Publisher -> [data1] -> op1 -> [data2] -> op2 -> [data3] -> Subscriber

• 

  mapPub

• 
  • onSubscribe, onError, onCompleted는 그대로 중계

    DelegateSub로 code refactoring

• 

  sumPub

• 어떻게 합계할것인가? onNext에서 어떻게 끝이라는걸 알지?
  • 모른다. onComplete에서 onNext호출
• 

  reducePub

• 

  Generic method, class 변환

  mapPub, reducePub 메서드의 generic화

• data type 변환(T -> R)
• 

  Reactor

• Flux(Publisher 구현)
• Ex)
  • 
  • log()로 onSubcribe, request, onNext등 메서드 로그를 볼수있음
• map
  • 
  • map operator에 의해 onSubscribe, onNext등이 두번씩 호출
• 스프링 MVC에 적용
  • 

[gaepury]

토비의 봄 TV 7회 스프링 리액티브 프로그래밍 3편(Reactive Streams - Operators) 요약

https://www.youtube.com/watch?v=Wlqu1xvZCak

• consume이 느린경우 onNext, onError, onCompleted 메서드에 스케줄러를 매핑 및 중계해서 다른 스레드에서 돌도록 처리함

  • publishOn의 구현방법

• request의 과정이 느릴때(데이터 생성이 느릴떄)

  • subcribeOn의 구현방법

• 생명주기

  • 조립, 구독, 실행

• 요약

  • flux는 publisher의 구현체
  • subscribeOn: 데이터 생성이 느릴떄 사용해라
    • 직접 구현: subOnPub Operator
      • 
      • 적용: subOnPub.subscribe(new Subscriber …{})
    • Slow Publisher, fast consumer
  • publishOn:
    • 직접 구현: pubOnPub Operator
      • 
    • Fast publisher, slow consumer
  • subscribeOn + publishOn
    • 
    • 
  • Thread name 지정하기
    • 
      • 스프링이 제공하는 CustomizableThreadFactory 이용
  • reactor 에서 publishOn ,subsribeOn
    • 
    • result
      • 
  • publishOn 스레드풀 셧다운
    • onError, onComplete에 구현
    • 
      • shundownNow (인터럽트 발생시켜 강제 셧다운)
  • Flux.interval
    • 데몬스레드여서 메인스레드 종료시 같이 종료됨
  • Take 구현
    • takePub
      • 
    • subscription
      • 

[gaepury]

토비의 봄 TV 12회 스프링 리액티브 프로그래밍 (8) WebFlux

https://www.youtube.com/watch?v=ScH7NZU_zvk

• 단일 api 호출 By Mono
  • 
  • 구독하는 코드가 별도로 없어도 Mono를 반환하면 Spring에서 Mono를 구독해줌.
• 멀티 api 호출 By Mono
  • 
    • Netty에 worker 스레드에서 동작
• Service nonblocking으로 실행하기
  • @Async + (Future, ListenableFuture, CompletableFutue)
  • CompletableFutue을 모노로 바꾸기
    • Mono.fromCompletionState
    • CompletableFuture는 CompletionState를 상속함
  • 

[gaepury]

토비의 봄 TV 13회 스프링 리액티브 프로그래밍 (9) Mono의 동작방식과 block()

https://www.youtube.com/watch?v=LK6NRV8tZBM

• .log()는 중간 퍼블리셔라고 생각하면 됨
  • Publisher -> publisher -> publisher -> subscriber
• doOnNext, .log() 코드 및 결과
  • 
  • 
  • just내 함수가 호출되면 구독 전에 미리 호출됨, 레이지로 준비하려면 fromSupplier 메서드 사용.
• spring에서 해주기전에 미리 subsribe한다면?
  • mono나 flux같은 publisher들은 하나 이상의 subscriber를 가질수 있음.
  • Publisher source의 타입
    • hot: 구독하는 시점부터 실시간으로 발생하는 데이터만 가져온다.
    • cold: 어느 subscriber가 요청해도 동일한 결과가 고정되있음, 구독할때마다 데이터를 처음부터 다 끝까지 가져온다
  • 결과 => 로그가 두번찍힘.
• block()
  • block()메서드는 내부에서 subscribe를 함, 따라서 결과는 로그가 두번찍힘.
  • Api나 db호출이 있으면 block(), return할때 IO가 두번 일어날수 있으므로 비효율적, 따라서 가능한한 사용하지 않는것이 좋음.

[gaepury]

10편(Flux의 특징과 활용방법)

• Flux 간단 예제(Flux.just)
  • 
  • Flux.just 인자론 여러개
• Mono.just안에 List vs Flux.just에 인자 여러개
  • 
  • 결과는 같다.
  • Mono.just로 할시 List 안에 Entity들을 조작,가공할때 reactor 라이브러리에 있는 operator를 이용할수 없음.
• HTTP 스트림을 지원하려면 Flux
  • MediaType 변경
  • 
  • 
• Stream을 이용해서 가져오기 + 특정 개수만큼 가져오기(take)
  • 
• onNext로 오는것을 delay 걸기
  • 
  • delay는 백그라운드에서 blocking됨
• Flux의 generate 사용
  • generate(Consumer)
    • sink
      • 데이터를 흘려보내는 역할
    • 
  • generate(Callable, BiFunction)
    • 
    • 첫번쨰 인자: 초기 상태값
    • 두번째 인자: 상태값과 sink을 보내서 다음 상태값을 return
• 여러 Flux 결합(merge)하기(with Flux interval)
  • Flux.zip 사용
  • Interval flux의 delay만 이용하고 es flux의 데이터만 가져오기
    • 
  • id만들때 interval의 값 이용하기
    • 
  • zip의 효율적인 ex
    • 여러 서비스에서 데이터를 호출하고(사용자 프로파일 정보, 사용자 디테일정보등) 결합하여 제공