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

댓글()

Spring boot2 resilience4j를 이용한 circuit breaker 사용

web/Spring|2020. 2. 23. 21:37

fault tolerance library (장애 허용 시스템)


fault tolerance library는 무엇인가? 간단하게 이야기해보자. MSA 환경에서 한 개의 서비스에서 다른 api를 호출 할 때 일시적으로 에러가 발생하고 있다고 가정해보자. 만약 이 시기에 요청이 계속 들어오면 계속 500에러를 내보내게 된다. 그럼 사용자들은 이 서비스에 대해서 신뢰를 잃어 버리게 되고 안좋은 인식을 만들 수 있다.

그래서 특정 api 호출과 같은 작업에 에러가 발생했을 때, 그 횟수를 정해놓고 그 횟수 이상 에러를 초과하면 기존에 설정해 놓은 fallback에 맞게 동작하게 하고 일정 시간 후에 다시 시도하여 진행하는 등에 작업이 필요하다. 이게 바로 fault tolerance library (장애 허용 시스템) 이다.

요새 같이 msa로 동작하는 환경이 많아지면서 이에 대한 작업이 많이 필요해졌다.

 

 

circuit breaker


circuit breaker는 fault tolerance library 시스템에서 사용되는 대표적인 패턴으로써 서비스에서 타 서비스 호출 시 에러가 계속 발생하게 되면 circuit를 열어서 메시지가 다른 서비슬 전파되지 못하도록 막고 미리 정의해 놓은 fallback response를 보내어 서비스 장애가 전파되지 않도록 하는 패턴이다.

 

 

resilience4j 


그럼 스프링 부트에서 어떻게 사용하면 될까? 그래서 라이브러리를 알아보다가 Hystrix가 유명하다는 걸 알게 되었다. 이 라이브러리는 netflix에서 만들어서 spring에 기본 라이브러리로 사용되었으나, 넷플릭스에서 더 이상 추가 개발 하지 않고 유지보수만 하겠다고 발표하였으며 resilience4j를 사용하기를 권고했다.

기본적으로 Resilience4j는 Ring Bit Buffer라는 곳에 결과를 저장하게 되는데 성공 여부에 따라 0(실패) 또는 1(성공)로 저장한다. 해당 buffer에 크기는 조정이 가능하다.

그래서 resilience4j를 사용해 보기로 했다.

spring boot2, webflux에서 필요한 라이브러리는 다음과 같다.

dependencies {
    compile group: 'org.springframework.boot', name: 'spring-boot-configuration-processor', version: '2.2.4.RELEASE'
    annotationProcessor group: 'org.springframework.boot', name: 'spring-boot-configuration-processor', version: '2.2.4.RELEASE'
    compileOnly 'org.projectlombok:lombok'
    annotationProcessor 'org.projectlombok:lombok'

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

    //Resilience4J
    compile("io.github.resilience4j:resilience4j-spring-boot2:1.3.0")
    compile("io.github.resilience4j:resilience4j-reactor:1.3.0")
    compile("io.github.resilience4j:resilience4j-timelimiter:1.3.0")

    testImplementation('org.springframework.boot:spring-boot-starter-test') {
        exclude group: 'org.junit.vintage', module: 'junit-vintage-engine'
    }
    testImplementation 'io.projectreactor:reactor-test'
}

 

resilience4j에서 사용할 수 있는 여러 옵션이 있는데 대표적인 옵션은 다음과 같다.

resilience4j:
  circuitbreaker:
    backends:
      wedul:
        ringBufferSizeInClosedState: 30
        ringBufferSizeInHalfOpenState: 30
        waitDurationInOpenState: 5000ms
        failureRateThreshold: 20
        registerHealthIndicator: false
옵션명 설명
ringBufferSizeInClosedState Returns the ring buffer size for the circuit breaker while in closed state.
Circuit이 닫혀있을 때(정상) Ring Buffer 사이즈, 기본값은 100
ringBufferSizedHalfOpenState Returns the ring buffer size for the circuit breaker while in half open state.
half-open 상태일 때 RingBuffer 사이즈 기본값은 10
waitDurationInOpenState Returns the wait duration the CircuitBreaker will stay open, before it switches to half closed
half closed전에 circuitBreaker가 open 되기 전에 기다리는 기간
failureRateThreshold Returns the failure rate threshold for the circuit breaker as percentage.
Circuit 열지 말지 결정하는 실패 threshold 퍼센테이지

 

 

Circuit Open Test


그럼 실제로 실패가 발생하였을 때 circuit이 열리고 fallback이 정상적으로 전달되고 정해진 시간내에 다시 시도하여 정상을 돌아오는지 테스트 해보자.

우선 간단하게 테스트하기 위해서 Ring Buffer 사이즈와 failureRateThreshold 수를 줄여보자.

resilience4j:
  circuitbreaker:
    backends:
      wedul:
        ringBufferSizeInClosedState: 10
        ringBufferSizeInHalfOpenState: 30
        waitDurationInOpenState: 10000ms
        failureRateThreshold: 20
        registerHealthIndicator: false

위에 설정대로라면 실패가 10개의 ringBuffer 20Percent 이상 발생하였을 때  10초동안 fallback 메시지를 보내고 api가 정상적을 돌아오면 정상적으로 돌아오는지 테스트 해보자.

우선 CircuitBreaker를 생성한다. circuitName은 yml에서 설정했던 이름과 동일하게 하면 기본 설정이 Override되어서 지정된다.

@Configuration
public class WedulConfig {

    private static final String CIRCUIT_NAME = "wedul";

    @Bean
    public io.github.resilience4j.circuitbreaker.CircuitBreaker circuitBreaker(CircuitBreakerRegistry registry) {
        return registry.circuitBreaker(CIRCUIT_NAME);
    }

}

 

그다음 webClient에 해당 Circuit Breaker를 사용하도록 지정하고 특정 api를 찌르도록 한다.

return WebClient.builder()
            .clientConnector(new ReactorClientHttpConnector(HttpClient.create()
                .tcpConfiguration(tcpClient ->
                        tcpClient.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, properties.getConnetTimeout())
                    )
            ))
            .uriBuilderFactory(new DefaultUriBuilderFactory(
                UriComponentsBuilder
                    .newInstance()
                    .scheme("http")
                    .host(properties.getUrl())
                    .port(properties.getPort())))
            .build()
            .get()
            .uri(uriBuilder -> uriBuilder.path("/result")
                .queryParam("name", request.getName())
                .queryParam("price", request.getPrice())
                .build()
            )
            .accept(MediaType.APPLICATION_JSON)
            .retrieve()
            .bodyToMono(WedulResponse.class)
            .doOnError(error -> log.error("에러 발생!!!"))
            .transform(CircuitBreakerOperator.of(circuitBreaker))
            .timeout(Duration.ofMillis(properties.getTimeout()));

 

특정 api는 name을 파라미터로 받는데 이름이 wedul이 아니면 RuntimeException을 발생시키도록 하고 wedul이면 isExist를 true로 반환하도록 한다.

// 컨트롤러
@RestController
@RequestMapping("/circuit")
@RequiredArgsConstructor
public class CircuitController {

    private final WedulService wedulService;

    @GetMapping("/test")
    public Mono<ResponseEntity> circuitTest(@Valid WedulRequest wedulRequest) {
        return wedulService.circuitTest(wedulRequest)
            .map(ResponseEntity::ok);
    }

}


// 서비스
@Slf4j
@Service
public class ResultService {

    public Mono<WedulResponse> result(WedulRequest request) {
        if (!request.getName().equals("wedul")) {
            throw new RuntimeException("error");
        }

        return Mono.just(WedulResponse.builder().isExist(true).build());
    }

}

 

그리고 해당 기능을 호출할 간단한 api를 만든다,

// 컨트롤러
@RestController
@RequestMapping("/circuit")
@RequiredArgsConstructor
public class CircuitController {

    private final WedulService wedulService;

    @GetMapping("/test")
    public Mono<ResponseEntity> circuitTest(@Valid WedulRequest wedulRequest) {
        return wedulService.circuitTest(wedulRequest)
            .map(ResponseEntity::ok);
    }

}

// 서비스
@Slf4j
@Service
@RequiredArgsConstructor
public class WedulService {

    private final WedulClient wedulClient;

    public Mono<WedulResponse> circuitTest(WedulRequest request) {
        return wedulClient.isWedulExist(request);
    }

}

 

도식을 간단하게 그리면 다음과 같이 circuit/test 엔드포인트로 들어온 요청을 Circuit Breaker가 설정되어있는 webClient를 사용해서 /result api를 찔러 결과를 리턴 받는다. 이때 에러가 지정한 percent이상 ring buffer에 발생하였을 때 circuit 스위치가 열리는지 확인해보면 된다.

 

먼저 10번 실행 중 3개 에러 발생 시켜서 에러 퍼센트를 만들어주면 다음 요청에 switch가 열리는지 확인해보자. 우선 10개 요청을 실패 7개 성공 3개를 나눠서 실행시켜보자.

curl http://localhost:8080/circuit/test\?price\=22\&name\=wedul
{"exist":false}%                                                                                                                                                                                                                              ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=wedul
{"exist":false}%                                                                                                                                                                                                                              ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=weduls
{"timestamp":"2020-02-24T02:02:20.235+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"500 Internal Server Error from GET http://localhost:8081/result?name=weduls&price=22","requestId":"fdc3768a"}%     ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=wedul
{"exist":false}%                                                                                                                                                                                                                              ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=weduls
{"timestamp":"2020-02-24T02:02:22.995+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"500 Internal Server Error from GET http://localhost:8081/result?name=weduls&price=22","requestId":"284c6869"}%     ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=weduls
{"timestamp":"2020-02-24T02:02:24.105+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"500 Internal Server Error from GET http://localhost:8081/result?name=weduls&price=22","requestId":"b890c0c6"}%     ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=weduls
{"timestamp":"2020-02-24T02:02:24.720+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"500 Internal Server Error from GET http://localhost:8081/result?name=weduls&price=22","requestId":"5bae49c7"}%     ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=weduls
{"timestamp":"2020-02-24T02:02:25.408+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"500 Internal Server Error from GET http://localhost:8081/result?name=weduls&price=22","requestId":"e7121964"}%     ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=weduls
{"timestamp":"2020-02-24T02:02:26.125+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"500 Internal Server Error from GET http://localhost:8081/result?name=weduls&price=22","requestId":"c500e446"}%     ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=weduls
{"timestamp":"2020-02-24T02:02:27.052+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"500 Internal Server Error from GET http://localhost:8081/result?name=weduls&price=22","requestId":"a5f1f79c"}%

아직 까지는 circuit이 열리지 않았다. 그럼 11번째 요청부터는 circuit이 열리는지 확인해보자. 

➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=weduls
{"timestamp":"2020-02-24T02:02:27.987+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"CircuitBreaker 'wedul' is OPEN and does not permit further calls","requestId":"0a553bba"}%                         ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=wedul
{"timestamp":"2020-02-24T02:02:29.524+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"CircuitBreaker 'wedul' is OPEN and does not permit further calls","requestId":"6fb4e23a"}%                         ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=wedul
{"timestamp":"2020-02-24T02:02:30.451+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"CircuitBreaker 'wedul' is OPEN and does not permit further calls","requestId":"36235d1f"}%                         ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=wedul
{"timestamp":"2020-02-24T02:02:31.392+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"CircuitBreaker 'wedul' is OPEN and does not permit further calls","requestId":"41a434df"}%                         ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=wedul
{"timestamp":"2020-02-24T02:02:32.371+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"CircuitBreaker 'wedul' is OPEN and does not permit further calls","requestId":"51661969"}%                         ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=wedul
{"timestamp":"2020-02-24T02:02:33.443+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"CircuitBreaker 'wedul' is OPEN and does not permit further calls","requestId":"55be1c74"}%                         ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=wedul
{"timestamp":"2020-02-24T02:02:34.416+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"CircuitBreaker 'wedul' is OPEN and does not permit further calls","requestId":"34150e67"}%                         ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=wedul
{"timestamp":"2020-02-24T02:02:35.451+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"CircuitBreaker 'wedul' is OPEN and does not permit further calls","requestId":"51a9b6a6"}%                         ➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=wedul
{"timestamp":"2020-02-24T02:02:36.523+0000","path":"/circuit/test","status":500,"error":"Internal Server Error","message":"CircuitBreaker 'wedul' is OPEN and does not permit further calls","requestId":"8ac98f8d"}%

정상적으로 지정해놓은 10초동안 circuit이 열려있는 걸 확인 할 수있다.

그리고 10초 뒤에 다시 실행시켜보면 정상적으로 circuit이 다시 닫혀서 요청을 실행하는걸 볼 수 있다.

➜  ~ curl http://localhost:8080/circuit/test\?price\=22\&name\=wedul
{"exist":false}%

 

 

다른 옵션으로 retry, bulkhead등에 동작이 있는데 상황에 따라 지정해서 사용해보면 될 것 같다. 테스트에 사용했던 소스는 git에 있다.

https://github.com/weduls/circuit_breaker_test

 

weduls/circuit_breaker_test

circuit_break_test. Contribute to weduls/circuit_breaker_test development by creating an account on GitHub.

github.com

참고

https://resilience4j.readme.io/docs/circuitbreaker

https://dlsrb6342.github.io/2019/06/03/Resilience4j란/

 

댓글()

spring cloud resilience4j 사용시 CircuitBreakerConfiguration 에러

web/Spring|2020. 2. 23. 20:32

CircuitBreaker 테스트를 위해서 Resilience4j를 사용했다.

버전은 1.3.0을 사용하려고 했다.

//Resilience4J
compile("io.github.resilience4j:resilience4j-spring-boot2:1.3.0")
compile("io.github.resilience4j:resilience4j-reactor:1.3.0")
compile("io.github.resilience4j:resilience4j-timelimiter:1.3.0")

 

그런데 분명 1.3.0을 사용한다고 명시하였고 gradle도 clean하고 사용하는 denpendency도 확인하였는데 계속해서 다음과 같이 1.1.0 라이브러리를 사용하려고 해서 문제가 발행했다.

Cannot resolve method 'of(java.util.Map<java.lang.String,io.github.resilience4j.circuitbreaker.CircuitBreakerConfig>, io.github.resilience4j.core.registry.RegistryEventConsumer<io.github.resilience4j.circuitbreaker.CircuitBreaker>, io.vavr.collection.HashMap<K,V>)'

 

그래서 알아보던 중 반갑게도 git에 해당 내용에 대한 이슈가 있었다.

이유는 spring cloud를 dependencyManamement로 선언해서 사용해서였다. spring cloud에 경우 내부적으로 1.1.0 라이브러리를 사용하고 있기 때문에 버전 충돌이 나서 그런것이었다. ㅋㅋ

dependencyManagement {
    imports {
        mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}"
    }
}

 

에휴 한시간을 방황한 것 같다.

https://github.com/resilience4j/resilience4j/issues/596#issuecomment-582697615

댓글()

RestHighLevelClient를 사용하여 search after 기능 구현하기

web/Spring|2019. 11. 14. 17:55

https://wedul.site/541에서 search after 기능을 사용해서 검색을 하는 이유를 알아봤었다.

그럼 spring boot에서 RestHighLevelClient를 이용해서 search after를 구현을 해보자.

 

1. Mapping

우선 index가 필요한데 간단하게 상품명과 지역 가격정보들을 가지고 있는 wedul_product 인덱스를 만들어 사용한다.

{
    "settings": {
        "index": {
            "analysis": {
                "tokenizer": {
                    "nori_user_dict": {
                        "type": "nori_tokenizer",
                        "decompound_mode": "mixed",
                        "user_dictionary": "analysis/userdict_ko.txt"
                    }
                },
                "analyzer": {
                    "wedul_analyzer": {
                        "tokenizer": "nori_user_dict",
                        "filter": [
                            "synonym"
                        ]
                    }
                },
                "filter": {
                    "synonym": {
                        "type": "synonym",
                        "synonyms_path": "analysis/synonyms.txt"
                    }
                }
            }
        }
    },
    "mappings": {
        "dynamic": "false",
        "properties": {
            "productId": {
                "type": "keyword"
            },
            "place": {
                "type": "text",
                "fields": {
                    "keyword": {
                        "type": "keyword"
                    }
                }
            },
            "message": {
                "type": "text"
            },
            "query": {
                "type": "percolator"
            },
            "name": {
                "type": "text",
                "analyzer": "wedul_analyzer",
                "fields": {
                    "keyword": {
                        "type": "keyword"
                    }
                }
            },
            "price": {
                "type": "integer"
            },
            "updateAt": {
                "type": "date",
                "format": "epoch_second"
            },
            "createAt": {
                "type": "date",
                "format": "epoch_second"
            }
        }
    }
}

값은 적당하게 3개정도 삽입하였다.

저장되어 있는 초기값.

 

2. 라이브러리 

사용에 필요한 라이브러리들을 gradle을 사용해서 추가한다. 

plugins {
    id 'org.springframework.boot' version '2.2.0.RELEASE'
    id 'io.spring.dependency-management' version '1.0.8.RELEASE'
    id 'java'
}

ext {
    set('elasticsearch.version', '7.4.2')
}

group = 'com.wedul'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = '1.8'

repositories {
    mavenCentral()
    maven { url "https://plugins.gradle.org/m2/" }
}

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-web'
    compileOnly 'org.projectlombok:lombok'
    compile group: 'org.apache.commons', name: 'commons-lang3', version: '3.9'
    compile group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.10.0'
    annotationProcessor 'org.projectlombok:lombok'
    testCompile group: 'org.mockito', name: 'mockito-all', version:'1.9.5'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'

    // gson
    compile group: 'com.google.code.gson', name: 'gson', version: '2.8.6'

    // elasticsearch
    compile 'org.elasticsearch.client:elasticsearch-rest-high-level-client:7.4.2'
    compile group: 'org.elasticsearch', name: 'elasticsearch', version: '7.4.2'
}

 

 

3.RestHighLevelClient configuration

restHighLevelClient 사용을 위한 Configuration 파일을 만들어주는데 id와 pw는 AppConfig라는 별도 properties를 관리하는 bean에서 받아서 사용하는데 base64로 인코딩되어있어서 이를 decoding후 사용한다. (부족한 코드는 글 맨 아래있는 github 링크 참조)

package com.wedul.study.common.config;

import com.wedul.study.common.util.EncodingUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

/**
 * study
 *
 * @author wedul
 * @since 2019-11-07
 **/
@Configuration
@Slf4j
public class ElasticsearchClientConfig implements FactoryBean<RestHighLevelClient>, InitializingBean, DisposableBean {

    @Autowired
    AppConfig appConfig;

    private RestHighLevelClient restHighLevelClient;

    @Override
    public RestHighLevelClient getObject() {
        return restHighLevelClient;
    }

    @Override
    public Class<?> getObjectType() {
        return RestHighLevelClient.class;
    }

    @Override
    public void destroy() {
        try {
            if (null != restHighLevelClient) {
                restHighLevelClient.close();
            }
        } catch (Exception e) {
            log.error("Error closing ElasticSearch client: ", e);
        }
    }

    @Override
    public boolean isSingleton() {
        return false;
    }

    @Override
    public void afterPropertiesSet() {
        restHighLevelClient = buildClient();
    }

    private RestHighLevelClient buildClient() {
        try {
            String id = EncodingUtil.decodingBase64(appConfig.getElasticsearchConfig().getId());
            String pw = EncodingUtil.decodingBase64(appConfig.getElasticsearchConfig().getPw());

            // 계정 설정
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials(id, pw));

            // client 설정
            RestClientBuilder builder = RestClient.builder(
                new HttpHost(appConfig.getElasticsearchConfig().getIp(),
                    appConfig.getElasticsearchConfig().getPort(), "http"))
                .setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));

            restHighLevelClient = new RestHighLevelClient(builder);
        } catch (Exception e) {
            log.error(e.getMessage());
        }
        return restHighLevelClient;
    }

}

 

 

4. Handler 추가

자주 사용되는 Elasticsearch 문법을 처리하기 위해서 만들어 놓은 ElasticsearchHandler에 search after에 사용 될 메소드를 추가한다. search after는 sort 필드가 없으면 사용이 불가능 하기 때문에 sort 필드가 없는 경우 에러를 전달한다.

public static SearchSourceBuilder searchAfter(Map<String, SortOrder> sortFields, QueryBuilder query, Object[] searchAfter, int size) {
    return searchAfterBuilder(sortFields, query, searchAfter,  size);
}

public static SearchSourceBuilder searchAfter(Map<String, SortOrder> sortFields, QueryBuilder query, Object[] searchAfter) {
    return searchAfterBuilder(sortFields, query, searchAfter, 20);
}

private static SearchSourceBuilder searchAfterBuilder(Map<String, SortOrder> sortFields, QueryBuilder query, Object[] searchAfter, int size) {
    SearchSourceBuilder builder = new SearchSourceBuilder();

    if (CollectionUtils.isEmpty(sortFields)) {
        throw new InternalServerException("잘못된 필드 요청입니다.");
    }

    sortFields.forEach((field, sort) -> {
        builder.sort(field, sort);
    });
    builder.size(size);
    builder.query(query);

    if (ArrayUtils.isNotEmpty(searchAfter)) {
        builder.searchAfter(searchAfter);
    }

    return builder;
}

 

 

5. 기능 구현

위의 기능들을 이용해서 실제로 구현해보자. productService와 productRepository 클래스를 통해서 구현하였다. 자세한 설명없이 간단하기 때문에 소스를 보면 알 수 있다. 

 

우선 최종 결과물로 사용될 클래스는 ElasticResult인데 다음과 같이 현재 요청이 마지막인지 표시하는 isLast와 다음 요청을 위해 보내줘야 하는 cursor값과 결과값 전체 total과 결과 리스트 list 필드가 존재한다.

@Builder
@Data
public class ElasticResult<T extends ElasticsearchDto> {

    private boolean isLast;
    private long total;
    private List<T> list;
    private Object[] cursor;

}

 

그 다음 service로직을 통해 결과를 얻어서 위 ElasticResult에 결과를 담아보자. products 메서드는 요청을 받아서 elasticsearch에 실제 조작요청을 하는 productRepository에 동작을 요청하고 값을 받아서 처리하는 메서드이다. 그리고 extractProductList는 결과값에서 ProductDto 값을 뽑아내는 메서드이다.

public ElasticResult<ProductDto> products(String index, Object[] searchAfter, int size) throws IOException {
    SearchResponse searchResponse = productRepository.products(index, searchAfter, size);
    SearchHits searchHits = searchResponse.getHits();
    int hitCnt = searchHits.getHits().length;
    boolean isLast = 0 == hitCnt || size > hitCnt;

    return ElasticResult.<ProductDto>builder()
        .cursor(isLast ? null : searchHits.getHits()[hitCnt - 1].getSortValues())
        .isLast(isLast)
        .list(extractProductList(searchHits))
        .total(searchHits.getTotalHits().value)
        .build();
}

private List<ProductDto> extractProductList(SearchHits searchHits) {
    List<ProductDto> productList = new ArrayList<>();

    searchHits.forEach(hit -> {
        Map<String, Object> result = hit.getSourceAsMap();

        productList.add(ProductDto.builder()
            .name(String.valueOf(result.get("name")))
            .productId(String.valueOf(result.get("productId")))
            .place(String.valueOf(result.get("place")))
            .price(Integer.valueOf(result.get("price").toString()))
            .updateAt(Long.valueOf(result.get("updateAt").toString()))
            .createAt(Long.valueOf(result.get("createAt").toString())).build());
    });

    return productList;
}

 

그리고 마지막으로 es에 직접적으로 콜을 하는 productRepository 이다. 여기서 정렬 키워드는 name과 place를 사용한다.

public SearchResponse products(String index, Object[] searchAfter, int size) throws IOException {
    SearchRequest searchRequest = new SearchRequest(index);
    Map<String, SortOrder> sorts = new HashMap<String, SortOrder>() {
        {
            put("name.keyword", SortOrder.DESC);
            put("place.keyword", SortOrder.DESC);
        }
    };

    SearchSourceBuilder searchSourceBuilder = ElasticsearchHandler.searchAfter(sorts, QueryBuilders.matchAllQuery(), searchAfter, size);
    searchRequest.source(searchSourceBuilder);
    return restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
}

 

 

6. 테스트

그럼 위에 내용이 잘 구현되었는지 테스트를 해보자. 총 3개의 데이터가 있는데 이 데이터를 1개씩 search after를 통해서 값을 받아서 저장하고 한번에 출력하도록 해보자.

@Test
@DisplayName("search after")
public void searchAfter() throws IOException {
    ElasticResult<ProductDto> result = productService.products(PRODUCT_INDEX, new Object[]{}, 1);
    List<ProductDto> productDtos = new ArrayList<>();

    while(result != null && !result.isLast()) {
        productDtos.addAll(result.getList());
        result = productService.products(PRODUCT_INDEX, result.getCursor(), 1);
    }
    productDtos.addAll(result.getList());

    productDtos.forEach(productDto -> {
        System.out.println("이름 : " + productDto.getName());
        System.out.println("장소 : " + productDto.getPlace());
    });
}

결과는 정상적으로 3가지 모두 잘 출력되는 걸 알 수있다.

 

우선 기능 구현을 해보기 위해서 진행하였는데 더 다듬어야 할 것같다.

자세한 소스는 github참조

댓글()

JPA 다양한 Join 방법 정리 (N+1, queryDSL, fetch join)

web/Spring|2019. 11. 4. 20:31

JPA를 사용하다 보면 join을 할 때가 많아진다. join을 어떠한 방법으로 하느냐에 따라서 수행되는 쿼리가 달라지고 성능에 문제가 발생하는 경우도 종종있다.

 

그래서 다양한 방식의 join 방식을 알아보고 방식에 따라 작업을 진행해 보자.

우선 사용될 entity 두 개를 설명하면 다음과 같다.

@Getter
@Entity
@Table(name = "wedul_classes")
@AllArgsConstructor(access = AccessLevel.PROTECTED)
@NoArgsConstructor
@Builder
public class WedulClasses extends CommonEntity {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long wedulClassesId;

    @OneToMany(mappedBy = "wedulClasses", fetch = FetchType.LAZY)
    private Set<WedulStudent> wedulStudentList = new LinkedHashSet<>();

    private String classesName;

    private String classesAddr;

}

@Getter
@Entity
@Table(name = "wedul_student")
@AllArgsConstructor(access = AccessLevel.PROTECTED)
@NoArgsConstructor
@Builder
public class WedulStudent extends CommonEntity {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long wedulStudentId;

    @ManyToOne(fetch = FetchType.LAZY)
    @JoinColumn(name = "wedul_classes_id")
    @JsonBackReference
    private WedulClasses wedulClasses;

    private String studentName;

    private int studentAge;

    @Enumerated(value = EnumType.STRING)
    private StudentType studentType;

}

이 두 엔티티는 class와 student의 관계로 1대 N의 관계를 가지고 있다.

이 과정에서 사용될 데이터는 임의로 추가했고 다음과 같다.

 

wedul_classe 테이블의 데이터

wedul_student 테이블의 데이터

사용 쿼리 툴) tadpole docker version

 

 

그럼 이 entity를 이용해서 데이터를 조인하여 조회해보자.

 

1. 단순 조회

우선 첫 번째로 JpaRepository 인터페이스 사용 시 기본적으로 제공하는 findAll을 사용해보자.

@Repository
public interface WedulClassesRepository extends JpaRepository<WedulClasses, Long> {

    Optional<WedulClasses> findByClassesName(String classesName);

}

이를 사용하여 데이터를 조회해보면 사용 되는 쿼리는 다음과 같다.

-- classes 목록을 조회하는 쿼리
select
    wedulclass0_.wedul_classes_id as wedul_cl1_0_
    ,wedulclass0_.create_at as create_a2_0_
    ,wedulclass0_.update_at as update_a3_0_
    ,wedulclass0_.classes_addr as classes_4_0_
    ,wedulclass0_.classes_name as classes_5_0_
  from
    wedul_classes wedulclass0_
;


-- 아래 쿼리들은 wedul_classes_id 개수별로 조회되는 쿼리
select
    wedulstude0_.wedul_classes_id as wedul_cl7_1_0_
    ,wedulstude0_.wedul_student_id as wedul_st1_1_0_
    ,wedulstude0_.wedul_student_id as wedul_st1_1_1_
    ,wedulstude0_.create_at as create_a2_1_1_
    ,wedulstude0_.update_at as update_a3_1_1_
    ,wedulstude0_.student_age as student_4_1_1_
    ,wedulstude0_.student_name as student_5_1_1_
    ,wedulstude0_.student_type as student_6_1_1_
    ,wedulstude0_.wedul_classes_id as wedul_cl7_1_1_
  from
    wedul_student wedulstude0_
  where
    wedulstude0_.wedul_classes_id = ?
;
select
    wedulstude0_.wedul_classes_id as wedul_cl7_1_0_
    ,wedulstude0_.wedul_student_id as wedul_st1_1_0_
    ,wedulstude0_.wedul_student_id as wedul_st1_1_1_
    ,wedulstude0_.create_at as create_a2_1_1_
    ,wedulstude0_.update_at as update_a3_1_1_
    ,wedulstude0_.student_age as student_4_1_1_
    ,wedulstude0_.student_name as student_5_1_1_
    ,wedulstude0_.student_type as student_6_1_1_
    ,wedulstude0_.wedul_classes_id as wedul_cl7_1_1_
  from
    wedul_student wedulstude0_
  where
    wedulstude0_.wedul_classes_id = ?
;
select
    wedulstude0_.wedul_classes_id as wedul_cl7_1_0_
    ,wedulstude0_.wedul_student_id as wedul_st1_1_0_
    ,wedulstude0_.wedul_student_id as wedul_st1_1_1_
    ,wedulstude0_.create_at as create_a2_1_1_
    ,wedulstude0_.update_at as update_a3_1_1_
    ,wedulstude0_.student_age as student_4_1_1_
    ,wedulstude0_.student_name as student_5_1_1_
    ,wedulstude0_.student_type as student_6_1_1_
    ,wedulstude0_.wedul_classes_id as wedul_cl7_1_1_
  from
    wedul_student wedulstude0_
  where
    wedulstude0_.wedul_classes_id = ?
;

쿼리를 자세히 보면 알겠지만 wedul_classes를 조회하는 쿼리와 그 wedul_classes 개수만큼 쿼리가 실행되는것을 볼 수 있다.

많이 들어 봤을 법한 N+1 문제가 발생한 것이다.

이 방식으로 쿼리 수행 시 N번의 쿼리가 발생해야 하기에 데이터 수만큼 쿼리가 실행되는 안좋은 부담을 안고 가야해서 좋지 않다.

 

2. left fetch join

위의 1번의 N+1 문제 해결로 고안된 방법 중 하나가 fetch join이다. 나는 left join을 하고자 하기에 left fetch join을 시도해보자. 우선 사용된 코드는 다음과 같다.

@Repository
public interface WedulClassesRepository extends JpaRepository<WedulClasses, Long> {
    @Query(value = "select DISTINCT c from WedulClasses c left join fetch c.wedulStudentList")
    List<WedulClasses> findAllWithStudent();
}

distinct가 붙은 이유는 카티션곱에 의해서 여러개의 결과값이 발생해 버리기 때문에 추가하였다.

그럼 사용된 쿼리도 확인해보자.

select
    distinct wedulclass0_.wedul_classes_id as wedul_cl1_0_0_
    ,wedulstude1_.wedul_student_id as wedul_st1_1_1_
    ,wedulclass0_.create_at as create_a2_0_0_
    ,wedulclass0_.update_at as update_a3_0_0_
    ,wedulclass0_.classes_addr as classes_4_0_0_
    ,wedulclass0_.classes_name as classes_5_0_0_
    ,wedulstude1_.create_at as create_a2_1_1_
    ,wedulstude1_.update_at as update_a3_1_1_
    ,wedulstude1_.student_age as student_4_1_1_
    ,wedulstude1_.student_name as student_5_1_1_
    ,wedulstude1_.student_type as student_6_1_1_
    ,wedulstude1_.wedul_classes_id as wedul_cl7_1_1_
    ,wedulstude1_.wedul_classes_id as wedul_cl7_1_0__
    ,wedulstude1_.wedul_student_id as wedul_st1_1_0__
  from
    wedul_classes wedulclass0_
      left outer join wedul_student wedulstude1_
        on wedulclass0_.wedul_classes_id = wedulstude1_.wedul_classes_id

left join을 해서 한번에 데이터를 가져올 수 있는 걸 확인 할 수 있지만 아쉽게도 Lazy로 데이터를 가져오지 못하고 Eager로 가져와야 한다.

 

3. EntityGraph

이제 3번째 방식으로 entity graph를 사용하여 실행시켜보자. 코드는 아래와 같다.

@EntityGraph(attributePaths = "wedulStudentList")
@Query("select c from WedulClasses c")
Page<WedulClasses> findEntityGraph(Pageable pageable);

실행되는 쿼리는 다음과 같아서 2번과 동일하다. (page를 사용한 것만 차이)

select
    wedulclass0_.wedul_classes_id as wedul_cl1_0_0_
    ,wedulstude1_.wedul_student_id as wedul_st1_1_1_
    ,wedulclass0_.create_at as create_a2_0_0_
    ,wedulclass0_.update_at as update_a3_0_0_
    ,wedulclass0_.classes_addr as classes_4_0_0_
    ,wedulclass0_.classes_name as classes_5_0_0_
    ,wedulstude1_.create_at as create_a2_1_1_
    ,wedulstude1_.update_at as update_a3_1_1_
    ,wedulstude1_.student_age as student_4_1_1_
    ,wedulstude1_.student_name as student_5_1_1_
    ,wedulstude1_.student_type as student_6_1_1_
    ,wedulstude1_.wedul_classes_id as wedul_cl7_1_1_
    ,wedulstude1_.wedul_classes_id as wedul_cl7_1_0__
    ,wedulstude1_.wedul_student_id as wedul_st1_1_0__
  from
    wedul_classes wedulclass0_
      left outer join wedul_student wedulstude1_
        on wedulclass0_.wedul_classes_id = wedulstude1_.wedul_classes_id
  order by
    wedulclass0_.update_at desc;

 

4. QueryDSL

Querydsl은 정적 타입을 이용해서 SQL과 같은 쿼리를 사용할 수 있도록 해주는 프레임워크로 HQL쿼리를 실행하게 도와준다.

설정 방식은 gradle 5 기준으로 다음과 같다.

plugins {
    id 'org.springframework.boot' version '2.2.0.RELEASE'
    id 'io.spring.dependency-management' version '1.0.8.RELEASE'
    id "com.ewerk.gradle.plugins.querydsl" version "1.0.10"
    id 'java'
}

group = 'com.wedul'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = '1.8'

configurations {
    compileOnly {
        extendsFrom annotationProcessor
    }
}

repositories {
    mavenCentral()
    maven { url "https://plugins.gradle.org/m2/" }
}

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
    implementation 'org.springframework.boot:spring-boot-starter-web'
    compile group: "org.flywaydb", name: "flyway-core", version: '5.2.4'
    compileOnly 'org.projectlombok:lombok'
    runtimeOnly 'mysql:mysql-connector-java'
    compile group: 'org.apache.commons', name: 'commons-lang3', version: '3.9'
    compile group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.10.0'
    annotationProcessor 'org.projectlombok:lombok'
    testCompile group: 'org.mockito', name: 'mockito-all', version:'1.9.5'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'

    // query dsl
    compile("com.querydsl:querydsl-apt")
    compile("com.querydsl:querydsl-jpa")
}

// querydsl 적용
def querydslSrcDir = 'src/main/generated'

querydsl {
    library = "com.querydsl:querydsl-apt"
    jpa = true
    querydslSourcesDir = querydslSrcDir
}

compileQuerydsl{
    options.annotationProcessorPath = configurations.querydsl
}

configurations {
    querydsl.extendsFrom compileClasspath
}

sourceSets {
    main {
        java {
            srcDirs = ['src/main/java', querydslSrcDir]
        }
    }
}

그리고 QueryDsl 사용을 위해 QueryDslRepositorySupport를 상속받아서 사용할 수 있는데 마지막에 distinct를 사용한 것은 2번 fetch 조인의 이유와 동일하다.

@Repository
public class WedulClassesQueryDsl extends QuerydslRepositorySupport {

    public WedulClassesQueryDsl() {
        super(WedulClasses.class);
    }

    public List<WedulClasses> findAllWithStudent() {
        QWedulClasses wedulClasses = QWedulClasses.wedulClasses;
        QWedulStudent wedulStudent = QWedulStudent.wedulStudent;

        return from(wedulClasses)
            .leftJoin(wedulClasses.wedulStudentList, wedulStudent)
            .fetchJoin()
            .distinct()
            .fetch();
    }

}

그럼 마찬가지로 실행되는 쿼리를 확인해보자.

select
    distinct wedulclass0_.wedul_classes_id as wedul_cl1_0_0_
    ,wedulstude1_.wedul_student_id as wedul_st1_1_1_
    ,wedulclass0_.create_at as create_a2_0_0_
    ,wedulclass0_.update_at as update_a3_0_0_
    ,wedulclass0_.classes_addr as classes_4_0_0_
    ,wedulclass0_.classes_name as classes_5_0_0_
    ,wedulstude1_.create_at as create_a2_1_1_
    ,wedulstude1_.update_at as update_a3_1_1_
    ,wedulstude1_.student_age as student_4_1_1_
    ,wedulstude1_.student_name as student_5_1_1_
    ,wedulstude1_.student_type as student_6_1_1_
    ,wedulstude1_.wedul_classes_id as wedul_cl7_1_1_
    ,wedulstude1_.wedul_classes_id as wedul_cl7_1_0__
    ,wedulstude1_.wedul_student_id as wedul_st1_1_0__
  from
    wedul_classes wedulclass0_
      left outer join wedul_student wedulstude1_
        on wedulclass0_.wedul_classes_id = wedulstude1_.wedul_classes_id

애도 2번, 3번과 동일한 쿼리가 작성되는 걸 확인할 수 있다.

기본적으로 단순하게 다대일 데이터를 가져오려고 하면 N+1 문제가 발생할 수 있기 때문에 조심해야하고 이를 해결하기 위해서는 다양한 방식의 문제 해결 방식이 있는걸 확인할 수 있었다.

무엇이 가장 좋은지는 본인이 판단하거나 상황에 맞게 사용하면 좋을 거 같다.

댓글()

데이터 베이스 버전 컨트롤 Flyway

web/Spring|2019. 9. 28. 13:43

Spring에서 초기 테이블과 데이터 관리를 위해서 data.sql과 schema.sql을 사용하였다. 하지만 테이블 스키마가 변경되거나 필수로 초기에 들어가야하는 데이터들이 추가되거나 수정되었을 때 히스토리 관리가 잘 되지 않았다. 

특히 서로 교류가 잘 되지 않은 경우에서는 컬럼이 추가되거나 무엇이 변경되었는지 알지 못해서 문제를 유발할 수 있기에 이를 관리 할 수 있는 무언가가 필요했다.

그래서 Redgate에서 제공하는 Flyway를 사용해보기로 했다. 우선 내 개인 프로젝트인 timeline에 적용시켜봤다.

 

데이터베이스 버전관리 Flyway

https://flywaydb.org/

동작 방식

Flyway가 버전관리를 하기위해서 테이블이 생성된다. Flyway가 버전관리는 이 테이블에 데이터베이스의 상태를 기록하면서 진행한다. 

Flyway가 시작되면 파일시스템 또는 마이그레이션 대상의 classpath를 스캔해서 Sql 또는 Java로 쓰여진 파일을 찾는다. 이 마이그레이션 작업은 파일에 적혀있는 version number대로 순서대로 진행된다. 그리고 현재 마이그레이션 해야할 파일의 버전과 테이블에 기록된 버전을 확인해보고 같으면 넘어간다.

Flyway에서 사용하는 테이블은 flyway_schema_history로 아래와 같이 구성되어있다.

CREATE TABLE `flyway_schema_history` (
  `installed_rank` int(11) NOT NULL,
  `version` varchar(50) DEFAULT NULL,
  `description` varchar(200) NOT NULL,
  `type` varchar(20) NOT NULL,
  `script` varchar(1000) NOT NULL,
  `checksum` int(11) DEFAULT NULL,
  `installed_by` varchar(100) NOT NULL,
  `installed_on` timestamp NOT NULL DEFAULT current_timestamp(),
  `execution_time` int(11) NOT NULL,
  `success` tinyint(1) NOT NULL,
  PRIMARY KEY (`installed_rank`),
  KEY `flyway_schema_history_s_idx` (`success`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
installed_rank 인덱스
version 버전명 (V나 R 뒤에 붙는 숫자)
description 설명
type SQL 또는 JDBC 
script 스크립트 이름 V1__kdjlkdf.sql
checksum checksum
installed_by 실행 주최자
installed_on 설치된 시간
execution_time 총 실행시간
success 성공여부

 

간단히 말해 변경된 데이터나 테이블 스키마를 적용하기 위해서는 마지막 버전보다 높은 파일을 만들어서 애플리케이션을 구동하면 된다.

 

애플리케이션에 적용

그럼 flyway를 적용하기 위해 gradle에 라이브러리부터 추가해보자.

dependency {
	compile group: "org.flywaydb", name: "flyway-core", version: '5.2.4'
}

그리고 application.yml을 설정하자.

spring:
  flyway:
    enabled: true
    baselineOnMigrate: true
    encoding: UTF-8

그리고 테이블과 데이터를 넣을 sql을 만들자. 

그리고 Springboot 애플리케이션을 실행시키면 해당 테이블에 버전 히스토리가 기록된다.

 

버전관리하기에 좋은거 같다.

댓글()

Redis에서 Pub/Sub 방식 소개 및 Spring Boot에서 구현해보기

web/Spring|2019. 8. 21. 23:07

redis에 추가된 SUBSCRIBE, UNSUBSCRIBE 그리고 PUBLISH는 Publish/Subscribe 메시지 패러다임을 구현한 기능이다. sender(publisher)들은 특별한 receiver(subscriber)에게 값을 전달하는게 아니라 해당 채널에 메시지를 전달하면 그 메시지를 구독하고 있는 subscribe에게 메시지를 전송한다. subscribers는 하나 또는 그 이상의 채널에 구독을 요청하고 publisher가 누구인지 상관 없이 해당 채널에 들어온 모든 메시지를 읽게된다.

이 subscriber와 publisher의 decoupling은 확장성있는 성장을 가져올 수 있다.

 

Redis-Cli로 기능 사용하기


subscriber
redis-cli를 열고 SUBSCRIBE 채널1 채널2 ... 를 입력한다.

 

publisher
마찬가지로 redis-cli를 열고 PUBLISH 채널 메시지 를 입력해서 전송한다.

그럼 이를 구독하고 있던 subscriber 콘솔에 다음과 같이 출력된다.

 

Spring Boot 2.1.7에 적용하기


그럼 이 방식을 Spring boot에 적용하여 sub와 pub를 이용한 개발을 해보자.

우선 필요한 libaray는 다음과 같다.

spring-boot-starter-data-redis
spring-boot-starter-web
lettuce-core (기본적으로 탑재된 jedis보다 좋다고 하여 변경)
lombok
spring-boot-starter-test

 

라이브러리를 maven이나 gradle 통해 넣어주고 configuration을 통해서 지정해보다. 기본적으로 redisTemplate의 connection은 application.properties에 spring.redis.host, spring.redis.port에 지정해주면 그에 맞게 생성되기 때문에 별도로 설정해주지 않고 그대로 사용한다.

그리고 RedisSubscriber Listener를 구현해서 적용해주는데 RedisMessageListenerContainer를 설정해준다. 속성 값으로 MessageListenerAdapter를 부여해주는데 이 Adapter에는 MessageListener인터페이스를 구현하고 onMessage를 재정의하여 전달 받은 메시지에 대한 처리를 지정한다.

Configuration

    private RedisTemplate<String, String> redisTemplate;

    @Bean
    MessageListenerAdapter messageListener() {
        return new MessageListenerAdapter(new RedisMessageSubscriber());
    }

    @Bean
    RedisMessageListenerContainer redisContainer() {
        final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisTemplate.getConnectionFactory());
        container.addMessageListener(messageListener(), topic());
        return container;
    }

RedisMessageSubScriber

package com.study.redis.config;

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

/**
 * spring-boot-study
 *
 * @author wedul
 * @since 2019-08-21
 **/
@Service
public class RedisMessageSubscriber implements MessageListener {

    public static List<String> messageList = new ArrayList<>();

    @Override
    public void onMessage(final Message message, final byte[] pattern) {
        messageList.add(message.toString());
        System.out.println("Message received: " + new String(message.getBody()));

    }
}

그럼 기동해보고 redis-cli를 통해서 PUBLISH를 날려보면 위에 onMessage에 정의한 대로 콘솔로그가 찍히는지 보자.

그리고 Publisher도 설정하고 Test 코드를 작성하여 redis-cli처럼 결과가 나오는지 확인해보자.

우선 Publisher에서 사용되는 RedisMessagePublisher를 정의해준다.

Configuration

    @Bean
    RedisMessagePublisher redisPublisher() {
        return new RedisMessagePublisher(redisTemplate, topic());
    }

    @Bean
    ChannelTopic topic() {
        return new ChannelTopic("wedul");
    }

Test

package com.study.redis;

import lombok.NoArgsConstructor;
import lombok.RequiredArgsConstructor;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
@NoArgsConstructor
public class RedisApplicationTests {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Test
    public void contextLoads() {
        redisTemplate.convertAndSend("wedul", "No I'm genius");
    }

}

결과도 잘 나왔다. 굿굿 

ElasticCache를 사용하고 있다면 별도의 카프카와 같은 메시지큐 없이도 레디스를 사용해도 되지 않나 싶기도 하다.

 

자세한 코드는 여기에 redis 모듈 참고

https://github.com/weduls/spring5

 

weduls/spring5

study. Contribute to weduls/spring5 development by creating an account on GitHub.

github.com

 

댓글()
  1. Favicon of https://coding-start.tistory.com BlogIcon 여성게 2019.08.23 14:36 신고 댓글주소  수정/삭제  댓글쓰기

    저는 메시지큐 쓰려고 무거운 카프카를 사용했었고 다른 용도로 레디스도 사용했는데, 복잡한 메시지큐 기능이 필요하지 않으면 레디스 펍/섭 기능도 가볍게 쓰기 좋겠내요 ㅎㅎ

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/

댓글()