Spring reactor 2.1.2 (netty 0.8.4) Mono.zip readTimeoutException 문제

web/Spring|2020. 7. 12. 20:34

 

Mono zip


각 Mono 처리 스레드를 병렬로 실행하고 이를 묶어서 사용할 수 있는게 Mono.zip이다. 

근데 Mono zip에서 병렬로 실행되는 작업 중 하나가 empty 또는 error가 발생 되면 바로 error 또는 complete를 내뱉게 되어있다. 하지만 각 Mono 구독 작업에 error와 empty 발생 시 문제에 대해 fallback 처리를 해주면 에러가 발생하더라도 그 로직을 타게 되어있다. 

 

하지만 2.1.2(netty 0.8.4) 버전을 사용하고 있을 때 호출 체인에서 첫 번째 요청의 실패 이후에 두 번째 요청이 정상적으로 이루어 지지 않아서 readTimeout이 발생되는 문제를 경험하였습니다.

 

이 문제를 해결하기 위해서 알아보던 중 2.1.2버전에 문제가 있는 것을 알게 되어 테스트를 해봤다.

 

 

테스트


아래 Mono.zip을 보면 두 개의 Mono 구독 작업을 병렬로 진행하도록 지정해놨고 각 작업 종료 후 response에 대한 부분을 출력하도록 해놨다.

public Mono<WedulResponse> circuitTest(WedulRequest request) {
    return Mono.zip(
        wedulClient.isWedulExist(request)
            .doOnError(e -> log.error("service error", e))
            .defaultIfEmpty(WedulResponse.builder().type("Error Return").build())
            .onErrorReturn(WedulResponse.builder().type("Error Return").build()),
        wedulTestClient.isWedulTestExist(request)
    ).map(
        d -> {
            System.out.println(d.getT2().getPage());
            System.out.println(d.getT1().getType());
            return WedulResponse.builder().isExist(d.getT1().isExist()).build();
        }
    ).doOnError(e -> log.error("error {}", e));
}

이 때 첫 번째 요청은 error가 발생하거나 empty 응답이 발생했을 때 기본값을 주도록 하고 socket timeout의 값은 1ms로 극단적으로 무조건 타임아웃이 나도록 지정해 놨다.

 

그리고 두 번째 요청http://wedul.space에서 사용중인 정상적인 api를 호출하도록 하였고 socket timeout 시간도 3000ms로 아주 넉넉하게 주었고 실제로 타임아웃이 날 이유가 없다.

 

그럼 정상적인 테스트 결과라고 한다면 아래와 같이 정상적인 응답이 와야한다. (page는 무조건 5로 나오게 지정해놨다.)

5
Error Return

 

 

실제로 응답은 예상된 대로 잘 왔다. 하지만 간혈적으로 아래와 같은 readTimeout exception이 별도로 계속 떨어졌다.

2020-07-12 20:17:30.998 ERROR 12214 --- [ctor-http-nio-3] r.netty.http.client.HttpClientConnect    : [id: 0x029e73cc, L:/127.0.0.1:63695 - R:localhost/127.0.0.1:8081] The connection observed an error

io.netty.handler.timeout.ReadTimeoutException: null

 

그래서 왜 그럴까 하고 검색을 해보니 2.1.2버전에 문제가 있어서 버전업을 하면 해결된다고 들었다. 

https://stackoverflow.com/questions/56048216/spring-webflux-timeout-with-multiple-clients

 

Spring webflux timeout with multiple clients

I have a service that interacts with a couple of other services. So I created separate webclients for them ( because of different basepaths). I had set timeouts for them individually based on https://

stackoverflow.com

 

그래서 버전을 2.2.4버전으로 업데이트하고 다시 테스트 해봤다. 실제로 아까 발생했던 readTimeoutException 문제는 더 발생하지 않았다.

 

 

실제 코드가 어떤 부분이 문제였는지는 찾지 못했지만 그래도 문제는 해결되어서 다행이다.

 

테스트 코드

https://github.com/weduls/circuit_breaker_test

댓글()

Spring5 리액티브 스트림 정리 및 api 전달 방식 정리

web/Spring|2019. 8. 16. 22:22

리액티브 또는 리액티브 스트림은 오늘날 spring framework에서 뜨거운 토픽으로 자리잡고 있다. 

그래서 나도 이전 포스팅에서도 정리도 하고 했었는데 아직 확실히 개념이 서질 않아서 다시 정리해봤다.

 

리액티브 스트림 (Reactive Stream) 이란?


리액티브 스트림은 무엇인가? 정확하게 공식문서에는 다음과 같이 기록되어 있다. (https://www.reactive-streams.org/)
Reactive Streams is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure.This encompasses efforts aimed at runtime environments (JVM and JavaScript) as well as network protocols.

이런 Reactive stream을 spring5에서 포함되었다.

  • Spring core framework는 Reactor와 RxJava를 통해 built-in 리액티브 프로그램을 할 수 있는 새로운 spring-flux 모듈을 추가하였다.
  • Spring security 5도 또한 reactive feature를 추가했다.
  • Spring Data umbrella project에서 Spring Data Commons에 새로운 ReactiveSortingRepository가 추가되었는데 가장먼저 redis, mongo, cassandra가 reactive에 지원한다. 불항하게도 일반적인 JDBC 드라이버의 블록킹 프로세스를 할 수 밖에 없는 디자인 때문에 Spring Data JPA는 이 특징에서 이점이 없다.
  • Spring Session또한 reactive feature를 추가하였고 2.0.0.M3qnxj SessionRepository내에 추가되었다.

 

Webflux 어플리케이션 만들기


스프링5를 통해서 reactive 프로그램을 만들어보면서 서비스를 확인해보자.

필요한 라이브러리
spring-boot-starter-parent
spring-webflux
jackson-databind
reactor-core
logback
lombok

데이터를 주고 받을 entity Post 객체

package com.study.webflex.dto;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * spring-boot-study
 *
 * @author wedul
 * @since 2019-08-14
 **/
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class Post {
    private int id;
    private String title;
    private String content;
}

 

데이터를 전달받을 Repository 클래스 DataRepository
- 우선 당장 데이터베이스를 선택하지 않고 이해를 먼저 돕기 위해서 가짜 데이터를 미리 static 블록을 이용해서 넣어놓자.

package com.study.webflex.dao;

import com.study.webflex.dto.Post;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

/**
 * spring-boot-study
 *
 * @author wedul
 * @since 2019-08-14
 **/
@Repository
public class PostRepository {

  private static final Map<Integer, Post> DATA = new HashMap<>();
  private static int ID_COUNTER = 0;

  static {
    // initial data
    Arrays.asList("First Post", "Second Post")
      .stream()
      .forEach(title -> {
          int id = ID_COUNTER++;
          DATA.put(id, Post.builder().id(id).title(title).content("content of " + title).build());
        }
      );
  }

  Flux<Post> findAll() {
    return Flux.fromIterable(DATA.values());
  }

  Mono<Post> findById(Long id) {
    return Mono.just(DATA.get(id));
  }

  Mono<Post> createPost(Post post) {
    int id = ID_COUNTER++;
    post.setId(id);
    DATA.put(id, post);
    return Mono.just(post);
  }

}

WebFlux를 사용하기 위한 어노테이션 @EnableWebFlux와 @Configuration을 달아준다.


WebFluxApi
webFlux는 기존 mvc또한 지원하기 때문에 아래와 같이 Controller를 만들어 엔드포인트를 정의하여 사용할 수 있다. 내부에서는 HttpServletRequest, HttpServletResponse객체 대신 ServerHttpRequest와 ServerHttpResponse 객체로 동작한다.

package com.study.webflex.controller;

import com.study.webflex.dto.Post;
import com.study.webflex.service.PostService;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * spring-boot-study
 *
 * @author wedul
 * @since 2019-08-15
 **/
@RestController
@RequiredArgsConstructor
public class PostController {

  private final PostService postService;

  @GetMapping(value = "")
  public Flux<Post> all() {
    return postService.findAll();
  }

  @GetMapping(value = "/{id}")
  public Mono<Post> get(@PathVariable(value = "id") int id) {
    return postService.findById(id);
  }

  @PostMapping(value = "")
  public Mono<Post> create(Post post) {
    return postService.createPost(post);
  }

}

실행하면 원하는 데이터를 추출해서 볼 수있다.

webflux mvc로 출력된 결과

그리고 또다른 형태로도 사용할 수 있게 제공하는데 RouterFunction과 HandelrFunction을 정의해서 구현해야한다.

HandlerFunction은 http요청을 ServletRequest객체로 가져와서 Mono형태로 값을 반환하고 RouterFunction은 http요청을 HandlerFunction으로 다시 Mono의 형태로 라우팅해준다.

우선 요청을 받아서 작업을 진행할 Handler를 정의한다.

package com.study.webflex.handler;

import com.study.webflex.dto.Post;
import com.study.webflex.service.PostService;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Mono;

/**
 * spring-boot-study
 *
 * @author wedul
 * @since 2019-08-15
 **/
@Component
@RequiredArgsConstructor
public class PostHandler {

  private final PostService postService;

  public Mono<ServerResponse> all(ServerRequest serverRequest) {
    return ServerResponse.ok().body(postService.findAll(), Post.class);
  }

  public Mono<ServerResponse> get(ServerRequest serverRequest) {
    String id = serverRequest.path();
    return ServerResponse.ok().body(postService.findById(Integer.valueOf(id)), Post.class);
  }

  public Mono<ServerResponse> create(ServerRequest serverRequest) {
    return serverRequest.bodyToMono(Post.class).doOnNext(post -> postService.createPost(post)).then(ServerResponse.ok().build());
  }

}

그리고 RouterFunction을 정의하자 하나의 route를 정의하여 엔드포인트를 지정할 수 있고 추가적으로 andRoute를 통해 연속적으로 지정할수도 있다.

/** * spring-boot-study * * *@author *wedul * *@since *2019-08-15 **/@ComponentScan@EnableWebFlux@Configurationpublic class WebFluxConfig {

  @Bean  
  public RouterFunction<?> routes(PostHandler postHandler) {
    return RouterFunctions.route(GET("/route").and(accept(APPLICATION_JSON)), postHandler::all)
      .andRoute(GET("/route/{id}").and(accept(APPLICATION_JSON)), postHandler::get)
      .andRoute(GET("/route/create").and(accept(APPLICATION_JSON)), postHandler::create);  
    }
}

RouterFunction으로 나온 결과

동일하게 결과가 잘 나오는것을 확인할 수 있다.


그럼 여기서 왜, 그리고 언제 spring reactive를 사용하는게 좋은것일까? 아직까지는 위의 예제를 봐도 크게 어떤 부분 때문에 비동기, 논 블록킹을 스프링에서 사용하는지 언제 사용하는게 효율적인지 잘 모르겠다. 이게 나도 가장 궁금해서 이유를 찾아봤다.

일반적인 상황에서 쓰레드는 요청이 들어오면 끝날때까지 유지된다. 만약 데이터에 접근하고 기록하고 하는 작업이 있다면 이 작업들이 마무리 될 때까지 기다리고 있어야해서 쓰레드 낭비가 커진다.

그래서 응답은 바로 전달하는 non blocking에 비동기로 작업이 진행되도록 하는게 유리하다.
그런데 궁금한게 또 생겼다. 비동기-논블록킹으로 api를 만들면 비동기로 작업이 진행중인데 작업이 종료된 후 어떻게 client에서 결과를 가져올 수 있는건가?

바로 SSE (Server Sent Event) 개념을 이용하여 데이터가 전달된다. 예전에 spring 3.2 부터 추가 되었던 비동기 프로세스에 정리한적이 있었다. https://wedul.site/157

마찬가지로 spring5 reactor에서도 이 개념을 이용하여 동작한다.

우리가 spring webFlux를 사용할 때 내부적으로는 다양한 변화가 발생한다. reactor api에서 제공하는 publisher정보를 우리가 subscribe할 때 publisher는 client에게 각각의 아이템을 serialize하여 대량으로 전달한다.

이런 방식으로 우리는 많은 쓰레드를 생성하지 않고도 대기하고 있는 쓰레드를 이용하여 비동기적으로 데이터를 받을 수 있다. webflux에서 이런 로직을 사용하기 위해서 별도의 작업이 필요하지 않다. 알아서 지원해준다.

위에 정리했었다고 언급했던 spring mvc 3.2부터 추가된 AsyncResult, DefferedResult, Ssemiter등을 사용하면 webflux와 비슷하게 사용하는 것 같지만 사실은 내부적으로 Spring mvc는 스레드를 하나 생성하여 long polling 작업을 위해서 쓰레드를 대기하고 있기 때문에 비동기의 장점을 이용하기에는 어렵다.

실제로 예전 직장에서 long polling으로 client와 세션을 유지시키고 있을 때 대량의 사용자가 붙으면 설정했던 thread 개수를 초과해서 문제가 생긴 경험이 있다.

webFlux에서 이문제가 해결된다니 정말 좋은 것 같다. 왜 사용하는지 조금은 이해가 되는 것 같다.

그럼 말로만 하지말고 실제로 클라이언트에게 값을 전달을 해주는지 테스트해보자. 위에서 했던 소스는 데이터 양도 적고 값이 바로 나오기 때문에 정말 그렇게 나오는지 알 수가 없었다.

그럼 대기시간을 부여해서 확인해보자.


 

쓰레드 반환 후 결과값은 추후에 client에게 전달해주는지 테스트


우선 FouterFunction에 엔드포인트를 하나 더 추가하자.

@Bean
public RouterFunction<?> routes(PostHandler postHandler) {
  return RouterFunctions.route(GET("/route").and(accept(APPLICATION_JSON)), postHandler::all)
    .andRoute(GET("/route/{id}").and(accept(APPLICATION_JSON)), postHandler::get)
    .andRoute(GET("/route/create").and(accept(APPLICATION_JSON)), postHandler::create)
    .andRoute(GET("/delay/client").and(accept(APPLICATION_JSON)), postHandler::clientDelay);
}

그리고 delay기능을 추가하여 ServerResponse를 반환해보자.
만약 정상적인 결과라면 위에 println이 먼저 로그에 찍히고 클라이언트에서 데이터는 3초뒤에 나올 것 이다.

public Mono<ServerResponse> clientDelay(ServerRequest serverRequest) {
  Flux<Post> post = Flux.interval(Duration.ofSeconds(2))
    .take(3)
    .flatMap(number -> postService.findById(number.intValue()));

  System.out.println("test");
  return ServerResponse.ok().body(post, Post.class);
}

예상대로 로그는 먼저 찍힌다.

그리고 브라우저에서 결과는 예상대로 3초뒤에 출력되었다.


이제 정리가 되었다.

결론을 내리면 결과값을 기다릴 필요가 없이 비동기 논블록킹으로 동작하고 쓰레드를 반환하면 더 효율적인 운영이 가능할 것 같다. 그리고 webflux api를 사용할 경우에 걱정할 필요없이 값이 완료되면 클라이언트에게 전달되는 걸 확인 할 수 있었다.

비동기-논블록킹 프레임워크에서 중간에 블록킹이 걸리면 비효율적일 것 같다. 그래서 당장은 jdbc를 쓰는 경우에서는 쓰기 어렵겠지만 NoSql을 사용하는 경우에는 충분히 고려해볼만 할 것 같다.


공부에 사용한 저장소
https://github.com/weduls/spring5


출처 및 도움이 되었던 사이트
https://supawer0728.github.io/2018/03/11/Spring-request-model3/
https://techannotation.wordpress.com/2018/04/24/spring-reactive-a-real-use-case/
https://inyl.github.io/programming/2018/03/10/springboot2_api.html
https://stackabuse.com/spring-reactor-tutorial/
https://supawer0728.github.io/2018/03/15/spring-http-stereamings/

댓글()

Spring reactor Mono와 Flux 정리

web/마이크로서비스|2019. 1. 6. 22:07

지금까지 Spring5에서 추가되었던 리액트 프로그램을 사용하여 간단한 프로그램을 만들어 봤지만 정확하게 Mono와 Flux에 차이와 정의를 정리하지 못한 것 같다. 이번기회에 두 개의 정확한 차이와 사용방법등을 정리해보자.


리액티브 프로그래밍
비동기 블록킹 프로세스로 동작하는 애플리케이션을 논블록킹 프로세스로 동작하기 위해서 지원하는 프로그래밍. (현재 node.js의 동작방식과 유사)


기존 Spring 블록킹 방식
웹에서 서버에 요청이 왔을때 서버는 요청에 대한 적절한 응답을 보내야 하는데 만약 작업이 오래 걸릴 경우에는 요청에 대한 응답이 모두 종료될 때까지 블록킹된다. Spring에서는 그래서 동시 요청 처리를 위해서 멀티 thread를 지원한다. 그러면 하나의 작업이 thread에서 진행되고 다른 thread가 다른 요청을 할당받아서 처리한다. 하지만 이렇게 결국 thread가 늘어나게 되는 경우에는 thread 할당에 필요한 리소스가 늘어나게 되어 비 효율적이 될 수도 있다.


Spring5의 Non blocking
Spring 5가 도입 되면서 클라이언트에 요청에 별도의 thread를 생성하지 않고 buffer를 사용해서 요청을 받고 뒤에서 처리하는 처리하는 thread는 여러개를 두어서 처리한다. 결국 node.js의 싱글스레드 논블로킹을 따라가는 것 같다.

그럼 왜 블로킹 방식을 지원하던 스프링에서 왜 논블로킹 방식을 생각하게 된걸까? 만약에 수천개의 스트림 데이터가 초당 계속 업데이트 되는 시스템이고 적절하게 응답을 해줘야할 때 기존의 블로킹 방식에 경우 상당한 부담을 받게 된다. 그래서 이런 부담을 효율적으로 처리하기 위해서 도입되었다.


Mono와 Flux
Mono는 0-1개의 결과만을 처리하기 위한 Reactor 객체
Flux는 0-N개의 결과물을 처리하기 위한 Reactor 객체

보통 여러 스트림을 하나의 결과를 모아줄 때 Mono를 쓰고 각각의 Mono를 합쳐서 하나의 여러 개의 값을 여러개의 값을 처리할 떄 Flux를 사용한다.

근데 이 부분에서 의문이 있다. 왜 그럼 Flux를 사용하면 되는거지 한개까지만 데이터를 처리할 수 있는 Mono라는 타입이 있는걸까? Mono와 Flux는 같은 Publisher 인터페이스를 구현해서 만들어졌다. 하지만 어떤 시스템에서는 Multi Result가 아닌 하나의 결과셋만 있는 경우가 있다. 그럴경우에는 Mono를 사용한다. 예를 들어 우리가 보통 자바에서 하나의 결과 또는 결과가 없는경우에 List를 사용해서 결과를 받지 않는다. 그와 동일한 개념이라고 생각하면 좋다.

그럼 Mono와 Flux를 사용해서 리액티브 프로그래밍을 하는 방식을정리해보자.

리액티브 스트림

  • 비동기 스트림 처리를 위한 표준으로써 next는 다음신호를 담고 complete는 신호가 끝난것 그리고 error은 신호보내는 도중 에러가 발생한 것을 의미한다.
  • Publisher가 전송하면 데이터는 sequence 대로 전송한다. 그러면 Subscriber가 데이터를 수신한다.
  • next, complete, error 신호를 발생시킨다.


기본적인 설명

1
2
3
4
5
6
// Integer 값을 발생하는 Flux 생성
Flux<Integer> seq = Flux.just(4, 5, 6); 
 
// 구독
seq.subscribe(System.out::println); 
 
cs


Flux.just(1, 2, 3);
--1-2-3-|→ 이처럼 1, 2, 3 세개의 next신호를 발생하고 마지막에 complete 신호를 발생시켜 시퀀스를 끝낸다.

Flux.just();
아무런 sequence가 없는 경우에는 complete 신호만 발생시킨다.

Mono.just(1);
--1-|→ Mono와 Flux의 차이는 Mono는 최대 발생할 수 있는 값이 1개이다.


구독과 신호 발생
sequence는 바로 신호를 발생하지 않는다. 구독을 하는 시점에 신호를 발생하기 시작한다.

1
2
3
4
Flux.just(1, 2, 3)
 .doOnNext(i -> System.out.println("호출: " + i))
 .subscribe(i -> System.out.println("출력 결과: " + i));
 
cs

-> doOnNext메소드는 consumer로부터 구독이 일어났을때 실행된다. 그래서 위에 메시지에 next 신호가 발생했을때 다음과 같은 결과가 발생한다.

호출: 1
출력 결과: 1
호출: 2
출력 결과: 2
호출: 3
출력 결과: 3


Subscriber 인터페이스 메서드 사용방법 정의
subscriber에서 제공하는 메소드는 다음과 같고 구독이 발생하면 onSubscribe가 호출되고 다음 값을 요청하면 onNext 오류가 발생하면 onError 모든 데이터 요청이 끝나면 onComplete가 호출된다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
Flux<Integer> seq = Flux.just(1, 2, 3);
 
seq.subscribe(new Subscriber<>() {
    private Subscription subscription;
    @Override
    public void onSubscribe(Subscription s) {
        // 구독 시작
        this.subscription = s;
        this.subscription.request(1);
    }
 
    @Override
    public void onNext(Integer i) {
        System.out.println("Costomer가 Publisher에게 데이터 요청: " + i);
        this.subscription.request(1);
    }
 
    @Override
    public void onError(Throwable t) {
        System.out.println("Subscriber.onError: " + t.getMessage());
    }
 
    @Override
    public void onComplete() {
        System.out.println("Subscriber.onComplete");
    }
});
cs

-> seq.subscribe 메서드에서 전달한 임의 Subscriber 객체를 onSubscribe 메서드에서 인자로 받아서 이를 필드로 저장하여 사용한다. request(1)은 한개의 데이터를 요청한다는 뜻이다. 만약 모든 데이터를 한번에 받고 싶다면 다음과 같이 지정하면 된다

1
2
3
4
5
6
@Override
public void onSubscribe(Subscription s) {
    System.out.println("Subscriber.onSubscribe");
    this.subscription = s;
    this.subscription.request(Long.MAX_VALUE);
}
cs


콜드 시퀀스와 핫 시퀀스
시퀀스는 구독 시점부터 데이터를 새로 생성하는 Cold sequence와 구독하는 customer와 상관 없이 데이터를 생성하는 hot sequence가 존재한다.

앞 예제 Flux.just()로 생성한 시퀀스가 콜드 시퀀스이다.
콜드 시퀀스는 위에 보면 알겠지만 subscribe가 발생하지 않는다.

Flux<Integer> seq = Flux.just(1, 2, 3);
seq.subscribe(v -> System.out.println(“첫번 째 요청: " + v)); // 구독
seq.subscribe(v -> System.out.println("두번 째 요청: " + v)); // 구독

-> 이 코드를 보면 알겠지만 seq 시퀀스는 구독을 두번한다.이 결과를 seq 시퀀스는 각 구독마다 데이터를 새롭게 생성한다. 마치 API에서 호출하는 것 처럼 매 호출마다 새로운 응답을 만들어 낸다.
첫번 째 요청: 1
첫번 째 요청: 2
첫번 째 요청: 3
두번 째 요청: 1
두번 째 요청: 2
두번 째 요청: 3

핫 시퀀스는 구독여부에 상관없이 데이터가 생성된다. 구독을 하면 구독한 시점이후에 발생하는 데이터부터 신호를 받는다.
-> 예전부터 있었던 데이터를 똑같은 응답을 받는게 아니라 구독을 시작한 부분부터 받는다.


다음 시간에는 스프링으로 직접 만들어보자.



Spring reactor
https://ahea.wordpress.com/2017/02/15/spring-reactive/

단계별 설명
http://wiki.sys4u.co.kr/pages/viewpage.action?pageId=8552586


댓글()