Chapter 10. 리액터 개요

애플리케이션 코드 개발의 두가지 형태

명령형 - 순차적으로 연속되는 작업이며, 각 작업은 한 번에 하나씩 그리고 이전 작업 다음에 실행된다. 데이터는 모아서 처리되고 이전 작업이 데이터 처리를 끝낸 후에 다음 작업으로 넘어 갈 수 있다
리액티브 - 데이터 처리를 위해 일련의 작업들이 정의되지만, 이 작업들은 병렬로 실행될 수 있다. 그리고 각 작업은 부분 집합의 데이터를 처리할 수 있으며, 처리가 끝난 데이터를 다음 작업에 넘겨주고 다른 부분 집합의 데이터로 계속 작업할 수 있다.

10.1 리액티브 프로그래밍 이해하기

리액티브 프로그래밍은 명령형 프로그래밍의 대안이 되는 패러다임이다. 명령형 프로그래밍의 한계를 해결할 수 있기 때문이다. 이런 한계를 이해하면 리액티브 모델의 장점을 더 확실하게 이해할 수 있다.

명령형 프로그래밍의 특징

명령형 프로그래밍은 한 번에 하나씩 순서대로 진행된다. 그렇기에, 데이터베이스에 데이터를 쓰거나 가져오는 것과 같은 것이라면 이 작업이 완료될 때까지 아무 것도 할 수 없다. 따라서 이 작업을 수행하는 스레드는 차단되며 이는 낭비가 된다.

리액티브 프로그래밍의 특징

이에 반해 리액티브 프로그래밍은 함수적이면서 선언적이다. 즉, 순차적으로 수행되는 작업 단계를 나타낸 것이 아니라 데이터가 흘러가는 파이프라인이나 스트림을 포함한다. 그리고 이런 리액티브 스트림은 데이터 전체를 사용할 수 있을 때까지 기다리지 않고 사용 가능한 데이터가 있을 때마다 처리되므로 사실상 입력되는 데이터는 무한할 수 있다.

10.1.1 리액티브 스트림 정의하기

리액티브 스트림은 넷플릭스, 라이트벤드, 프보탈의 엔지니어들에 의해 2013년 말에 시작되었다. 리액티브 스트림은 차단되지 않는 백 프레셔를 가지는 비동기 스트림 처리의 표준을 제공하는 것이 목적이다.

백 프레셔: 데이터를 소비하는 컨슈머가 처리할 수 있는 만큼으로 전달 데이터를 제한함으로써 지나치게 빠른 데이터 소스로부터의 데이터 전달 폭주를 피할 수 있는 수단이다.

리액티브 스트림은 4개의 인터페이스인 Publisher(발행자), Subscriber(구독자), Subscription(구독), Processor(프로세서)로 요약할 수 있다.

Publisher→(0개 이상의 Processor)→Subscriber 구조

/**
Publisher interface
	subscribe() - Subscriber가 Publisher를 구독 신청하기 위한 메서드
**/
public interface Publisher<T> {
	void subscribe(Subscriber<? super T> subscriber);
}

/**
Subscriber interface
	onSubscribe() - 첫번째 수신 요청을 위한 메서드로 인자로 Subscription을 전달한다
	onNext() - Subscribe가 데이터 요청이 완료되면 추가로
						 Publisher에게 데이터를 요청하는 메서드
	onError() - 에러 발생시 호출되는 메서드
	onComplete() - Publisher가 작업 완료 후 호출하여
								 Subscriber에게 작업 완료 여부를 알려주는 메서드
**/
public interface Subscriber<T> {
	void onSubscribe(Subscription sub);
	void onnext(T item);
	void onError(Throwable ex);
	void onComplete();
}

/**
Subscription interface
	request() - Subscriber가 데이터를 요청하기 위한 메서드
	cancel() - Subscriber가 구독 취소 요청을 위한 메서드
**/
public interface Subscription {
	void request(long n);
	void cancel();
}

/**
Subscriber와 Publisher를 결합한 인터페이스
**/
public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {}

하지만, 리액티브 스트림 인터페이스는 스트림을 구성하는 기능이 없다. 이에 따라 프로젝트 리액터에서는 리액티브 스트림을 구성하는 API를 제공하여 리액티브 스트림 인터페이스를 구현하였다. 이번 장의 나머지에서는 프로젝트 리액터를 살펴볼 것이다.

10.2 리액터 시작하기

명령형 코드 vs 리액티브 코드

명령형

스레드에서 한 단계씩 차례대로 실행된다. 그리고 각 단계가 완료될 때까지 다음 단계로 이동하지 못하게 실행 중인 스레드를 막는다.

String name ="Craig";
String capitalName = name.toUpperCase();
String greeting = "Hello, " + capitalName + "!";
System.out.println(greeting);

리액티브

리액티브 코드가 단계별로 실행되는 것처럼 보이지만, 실제로는 데이터가 전달되는 파이프라인을 구성하는 것이다. 그리고 각 단계에서는 어떻게 하든 데이터가 변경된다. 또한, 각 오퍼레이션은 같은 스레드로 실행되거나 다른 스레드로 실행될 수 있다.

Mono.just("Craig")
		.map(n -> n.toUpperCase())
		.map(cn -> "Hello, " + cn + "!")
		.subscribe(System.out:println);

just() 오퍼레이션은 첫 번째 것을 생성한다. 첫번쨰 Mono가 값을 방출하면 이 값이 첫 번째 map() 오퍼레이션에 전달되어 대문자로 변경되고 다른 mono를 생성한다. 이렇게 생성된 두 번째 Mono가 데이터를 방출하면 이 데이터가 두 번째 map()에 전달되어 문자열 결합이 수행되며, 이 결과는 세 번째 Mono를 생성하는 데 사용된다. 그리고 끝으로 subscribe() 호출에서는 세 번째 Mono를 구독하여 데이터를 수신하고 출력한다.

Mono와 Flux는 리액티브의 두 가지 핵심 타입이다.두 개 모두 리액티브 스트림의 Publisher 인터페이스를 구현한 것이다.

Mono: 하나의 데이터 항목만 갖는 데이터셋에 최적화된 리액티브 타입

Flux: 0, 1 또는 다수의(무한일 수 있는) 데이터를 갖는 파이프라인

10.2.1 리액티브 플로우의 다이어그램

리액티브 플로우는 마블 다이어그램(marble diagram)으로 나타내곤 한다.

제일 윗 부분 - Flux나 Mono를 통해 전달되는 데이터의 타임라인
중앙 부분 - 오퍼레이션
제일 밑 부분 - Flux나 Mono의 타임라인

10.2.2 리액터 의존성 추가하기

<!-- 리액터를 사용하기 위해 필요한 라이브러리 -->
<dependency>
	<groupId>io.projectreactor</groupId>
	<artifactId>reactor-core</artifactId>
</dependency>

<!-- 리액터 테스트를 작성하기 위해 필요한 라이브러리 -->
<dependency>
	<groupId>io.projectreactor</groupId>
	<artifactId>reactor-test</artifactId>
	<scope>test</scope>
</dependency>

10.3 리액티브 오퍼레이션 적용하기

Flux와 Mono는 리액터가 제공하는 가장 핵심적인 구성 요소이다. 이 두가지를 이용하여 파이프라인을 생성하며, 이 두가지 내에는 500개 이상의 오퍼레이션이 있다.

생성 오퍼레이션 (creation)
조합 오퍼레이션 (combination)
변환 오퍼레이션 (transformation)
로직 오퍼레이션 (logic)

10.3.1 리액티브 타입 생성하기

객체로부터 생성하기

@Test
public void createAFlux_just() {
	Flux<String> fruitFlux = Flux
			.just("Apple", "Orange", "Grape", "Banana", "Strawberry");
}

fruitFlux.subscribe(
	f->System.out.println("Here's some fruit: " + f)
};

@Test
public void createAFlux_just() {
	Flux<String> fruitFlux = Flux
		.just("Apple", "Orange", "Grape", "Banana", "Strawberry");
  
	StepVerifier.create(fruitFlux)
		.expectNext("Apple")
		.expectNext("Orange")
		.expectNext("Grape")
		.expectNext("Banana")
		.expectNext("Strawberry")
		.verifyComplete();
}

컬렉션으로부터 생성하기

Flux는 배열, Iterable 객체, 자바 Stream 객체로부터 생성될 수 있다.

@Test
public void createAFlux_fromArray() {
  String[] fruits = new String[] {
      "Apple", "Orange", "Grape", "Banana", "Strawberry" };
  
	// fromArray()를 활용한 Flux 생성
  Flux<String> fruitFlux = Flux.fromArray(fruits);
  
  StepVerifier.create(fruitFlux)
      .expectNext("Apple")
      .expectNext("Orange")
      .expectNext("Grape")
      .expectNext("Banana")
      .expectNext("Strawberry")
      .verifyComplete();
}

@Test
public void createAFlux_fromIterable() {
  List<String> fruitList = new ArrayList<>();
  fruitList.add("Apple");
  fruitList.add("Orange");
  fruitList.add("Grape");
  fruitList.add("Banana");
  fruitList.add("Strawberry");
  
	// fromIterable()를 활용한 Flux 생성
  Flux<String> fruitFlux = Flux.fromIterable(fruitList);
  
  StepVerifier.create(fruitFlux)
      .expectNext("Apple")
      .expectNext("Orange")
      .expectNext("Grape")
      .expectNext("Banana")
      .expectNext("Strawberry")
      .verifyComplete();
}

@Test
public void createAFlux_fromStream() {
 Stream<String> fruitStream = 
      Stream.of("Apple", "Orange", "Grape", "Banana", "Strawberry");
 
 // fromStream()을 활용한 Flux 생성
 Flux<String> fruitFlux = Flux.fromStream(fruitStream);
  
 StepVerifier.create(fruitFlux)
     .expectNext("Apple")
     .expectNext("Orange")
     .expectNext("Grape")
     .expectNext("Banana")
     .expectNext("Strawberry")
     .verifyComplete();
}

데이터 없이 매번 새 값으로 숫자를 방출하는 Flux만 필요한 경우가 있다.

range()를 활용한 데이터 생성

@Test
public void createAFlux_range() {
 Flux<Integer> rangeFlux = 
     Flux.range(1, 5);
 
 StepVerifier.create(intervalFlux)
     .expectNext(1)
     .expectNext(2)
     .expectNext(3)
     .expectNext(4)
     .expectNext(5)
     .verifyComplete();
}

interval()를 활용한 데이터 생성

시작 값과 종료 값대신 값이 방출되는 시간 간격이나 주기를 지정한다.

@Test
public void createAFlux_interval() {
 Flux<Long> intervalFlux = 
     Flux.interval(Duration.ofSeconds(1))
         .take(5);
 
 StepVerifier.create(intervalFlux)
     .expectNext(0L)
     .expectNext(1L)
     .expectNext(2L)
     .expectNext(3L)
     .expectNext(4L)
     .verifyComplete();
}

10.3.2 리액티브 타입 조합하기

두 개 이상의 리액티브 타입으로 결합이나 분할하는 오퍼레이션

리액티브 타입 결합하기

두 개의 Flux 스트림을 mergeWith() operation을 사용하여 결합할 수 있다.

@Test
public void mergeFluxes() {
  
  // Flux는 가능한 빨리 데이터를 방출한다.
  // 번갈아 구독함을 보여주기 위해 delayElements를 사용했다.
  
  Flux<String> characterFlux = Flux
      .just("Garfield", "Kojak", "Barbossa")
      .delayElements(Duration.ofMillis(500));
  Flux<String> foodFlux = Flux
      .just("Lasagna", "Lollipops", "Apples")
      .delaySubscription(Duration.ofMillis(250))
      .delayElements(Duration.ofMillis(500));
  
  Flux<String> mergedFlux = characterFlux.mergeWith(foodFlux);

  StepVerifier.create(mergedFlux)
      .expectNext("Garfield")
      .expectNext("Lasagna")
      .expectNext("Kojak")
      .expectNext("Lollipops")
      .expectNext("Barbossa")
      .expectNext("Apples")
      .verifyComplete();
}

mergeWith()는 Flux들의 값이 완벽하게 번갈아 방출되게 보장할 수 없으므로 필요하다면 zip() 오퍼레이션을 대신 사용할 수 있다.

zip() 오퍼레이션은 번갈아 가져와 새로운 flux를 생성한다.

@Test
  public void zipFluxes() {
    Flux<String> characterFlux = Flux
        .just("Garfield", "Kojak", "Barbossa");
    Flux<String> foodFlux = Flux
        .just("Lasagna", "Lollipops", "Apples");
    
    Flux<Tuple2<String, String>> zippedFlux = 
        Flux.zip(characterFlux, foodFlux);
    
    StepVerifier.create(zippedFlux)
          .expectNextMatches(p -> 
              p.getT1().equals("Garfield") && 
              p.getT2().equals("Lasagna"))
          .expectNextMatches(p -> 
              p.getT1().equals("Kojak") && 
              p.getT2().equals("Lollipops"))
          .expectNextMatches(p -> 
              p.getT1().equals("Barbossa") && 
              p.getT2().equals("Apples"))
          .verifyComplete();
  }

zippedFlux로부터 방출되는 각 항목은 Tuple2(두 개의 다른 객체를 전달하는 컨테이너 객체), 각 소스 Flux가 순서대로 방출하는 항목을 포함한다.

만일 Tuple2가 아닌 다른 타입을 사용하고 싶다면 우리가 원하는 객체를 생성하는 함수를 zip()에게 제공하면 된다

@Test
  public void zipFluxesToObject() {
    Flux<String> characterFlux = Flux
        .just("Garfield", "Kojak", "Barbossa");
    Flux<String> foodFlux = Flux
        .just("Lasagna", "Lollipops", "Apples");
    
    Flux<String> zippedFlux = 
        Flux.zip(characterFlux, foodFlux, (c, f) -> c + " eats " + f);
    
    StepVerifier.create(zippedFlux)
          .expectNext("Garfield eats Lasagna")
          .expectNext("Kojak eats Lollipops")
          .expectNext("Barbossa eats Apples")
          .verifyComplete();
  }

먼저 값을 방출하는 리액티브 타입 선택하기

두 개의 Flux를 결합하는 대신 먼저 값을 방출하는 소스 Flux의 값을 발행하는 Flux는 first() 오퍼레이션을 사용할 수 있다.

@Test
public void firstFlux() {
  // delay needed to "slow down" the slow Flux
  
  Flux<String> slowFlux = Flux.just("tortoise", "snail", "sloth")
        .delaySubscription(Duration.ofMillis(100));
  Flux<String> fastFlux = Flux.just("hare", "cheetah", "squirrel");
  
  Flux<String> firstFlux = Flux.first(slowFlux, fastFlux);
  
  StepVerifier.create(firstFlux)
      .expectNext("hare")
      .expectNext("cheetah")
      .expectNext("squirrel")
      .verifyComplete();
}

10.3.3 리액티브 스트림의 변환과 필터링

리액티브 타입으로부터 데이터 필터링하기

앞에서 원하는 개수의 항목을 무시하는 것은 skip() 오퍼레이션을 사용한다.

@Test
public void skipAFew() {
  Flux<String> countFlux = Flux.just(
      "one", "two", "skip a few", "ninety nine", "one hundred")
      .skip(3);
 
  StepVerifier.create(countFlux)
      .expectNext("ninety nine", "one hundred")
      .verifyComplete();
}

특정 시간이 지난 후에 오는 데이터로 Flux를 생성하기 위해선 skip() 안에 Duration을 매개변수로 지정할 수 있다.

@Test
public void skipAFewSeconds() {
  Flux<String> countFlux = Flux.just(
      "one", "two", "skip a few", "ninety nine", "one hundred")
      .delayElements(Duration.ofSeconds(1))
      .skip(Duration.ofSeconds(4));
 
  StepVerifier.create(countFlux)
      .expectNext("ninety nine", "one hundred")
      .verifyComplete();
}

반대로 특정 개수만큼만 먼저 받고 싶다면 take()를 특정 시간동안의 데이터만 사용해서 flux를 만들기 위해선 take(Duration)을 사용할 수 있다.

@Test
  public void take() {
    Flux<String> nationalParkFlux = Flux.just(
        "Yellowstone", "Yosemite", "Grand Canyon", "Zion", "Acadia")
        .take(3);
   
    StepVerifier.create(nationalParkFlux)
        .expectNext("Yellowstone", "Yosemite", "Grand Canyon")
        .verifyComplete();
  }
  
  @Test
  public void takeForAwhile() {
    Flux<String> nationalParkFlux = Flux.just(
        "Yellowstone", "Yosemite", "Grand Canyon", "Zion", "Grand Teton")
        .delayElements(Duration.ofSeconds(1))
        .take(Duration.ofMillis(3500));
   
    StepVerifier.create(nationalParkFlux)
        .expectNext("Yellowstone", "Yosemite", "Grand Canyon")
        .verifyComplete();
  }

리액티브 데이터 매핑하기

Flux나 Mono에 가장 많이 사용하는 오퍼레이션 중 하나는 발행된 항목을 다른 형태나 타입으로 매핑(변환)하는 것이다. 다른 형태나 타입으로 매핑하기 위해 map()과 flatMap() 오퍼레이션을 사용한다.

@Test
  public void map() {
    Flux<Player> playerFlux = Flux
      .just("Michael Jordan", "Scottie Pippen", "Steve Kerr")
      .map(n -> {
        String[] split = n.split("\\s");
        return new Player(split[0], split[1]);
      });
    
    StepVerifier.create(playerFlux)
        .expectNext(new Player("Michael", "Jordan"))
        .expectNext(new Player("Scottie", "Pippen"))
        .expectNext(new Player("Steve", "Kerr"))
        .verifyComplete();
  }

map()을 활용하면 매핑 수행 방식이 동기적으로 수행된다. 따라서, 비동기적으로 사용하려면 flatMap()을 사용해야 한다.

단순히 flatMap()을 사용하면 동기적으로 실행된다. subscribeOn()을 사용하여 구독이 동시적으로 처리되어야 한다는 것을 지정했다. (동시성 모델을 subscribeOn()의 인자로 지정하였다)

Schedule의 동시성 모델

immeidate() - 현재 스레드에서 구독 실행
single() - 단일의 재사용 가능한 스레드에서 구독 실행. 모든 호출자에 대해 동일한 스레드를 재사용함
newSingle() - 매 호출마다 전용 스레드에서 구독 실행
elastic() - 무한하고 신축성 있는 풀에서 가져온 작업 스레드에서 구독을 실행. 필요 시 새로운 작업 스레드가 생성되며, 유휴 스레드는 제거된다.
parallel() - 고정된 크기의 풀에서 가져온 작업 스레드에서 구독을 실행하며, CPU 코어의 개수가 크기가 된다.

리액티브 스트림의 데이터 버퍼링하기

Flux를 통해 전달되는 데이터를 처리하는 동안 데이터 스트림을 작은 덩어리로 분할하면 도움이 될 수 있다. 이때 buffer() 오퍼레이션을 사용할 수 있다.

@Test
  public void buffer() {
    Flux<String> fruitFlux = Flux.just(
        "apple", "orange", "banana", "kiwi", "strawberry");
    
    Flux<List<String>> bufferedFlux = fruitFlux.buffer(3);
    
    StepVerifier
        .create(bufferedFlux)
        .expectNext(Arrays.asList("apple", "orange", "banana"))
        .expectNext(Arrays.asList("kiwi", "strawberry"))
        .verifyComplete();
  }

위 코드는 Flux로부터 리액티브가 아닌 List 컬렉션으로 버퍼링되는 값은 비생산적인 것처럼 보인다. 그러나 buffer()를 flatMap()과 같이 사용하면 각 List 컬렉션을 병행 처리 할 수 있다.

@Test
  public void bufferAndFlatMap() throws Exception {
    Flux.just(
        "apple", "orange", "banana", "kiwi", "strawberry")
        .buffer(3)
        .flatMap(x -> 
          Flux.fromIterable(x)
            .map(y -> y.toUpperCase())
            .subscribeOn(Schedulers.parallel())   
            .log()
        ).subscribe();
  }

만약 어떤 이유로든 Flux가 방출하는 모든 항목을 List로 모을 필요가 있다면 인자를 전달하지 않고 buffer()를 호출하면 된다.

Flux<List<String>> bufferedFlux = fruitFlux.buffer();

10.3.4 리액티브 타입에 로직 오퍼레이션 수행하기

Mono나 Flux가 발행한 항목이 어떤 조건과 일치하는지만 알아야 할 경우가 있다. 이때는 all() 이나 any() 오퍼레이션이 그런 로직을 수행한다.

@Test
public void all() {
  Flux<String> animalFlux = Flux.just(
      "aardvark", "elephant", "koala", "eagle", "kangaroo");
  
  Mono<Boolean> hasAMono = animalFlux.all(a -> a.contains("a"));
  StepVerifier.create(hasAMono)
    .expectNext(true)
    .verifyComplete();
  
  Mono<Boolean> hasKMono = animalFlux.all(a -> a.contains("k"));
  StepVerifier.create(hasKMono)
    .expectNext(false)
    .verifyComplete();
}

@Test
public void any() {
  Flux<String> animalFlux = Flux.just(
      "aardvark", "elephant", "koala", "eagle", "kangaroo");
  
  Mono<Boolean> hasAMono = animalFlux.any(a -> a.contains("a"));
  
  StepVerifier.create(hasAMono)
    .expectNext(true)
    .verifyComplete();
  
  Mono<Boolean> hasZMono = animalFlux.any(a -> a.contains("z"));
  StepVerifier.create(hasZMono)
    .expectNext(false)
    .verifyComplete();
}

요약

리액티브 프로그래밍에서는 데이터가 흘러가는 파이프라인을 생성한다.
리액티브 스트림은 Publisher, Subscriber, Subscription, Transformer의 네 가지 타입을 정의한다.
프로젝트 리액터는 리액티브 스트림을 구현하며, 수많은 오퍼레이션을 제공하는 Flux와 Mono의 두 가지 타입으로 스트림을 정의한다.
스프링 5는 리액터를 사용해서 리액티브 컨트롤러, 리퍼지터리, REST 클라이언트를 생성하고 다른 리액티브 프레임워크를 지원한다.

chapter10 sijun - JAVA-JIKIMI/SPRING-IN-ACTION-5 GitHub Wiki

Chapter 10. 리액터 개요

애플리케이션 코드 개발의 두가지 형태

10.1 리액티브 프로그래밍 이해하기

명령형 프로그래밍의 특징

리액티브 프로그래밍의 특징

10.1.1 리액티브 스트림 정의하기

10.2 리액터 시작하기

명령형 코드 vs 리액티브 코드

명령형

리액티브

10.2.1 리액티브 플로우의 다이어그램

10.2.2 리액터 의존성 추가하기

10.3 리액티브 오퍼레이션 적용하기

10.3.1 리액티브 타입 생성하기

객체로부터 생성하기

컬렉션으로부터 생성하기

10.3.2 리액티브 타입 조합하기

리액티브 타입 결합하기

먼저 값을 방출하는 리액티브 타입 선택하기

10.3.3 리액티브 스트림의 변환과 필터링

리액티브 타입으로부터 데이터 필터링하기

리액티브 데이터 매핑하기

리액티브 스트림의 데이터 버퍼링하기

10.3.4 리액티브 타입에 로직 오퍼레이션 수행하기

요약

⚠️ GitHub.com Fallback ⚠️

chapter10 sijun - JAVA-JIKIMI/SPRING-IN-ACTION-5 GitHub Wiki

Chapter 10. 리액터 개요

애플리케이션 코드 개발의 두가지 형태

10.1 리액티브 프로그래밍 이해하기

명령형 프로그래밍의 특징

리액티브 프로그래밍의 특징

10.1.1 리액티브 스트림 정의하기

10.2 리액터 시작하기

명령형 코드 vs 리액티브 코드

명령형

리액티브

10.2.1 리액티브 플로우의 다이어그램

10.2.2 리액터 의존성 추가하기

10.3 리액티브 오퍼레이션 적용하기

10.3.1 리액티브 타입 생성하기

객체로부터 생성하기

컬렉션으로부터 생성하기

10.3.2 리액티브 타입 조합하기

리액티브 타입 결합하기

먼저 값을 방출하는 리액티브 타입 선택하기

10.3.3 리액티브 스트림의 변환과 필터링

리액티브 타입으로부터 데이터 필터링하기

리액티브 데이터 매핑하기

리액티브 스트림의 데이터 버퍼링하기

10.3.4 리액티브 타입에 로직 오퍼레이션 수행하기

요약

⚠️ **GitHub.com Fallback** ⚠️

⚠️ GitHub.com Fallback ⚠️