[Spring Webflux] 2. Mono and Flux

23 Jun 2019 spring    

Mono and Flux

  • org.reactivestreams.Publisher의 구현체
  • 시퀀스를 제공하는 발행자 역할
  • Mono : 0-1 개의 데이터 전달
  • Flux : 0-N개의 데이터 전달

간단하게 사용해보기

Mono

empty()

Mono data = Mono.empty();

just()

Mono data = Mono.just("");

error()

Mono.error(RuntimeException::new);

위 내용에 대한 전체소스
@Test
public void mono() {
    Mono.empty().subscribe(s -> log.info("empty() : {}", s)); //출력안됨

    Mono.just("Test Value")
        .subscribe(s -> log.info("just() : {}", s));

    Mono.error(RuntimeException::new)
        .doOnError(e -> log.error(e.getMessage(), e))
        .subscribe(s -> log.info("error() : {}", s));
}

Flux

  • 한개의 시퀀스가 전달 될 때마다 doOnNext 이벤트 발생
  • 모든 데이터가 전달 완료되면 doOnComplete 이벤트 발생
  • 전달 과정에서 오류가 발생하면 doOnError 이벤트발생
  • Publisher는 subscribe가 되었을 경우에만 데이터를 전달함. subscribe() 호출 필요
    • 매개변수로 넣을 수 있는 Consumer의 경우 doOnNext가 호출 될 때마다 실행되는 함수
    • 소스 : 아래 소스 .subscribe(); 를 제거하게되면, 실행되지 않는다.
@Test
public void test() {
	Flux.just(1, 2, 3)
		.doOnNext(i -> log.info("doOnNext : {} ", i))
		.subscribe();
}

- 결과

xxx.FluxTestCase - doOnNext : 1 
xxx.FluxTestCase - doOnNext : 2 
xxx.FluxTestCase - doOnNext : 3
empty()

Flux data = Flux.empty();

just()

Flux data = Flux.just("", "");

range()

Flux.range(0, 10);

fromArray(), fromIterable(), fromStream()

Flux.fromArray(new String[]{“value 1”, “value 2”, “value 3”}); Flux.fromIterable(Arrays.asList(“value 1”, “value 2”, “value 3”)); Flux.fromStream(IntStream.range(0, 10).boxed())

위 내용에 대한 전체소스
  • log() 함수를 붙여 로그를 확인 할 수 있다.
@Test
public void flux() throws InterruptedException {
    Flux.empty().subscribe(s -> log.info("empty() : {}", s)); //출력안됨
    Flux.just("value 1", "value 2", "value 3")
        .subscribe(s -> log.info("just() : {}", s));

    Flux.range(0, 10)
        .subscribe(s -> log.info("range() : {}", s));

    Flux.fromArray(new String[]{"value 1", "value 2", "value 3"})
        .subscribe(s -> log.info("fromArray() : {}", s));

    Flux.fromIterable(Arrays.asList("value 1", "value 2", "value 3"))
        .subscribe(s -> log.info("fromIterable() : {}", s));

    Flux.fromStream(IntStream.range(0, 10).boxed())
        .subscribe(s -> log.info("fromStream() : {}", s));

    Flux.interval(Duration.ofMillis(100))
        .map(item -> "tick: " + item)
        .take(10)
        .subscribe(s -> log.info("fromIterable() : {}", s));

    Thread.sleep(150);

    Flux.error(RuntimeException::new)
        .doOnError(e -> log.error(e.getMessage(), e))
        .subscribe(s -> log.info("error() : {}", s));
}

Related Posts