10. Mono & Flux

1. Mono

 

Mono로 감싸놓은것이 어떤의미인가?

• 아래의 코드는 중, mono 부분은 Subscriber가 subscribe 하는 순간에 실행됨

@GetMapping("/")
Mono<String> hello() {
    log.info("pos1");
    Mono<String> m = Mono.just("Hello WebFlux").log();
    log.info("pos2");
    return m;
}

결과 출력

 

pos1

pos2

onSubscribe()

request(unbounded) => backPressure

onNext(Hello WebFlux)

onComplete()

---

Mono.just() 는 준비된 결과를 필요로 하므로 아래의 순서로 수행

@GetMapping("/")
Mono<String> hello() {
    log.info("pos1");
    Mono<String> m = Mono.just(generateHello()).log();
    log.info("pos2");
    return m;
}

private String generateHello() {
    log.info("method generateHello()");
    return "Hello Webflux";
}

결과 출력

pos1

method generateHello()

pos2

onSubcribe()

...

---

Mono.supplier Subscribe 이후에 수행됨

@GetMapping("/")
Mono<String> hello() {
    log.info("pos1");
    Mono<String> m = Mono.fromSupplier(() -> generateHello()).log();
    log.info("pos2");
    return m;
}

private String generateHello() {
    log.info("method generateHello()");
    return "Hello Webflux";
}

결과 출력

pos1

pos2

onSubscribe()

request()

method generateHello()

---

Mono 나 Flux 는 여러개의 Subscriber를 가질 수 있다.

• Cold Type
  • 데이터가 만들어져서 고정이 돼있는 것
  • 어느 subscriber가 요청을해도 동일하게 Publishing 하는 경우
• Hot Type
  • 실시간에서 일어나는 외부의 이벤트들

---

block()

• 내부에서 subscribe() 를 수행
• publisher 가 제공하는 결과값을 Mono/Flux 컨테이너에서 제거하고 블로킹됨

---

2. Flux

 

결과값이 같다.

• Flux 는 하나하나의 엔티티레벨의 Operator 를 사용할 수 있음
• HttpStream 을 지원하려면 결과를 리턴할 때 chunk 단위로 데이터를 보내려면 flux가 좋다.

@GetMapping("/event/{id}")
Mono<List<Event>> event(@PathVariable long id) {
    List<Event> list = Arrays.asList(new Event(1L, "event1"), new Event(2L, "event2"));
    return Mono.just(list);
}

@GetMapping("/events")
Flux<Event> events() {
    return Flux.just(new Event(1L, "event1"), new Event(2L, "event2"));
}

 

@GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
Flux<Event> events() {
    Flux<Event> es = Flux
            .<Event, Long>generate(() -> 1L, (id, sink) -> {
                sink.next(new Event(id, "value" + id));
                return id + 1;
            });

    Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));

    // 한쌍씩 묶는다.
    return Flux.zip(es, interval)
            .map(tu -> tu.getT1());
}