Spring WebFlux (2)

이전 글에 이어서 Project Reactor와 WebFlux에 관한 글로 마무리 하도록 하겠습니다.

Reqctive Programming 과 Reactive Streams에 관해 궁금한 분들은 이전글 참고 부탁드립니다.

 

Spring WebFlux (1)

 

 

목차

1. Reactive Programming
2. Reactive Streams 개요
   • Reactive Streams 개념
   • Reactive Streams 소개
3. Project Reactor
   • Project Reactor 소개
   • Flux와 Mono
   • Flux, Mono API 소개
4. WebFlux와 R2DBC, WebClient
   • Spring WebFlux
   • WebFlux App 구현 - Annotation방식, Fuctional Endpoint 방식
   • R2DBC
   • WebClient

 

---

 

3. Project Rector

Project Reactor 소개

• Project Reactor
  • Pivotal의 오픈소스 프로젝트로, JVM 위에서 동작하는 Non-Blocking 애플리케이션을 만들기 위한 Reactive 라이브러리입니다.
  • Project Reactor는 RxJava와 함께 Reactive Stream의 구현체이기도 하고, Spring Framework5 부터 리액티브 프로그래밍을 위해 지원되는 라이브러리입니다.
  • Project Reactor는 RxJava와 많은 공통점이 있으며 큰 차이점이 있다면 Reactor는 최소 Java8 이상에서 동작하며 Java8의 특징을 잘 지원 한다는 점입니다.

 

• reactor-core
  • Project Reactor의 코어 라이브러리입니다.
  • Reactive Streams의 구현체입니다.
  • 최소 java8 이상이 필요합니다.

 

• reactor-test
  • 테스트 도구입니다.
  • StepVerifier

 

• reactor-netty
  • spring boot 에서 자동으로 reactor netty를 기본 서버로 사용하도록 하고 있습니다.

 

 

Mono와 Flux

• Flux와 Mono는 Reactive Streams 인터페이스 중에서 데이터를 제공하는 발행자 역할을 하는 Publisher의 구현체입니다.
• 간단한게 Flux는 0 ~ N 개의 데이터를 전달하며 Mono는 0 ~ 1 개의 데이터를 전달합니다.

 

[ Flux ]

• Flux는 0 ~ N 개까지 다수의 응답을 만드는데 특화되어 있는 Publisher 구현체입니다.
• onComplete나 onError가 되기 전까지는 무한 생성가능한 Stream으로 생각해야 합니다.
• 데이터를 전달할때마다 onNext 이벤트를 발생시키고, 모든 데이터의 전달처리가 완료되면 onComplete, 데이터를 전달하는 과정에서 오류가 발생하면 onError로 종료됩니다.
• onNext, onComplete, onError는 모두 옵셔널하다는 것입니다. onNext는 없지만 onComplete는 있을 수도 있고, onComplete가 없는 경우도 있습니다.

 

 

 

 

[ Mono ]

• 최대 1개의 응답을 만드는데 특화되어 있는 Publisher 구현체입니다.
• Flux와는 다르게 데이터의 전달처리가 완료되면 onComplete나 onError가 실행됩니다.
• 하나의 응답결과만 리턴하면 되기 때문에 별도로 값은 필요없고, 완료개념만 있으면 되는 비동기 처리도 표현할 수 있습니다.

 

 

 

• 스트림을 하나의 결과로 모아줄 때 Mono를 쓰고, 각각의 Mono를 합쳐서 여러 개의 값을 처리할 때는 Flux를 사용합니다.
• Flux도 하나의 데이터만 담을 수 있지만 Mono는 이를 보장합니다. Flux의 경우 데이터를 Subscriber에게 전달하더라도 스트림의 종료는 선택적으로 일어나게 되지만, Mono는 데이터를 전달하고 스트림이 항상 종료됩니다.

 

Reactor 3 Reference Guide

 

 

 

Flux, Mono API 소개

[ Flux를 생성하는 방법 ]

• Flux의 subscribe()가 실행되기 전까지는 어떤 일도 발생하지 않습니다.
• Subscriber가 구독을 하는 과정은 Publisher에 정의된 subscribe() 메서드를 이용합니다.
• subscribe()의 인자로 전달되는 Consumer 객체는 Subscriber의 onNext() 이벤트가 발생 했을 때 실행됩니다.
• onSubscribe() → request() → onNext() → onNext() → onComplete()

 

Flux<String> flux = Flux.just("자바", "스칼라", "파이썬");
Flux<Integer> flux = Flux.range(1, 5);
Flux<String> flux = Flux.fromArray(new String[] {"자바", "스칼라", "파이썬"});
Flux<String> flux = Flux.fromIterable(Arrays.asList("자바", "스칼라", "파이썬");

 

[ 테스트 방법 ]

• Reactor-test 프로젝트에 존재하는 StepVerifier를 이용할 수 있습니다.
• Publisher를 구독하면서 예상값과 순서를 검증할 수 있습니다.
• create 메서드로 인스턴스를 생성할 수 있으며 반드시 verifyComplete() 메서드를 호출해야 합니다.

 

void expectFooBarComplete(Flux<String flux) {
    StepVerifier.create(flux)
                            .expectNext("foo")
                            .expectNext("bar")
                            .verifyComplete();
}

 

[ map() 과 filter() ]

• map() 메서드
  • map() 메서드는 1 - 1 방식으로 변환 해줍니다.
  • 자바 스트림의 map()과 유사합니다.

 

static void fluxMapTest() {
    Flux.just("a", "bc", "def", "wxz")
            .map(String::length)
            .subscribe(System.out::println);
}

// 결과
1
2
3
3

 

• filter() 메서드
  • filter()를 이용하면 시퀀스가 생성한 데이터를 걸러낼 수 있습니다.
  • filter에 전달한 함수의 결과가 true인 데이터만 전달하고 나머지 데이터는 전달하지 않습니다.

 

static void fluxFilterTest() {
    Flux.range(1, 10)
            .filter(num -> num % 2 == 0)
            .subscribe(x -> System.out.print(x + " -> "));
}

// 결과
2 -> 4 -> 6 -> 8 -> 10 ->

 

[ flatMap() ]

• flatMap()은 1개의 데이터로 부터 시퀀스를 생성할 때 사용한다. 즉 1 - n 방식의 변환을 처리한다.
• flatMap에 전달한 함수가 생성하는 각 Flux는 하나의 시퀀스처럼 연결된다. 그래서 flatMap()의 결과로 생성되는 Flux의 타입이 Flux<Flux>가 아니라 Flux이다.
• 아래 예제에서는 range별로 리스트로 묶은 Flux를 리턴하기 때문에 flatMap의 리턴 타입이 Flux<List>가 되는 것이다.

 

static void fluxFlatMapTest() {
    Flux<List<Integer>> flux = Flux.just(1, 2, 3)
            .flatMap(i -> Flux.range(1, i).collectList());

    flux.subscribe(System.out::println);
}

// 결과
[1]
[1, 2]
[1, 2, 3]

 

[ flatMap()과 map() 차이 ]

• flatMap()

<R> Flux<R> : flatMap(Function<? superT,? extendsPublisher<? extends R>> mapper)

Transform the elements emitted by this Flux asynchronously into Publishers, then flatten these inner publishers into a single Flux through merging, which allow them to interleave

 

• flatMap의 경우 공식문서에 따르면 내부에서 Publisher를 리턴하기 때문에 내부 로직이 비동기적으로 작동하는 것을 확인할 수 있습니다.
• flatMap의 내부에는 publisher가 리턴되어야 하지만 flatMap의 반환타입은 Flux이기 때문에 평탄화 작업이 발생합니다.

 

 

• map()

<V> Flux<V> : map(Function<? super T, ? extends V> mapper)

Transform the items emitted by this Flux by applying a synchronous function to each item, which may produce null values.

 

• map()의 경우는 동기적으로 함수가 진행되는 것을 알 수 있습니다.
• 또한 map함수가 리턴하는 값 그대로 Flux에 넣어줍니다.

 

 

• 아래는 map, flatMap의 비교 코드 입니다.
  • map과 flatMap의 리턴 값이 다른것을 알 수 있습니다.
  • 만약 map의 리턴 값이 Mono.just(el * 2) 였다면 map이 지난 후 스트림의 타입은 Flux<Mono> 형태일 것입니다.

 

static void fluxMapFlatMapTest() {
    Flux<Integer> flux = Flux.just(1, 2, 3)
            .map(el -> el * 2)
            .flatMap(el -> Mono.just(el * 2));

    flux.subscribe(System.out::println);
}

// 결과
4
8
12

 

• map, flatMap 참고 블로그

 

5 common mistakes of Webflux novices

 

 

 

[ defaultEmpty() ]

• 빈 시퀀스인 경우 기본 값을 사용할 수 있습니다.
• 시퀀스에 데이터가 없을 때 특정 값을 기본으로 사용하고 싶다면 defaultEmpty() 메서드를 사용하면 됩니다.
• Mono와 Flux 모두 defaultEmpty()를 제공합니다.

 

static void fluxDefaultEmpty() {
    Flux.empty().defaultIfEmpty(10)
            .subscribe(System.out::println);
}

// 결과
10

 

[ collect() ]

• 컬렉션으로 변환해주는 메서드입니다.
• collect() 메서드는 Flux에서 발생하는 데이터를 모아서 Collection 형태로 변환하여 Mono 형태로 스트림을 변환합니다.
• onNext 이벤트는 collect() 메서드에 의해 스트림의 모든 데이터들이 Collection으로 합쳐지기 전까지 발생하지 않습니다.

 

static void fluxCollect() {
    Flux.range(1, 3)
            .collect(() -> new ArrayList<Integer>(),
                    (collection, item) -> collection.add(item))
            .subscribe(
                    item -> System.out.println("[Subscriber] onNext : " + item),
                    e -> System.out.println("[Subscriber] onError : " + e.getMessage()),
                    () -> System.out.println("[Subscriber] onComplete")
            );
}

// 결과
[Subscriber] onNext : [1, 2, 3]
[Subscriber] onComplete

 

 

[ flatMapMany() ]

• Mono를 Flux로 변환해주는 메서드입니다.
• flatmap()이 Flux에서 Flux로, Mono에서 Mono를 생성하여 합칠 때 사용하는 것과는 다르게, Mono에서 Flux로 Publisher를 변환하여 합칠 때 사용합니다.
• flatMapMany() 메서드가 호출된 이후에 사용되는 Publisher를 Flux로 변환되어 사용됩니다.

 

static void fluxFlatMapMany() {
    Mono.just(1)
            .flatMapMany(item -> Flux.just(3, 2, 1))
            .subscribe(
                    item -> System.out.println("[Subscriber] onNext : " + item),
                    e -> System.out.println("[Subscriber] onError : " + e.getMessage()),
                    () -> System.out.println("[Subscriber] onComplete")
            );
}

// 결과
[Subscriber] onNext : 3
[Subscriber] onNext : 2
[Subscriber] onNext : 1
[Subscriber] onComplete

 

 

[ BackPressure ]

• subscribe 메서드에 Consumer 파라미터로 BaseSubscriber 를 커스텀해서 이용할 수 있습니다.
• 이때 hookOnNext에서 추가로 더 request하지 않는다면 처음 onSubscribe할 때 10개의 요청을 보내고 그 이후는 따로 동작하지 않습니다.
• 또한 onComplete 메서드도 실행되지 않습니다.
• 따라서 아래 코드에서는 10의 배수일때 10개 씩 요청을 추가하였고 마지막인 15를 처리한 후 onComplete까지 실행되는 것을 확인할 수 있습니다.

 

static void fluxBackPressureWithBaseSubscriber() {
    Flux.range(1, 15)
            .subscribe(new BaseSubscriber<>() {
                @Override
                protected void hookOnSubscribe(Subscription subscription) {
                    request(10);
                }

                @Override
                protected void hookOnNext(Integer value) {
                    System.out.println("onNext : " + value);
                    if (value % 10 == 0) {
                        System.out.println("onNext request");
                        request(10);
                    }
                }

                @Override
                protected void hookOnComplete() {
                    System.out.println("onComplete");
                }
            });
}

// 결과
onNext : 1
onNext : 2
onNext : 3
onNext : 4
onNext : 5
onNext : 6
onNext : 7
onNext : 8
onNext : 9
onNext : 10
onNext request
onNext : 11
onNext : 12
onNext : 13
onNext : 14
onNext : 15
onComplete

 

• 아래는 기본 request 세팅 값입니다.
  • Backpressure: Unbounded request
  • 첫 번째 요청은 구독 시점에서 최종 Subscriber로부터 온다. 요청을 하는 가장 간단한 방법은 즉시 구독 시점에서 unbounded request, 즉 request(Long.MAX_VALUE) 를 트리거하는 방법이다. Reactor에서는 다음과 같은 방식으로 unbounded request를 요청할 수 있다.
  • subscribe() 호출 (Consumer<? super Subscription> 람다식을 따로 정의하지 않으면 unbounded request)

 

3. Reactor Backpressure

 

 

[ zipWith() ]

• 2개의 시퀀스 병합
• zipWith() 메서드를 통해 다른 스트림과 병합하여 이벤트를 생성할 수 있습니다.
• 1부터 3까지 3개의 데이터를 갖는 스트림과 1부터 무한한 갯수의 데이터를 갖는 스트림이 zipWith() 메서드를 통해 병합되어도 결과를 보면 더 적은 데이터를 갖는 스트림에 맞춰 onNext 이벤트를 발생시킨다.

 

static void fluxZipWith() {
    Flux.range(1, 3)
            .map(i -> i * 10)
            .zipWith(Flux.range(1, Integer.MAX_VALUE),
                    (first, second) -> String.format("first Flux %d, Second Flux %d", first, second))
            .subscribe(
                    item -> System.out.println("[Subscriber] onNext : " + item),
                    e -> System.out.println("[Subscriber] onError : " + e.getMessage()),
                    () -> System.out.println("[Subscriber] onComplete")
            );
}

// 결과
[Subscriber] onNext : first Flux 10, Second Flux 1
[Subscriber] onNext : first Flux 20, Second Flux 2
[Subscriber] onNext : first Flux 30, Second Flux 3
[Subscriber] onComplete

 

 

[ zip() ]

• 여러 개의 시퀀스를 병합해줍니다.
• zipWith() 메서드는 한번에 결합하는 Publisher의 갯수가 2개로 한정이지만, zip()메서드는 여러 개의 시퀀스를 병합할 수 있습니다.
• zip() 메서드는 최대 8개의 Publisher를 결합할 수 있도록 인자가 최소 2개부터 8개까지 존재합니다.

 

static void fluxZip() {
    var flux1 = Flux.range(1, 15);
    var flux2 = Flux.range(1, 10).map(el -> el * 10);
    var flux3 = Flux.range(1, 5).map(el -> el * 100);

    Flux.zip(flux1, flux2, flux3)
            .subscribe(
                    item -> System.out.println("[Subscriber] onNext : " + item),
                    e -> System.out.println("[Subscriber] onError : " + e.getMessage()),
                    () -> System.out.println("[Subscriber] onComplete")
            );
}

// 결과
[Subscriber] onNext : [1,10,100]
[Subscriber] onNext : [2,20,200]
[Subscriber] onNext : [3,30,300]
[Subscriber] onNext : [4,40,400]
[Subscriber] onNext : [5,50,500]
[Subscriber] onComplete

 

 

[ concatWith() ]

• 시퀀스 순서대로 연결
• concatWith() 메서드를 사용하면 발생하는 데이터를 순서대로 연결할 수 있습니다.
• concatWith로 연결한 시퀀스는 이전 시퀀스가 종료된 뒤에 구독을 시작합니다.
• 아래 코드에서 seq1이 종료된 뒤에 seq2가 구독을 시작하고 seq2가 종료된 뒤에 seq3 구독을 시작합니다.

 

static void fluxConcatWith() {
    Flux<Integer> seq1 = Flux.just(1, 2, 3);
    Flux<Integer> seq2 = Flux.just(4, 5, 6);
    Flux<Integer> seq3 = Flux.just(7, 8, 9);

    seq1.concatWith(seq2).concatWith(seq3)
            .subscribe(el -> System.out.print(el + " -> "));
}

// 결과
1 -> 2 -> 3 -> 4 -> 5 -> 6 -> 7 -> 8 -> 9 ->

 

 

[ mergeWith() ]

• 시퀀스 발생한 순서대로 연결해줍니다.
• 시퀀스의 연결 순서가 아니라 시퀀스가 발생하는 데이터 순서대로 섞고 싶다면 mergeWith()를 사용합니다.
• 아래 코드에서 sleep 이 없으면 콘솔에 데이터가 찍히기 전에 끝나기 때문에 sleep을 넣었습니다.

 

static void fluxMergeWith() throws InterruptedException {
    Flux<String> tick1 = Flux.interval(Duration.ofSeconds(1)).map(tick -> tick + "초틱");
    Flux<String> tick2 = Flux.interval(Duration.ofMillis(700)).map(tick -> tick + "밀리초틱");
    tick1.mergeWith(tick2).subscribe(System.out::println);

    Thread.sleep(3000);
}

// 결과
0밀리초틱
0초틱
1밀리초틱
1초틱
2밀리초틱
3밀리초틱
2초틱

 

---

 

 

4. WebFlux와 R2DBC, WebClient

Spring WebFlux

• Spring WebFlux는 Spring 5에서 새롭게 추가된 모듈입니다.
• WebFlux는 클라이언트, 서버에서 reactive 스타일의 어플리케이션 개발을 도와주는 모듈이며 non-blocking에 reactive streams를 지원합니다.
• Spring WebFlux의 장 단점은 다음과 같습니다.
  • 장점 : 고성능, spring과 완벽한 통합, netty 지원, 비동기 non-blocking 메세지 처리
  • 단점 : 오류처리가 복잡합니다.

 

• Spring WebFlux를 사용하면 좋은 케이스
  • 마이크로 서비스 환경에 좋습니다.
  • 스트리밍 서비스 데이터를 사용하는 환경에 좋습니다.
  • 비동기 / Non-blocking API 호출
  • 비동기 / Non-blocking 데이터 액세스
  • 비동기 웹 요청 처리
  • Reactor(Flux, Mono), RxJava, Flow 등을 이용하는 코드

 

• Spring WebFlux를 사용할 필요가 없는 경우
  • WebFlux가 왜 필요한지 분명하게 모르는 경우
  • Spring MVC로 개발해도 아무 문제가 없는 경우
  • 블로킹 I/O가 서버, 코드, 라이브러리에 존재하는 경우
  • 리모트 서비스, API 호출이 전혀 없고, NoSQL을 사용하지 않는경우

 

Spring WebFlux 방식과 Spring MVC 방식 비교

• 아래는 Reactive Stack과 Servlet Stack 비교사진 입니다.
  • 스프링 부트에서는 기본으로 Netty로 설정되어 있습니다.
  • 아래는 Netty 관련 글입니다.

 

[Spring] Reactive Spring WebFlux: Tomcat과 Netty

 

 

 

WebFlux App 구현 - Annotation방식, Fuctional Endpoint 방식

• 기존 Annotation을 이용한 Spring MVC 방식

 

 

 

• Annotation을 이용한 Spring WebFlux 방식
  • 웹 서버로 들어온 요청이 HttpHandler에게 전달되면, HttpHandler는 전처리 후 webHandler에게 처리를 위임합니다.
  • WebHandler 내부에서는 HandlerMapping, HandlerAdapter, HandlerResultHandler 3가지 컴포넌트가 요청과 응답을 처리합니다.
  • 처리가 끝나면 HttpHandler는 후처리 후 응답을 종료합니다.
  • HandlerMapping, HandlerAdapter는 MVC가 사용하는 컴포넌트와 역할과 이름이 같지만, 동작 방식이 다르기 때문에 별도의 인터페이스를 사용합니다.

 

 

 

Spring WebFlux의 2가지 개발 방식

1. 기존 어노테이션을 이용하는 방식 (Annotated Controllers)

• @Controller, @RestController
• @GetMapping, @PostMapping 등의 기존 어노테이션 그대로 사용
• 위에서 본 흐름이 해당 방식의 흐름입니다.
• 응답 값으로 다양한 객체들을 전달할 수 있습니다.
  • ResponseEntity
  • Object
  • ServerSentEvent
  • Mono, View name
  • void, Mono, null

 

 

2. 새로운 함수형 모델 (Functional Web)

• HandlerFunction : 요청을 처리하는 함수를 의미합니다.
• RouterFunction : HandlerFunction에게 라우팅 처리를 하는 함수를 의미합니다.

 

라우터 함수(RouterFunction)는 RequestPredicate를 통해 클라이언트에서 들어온 request를 관리하는 라우터 역할을 하게 됩니다.

핸들러 함수(HandlerFunction)는 라우터 함수로 들어온 request에 대한 처리를 정의합니다.

 

 

• 함수형 방식에 대한 코드는 아래 블로그를 참고하면 도움될 것 같습니다.

 

Spring Webflux의 Functional Endpoints 사용법. (with RouterFunction)

 

 

 

R2DBC

• R2DBC(Reactive Relational Database Connectivity) 개념
  • R2DBC는 SQL 데이터베이스를 위한 리액티브 API 입니다.
  • Reactive Streams를 기반으로 SQL을 실행하는데 필요한 커넥션, 쿼리 실행, 트랜잭션 처리 등에 대한 API를 정의하고 있습니다.
  • JDBC API 처럼 R2DBC는 API만 정의하고 있고, r2dbc-spi 모듈이 SPI(Service-provider interface)로서 각 드라이버는 SPI에 정의된 인터페이스를 알맞게 구현하였습니다.

 

 

 

• R2DBC는 언제 사용해야 하는가??
  • Reactive Programming을 하는 과정에서 Database 사용이 필요한 경우에 사용합니다.
  • 많은 예제로 Project Reactor 기반의 Spring WebFlux를 사용하며 Database를 사용해야할 때 R2DBC를 많이 사용합니다.
• WebFlux에서 JPA를 사용하지 않는 이유
  • JPA는 기본적으로 비동기 처리를 제공하지 않습니다. 따라서 WebFlux에서 사용하게 된다면 DB 접근 부분에서 block되기 때문에 WebFlux를 사용할 이유가 사라지게 됩니다.
  • WebFlux같은 조금의 Thread를 계속해서 사용하는 경우 이런 block이 되는 작업은 비효율적이며 시스템에 영향이 갈 수 있기 때문에 JPA가 아닌 비동기 처리를 지원하는 R2DBC를 사용하는 것이 좋습니다.
• R2DBC의 장점
  • Blocking Programming일 때보다 높은 동시성이 요구되는 상황에서 더 좋은 성능을 낼 수 있습니다.
  • 아래 글을 참고하면 좋을 것 같습니다.

 

 

Spring: Blocking vs non-blocking: R2DBC vs JDBC and WebFlux vs Web MVC

 

 

• ReactiveCRUDRepository 인터페이스에 대한 사용 방법은 아래 자세히 나와있으니 해당 글을 참고하면 도움이 될 것 같습니다.

 

 

R2DBC Repositories

 

 

WebClient

• WebClient는 Spring WebFlux 라이브러리에 속하는 클라이언트 입니다.
• 기존에 Spring MVC에서는 RestTemplate을 주로 사용하지만, WebClient는 동기와 비동기를 모두 지원하기 때문에 RestTemplate을 대체 할 수 있습니다.
• 장기적으로 RestTemplate은 deprecated 된 상태이기 때문에 WebClient를 사용하는 것이 더 좋아보입니다.

 

Web on Reactive Stack

 

 

---

 

이것을 끝으로 이번 Spring WebFlux에 관한 글은 마무리하도록 하겠습니다.

 

개인적으로 교육도 괜찮았다고 생각했는데 이를 직접 블로그에 정리하면서 교육 자료 외에 정말 많은 자료들과 블로그들을 찾아가며 정리할 수 있는 시간이었습니다.

 

기존에 알고있던 내용보다 훨씬 Reactive Programming에 관한 개념을 잘 잡고 넘어갈 수 있어서 이를 계기로 후에 WebFlux를 도입할 경우가 생긴다면 잘 고려해보고 프로젝트를 진행하면 좋을 것 같습니다.