SSE-Spring Webflux 실습

Spring Webflux 간단 설명

1. Spring Webflux란?

• Spring Webflux는 Spring Framework 5부터 추가된 비동기적인 반응형 프로그래밍 모델입니다.
• 기존의 Spring MVC와 달리 Servlet API를 사용하지 않고, Netty와 같은 서버를 사용하여 비동기적으로 요청을 처리합니다.
• Spring Webflux는 Mono와 Flux라는 Publisher를 사용하여 비동기적인 데이터 처리를 지원합니다.
• Reactive Streams 사양을 준수하고, Backpressure를 지원합니다.
  - Project Reactor 라이브러리를 사용하여 Publisher를 구현하고, 리액티브 스트림을 사용할 수 있습니다.

2. Spring Webflux의 장점

• 비동기적인 처리: 블로킹 I/O를 사용하지 않고 비동기적으로 요청을 처리하여 더 많은 요청을 처리할 수 있습니다.
• 반응형 프로그래밍: Publisher를 사용하여 데이터를 비동기적으로 처리하고, 리액티브 스트림을 사용하여 데이터를 처리할 수 있습니다.
• 함수형 프로그래밍: 람다식과 함수형 인터페이스를 사용하여 코드를 간결하게 작성할 수 있습니다.

3. 의존성 추가

gradle:

implementation 'org.springframework.boot:spring-boot-starter-webflux'

maven:

<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

실습

1. Flux와 Mono

• Project Reactor의 Mono와 Flux는 Publisher 인터페이스를 구현한 클래스로, 비동기적인 데이터 처리를 지원합니다.
• Mono: 0개 또는 1개의 데이터를 비동기적으로 반환
• Flux: 0개 이상의 데이터를 비동기적으로 반환하는 Publisher

예시:

// Flux는 1부터 5까지의 숫자를 1초 간격으로 반환하고, Mono는 특정 고객 정보를 반환합니다.
@GetMapping("/customer/{id}")
public Mono<Customer> findById(@PathVariable Long id) {
    return customerRepository.findById(id).log();
}

@GetMapping("/flux")
public Flux<Integer> flux() {
    return Flux.just(1, 2, 3, 4, 5).delayElements(Duration.ofSeconds(1)).log();
}

2. Sinks

• Project Reactor의 Sinks는 비동기적인 데이터 처리를 위한 클래스로, 데이터를 소비하거나 생성할 수 있습니다.

  private final Sinks.Many<Customer> sink = Sinks.many().multicast().onBackpressureBuffer();

  @GetMapping("/customer/sse")
  public Flux<ServerSentEvent<Customer>> findAllSSE() {
    return sink.asFlux().map(c -> ServerSentEvent.builder(c).build()).doOnCancel(() -> {
      sink.asFlux().blockLast();
    });
  }

• Sinks.Many는 다수의 데이터를 생성하거나 소비할 수 있습니다.
• multicast(): 다수의 Subscriber가 데이터를 소비할 수 있습니다.
• onBackpressureBuffer(): Backpressure를 지원합니다.
• asFlux(): Sinks를 Flux로 변환합니다.

blockLast() 메소드는 Flux를 블로킹하고, 마지막 데이터를 반환합니다.

• ServerSentEvent: SSE를 사용하여 데이터를 전송할 수 있습니다.

  @PostMapping("/customer")
  public Mono<Customer> save() {
    return customerRepository
        .save(new Customer("josh", "long"))
        .doOnNext(c -> sink.tryEmitNext(c))
        .log();
  }

• tryEmitNext(): Sinks에 데이터를 전송합니다.

브라우저에서 SSE 수신하기참고