🔥 이전까지 해왔던 것들을 되돌아보자
이전까지는 기존의 Spring MVC 코드에서 Webflux로 갈아타기 위해서 Cache hit 방식의 비동기 기반 repository를 작성해보았습니다.
이제부터는 함수형 엔드포인트를 적용하기 위해 제가 어떤 코드를 작성했는지 설명드리겠습니다.
🔥 본격적인 글 작성 이전에 배경지식부터 깔고 시작하겠습니다
우선 Webflux로만 작성된 코드 예시를 하나 보여드리겠습니다.
@GetMapping("/search-shops")
fun searchShopNumbers(@RequestParam request: SearchRequest): Mono<List<Long>> {
// 휴무일여부 + 배달가능센터 동시 조회
return Mono.zip(
// 해당 메서드의 반환값은 Mono
holidayService.isHoliday(request.currentDate),
// 해당 메서드의 반환값은 Mono
deliverableCenterService.findDeliverableCenter(request.location)
)
// zip의 결과로 tuple반환. 이를 풀어서 request로 전환
.map { tuple -> toServiceRequest(tuple.t1, tuple.t2, request) }
// findShopNumbers(it)은 Mono로 반환.. flatmap으로 비동기 처리 필요
.flatMap { findShopNumbers(it) }
// 가게번호 반환!
.map {shops -> shops.map { it.shopNumber } }
}
private fun findShopNumbers(request: ServiceSearchRequest): Mono<List<Shop>> {
// ...
}(해당 코드는 여기서 참고했습니다 -> 배민광고리스팅 개발기 )
위의 코드는 순수하게 webflux로만 코드를 작성한 케이스입니다. 이 방식은 그렇게 좋지 못하다고 저는 판단하였는데요, 이유는 다음과 같습니다.
1. 비동기 로직들이 모두 callback 형식으로 주렁주렁 매달려있기 때문에 로직의 분리가 어렵다.
2. 가독성이 매우 떨어진다.
위의 문제 말고도 여럿 문제들이 분명히 많겠지만, 저는 이를 Coroutines로 한번 해결해보고자 마음 먹게 되었습니다.
우선 제가 생각한 Functional Endpoint의 구조는 다음과 같습니다.
1. Router layer: nest를 이용해서 간단하게 endpoint를 정의한다
2. Handler layer: service layer에 의존해서 router에서 정의한 endpoint를 실행하는 최종 로직을 작성한다
3. Service layer: 실제 비지니스 로직을 작성한다
4. Repository layer: Data access 로직을 작성한다. 실제 비지니스 로직과 DB 사이의 물리적 연결을 추상화시키는 역할을 수행한다
🚨 이번 예시에서는 Service layer 없이 Handler에다가 바로 repository layer를 의존하는 방식으로 작성하였는데요, 이런 경우에는 단위 테스트를 할때 문제가 발생하기 때문에 이걸 읽으시는 분은 이 점 주의해주시길 바랍니다!
이전 포스팅까지는 repository 까지 작성해보았기 때문에 이번 시간에는 handler layer부터 작성을 시작하겠습니다!
🔥 Handler 구현을 보여드리겠습니다.
이 로직은 위드마켓에서 shop에 대한 review를 조회하는데 사용하는 handler logic입니다.
우선 코드부터 보여드리겠습니다.
👉 전체 코드
🔨 ShopReviewHandler.kt
@Component
class ShopReviewHandler(
private val shopReviewRepository: ShopReviewRepository,
private val resultFactory: ResultFactory
) {
/** reviewId와 reviewTitle을 기반으로 review 하나를 가져오는 메소드
* @param reviewId review의 id
* @param reviewTitle review의 제목
*/
suspend fun findReviewByIdAndTitle(request: ServerRequest): ServerResponse = coroutineScope {
// id와 title을 request로부터 받아오고, 존재하지 않으면 바로 에러 처리를 수행한다
val reviewId = request.queryParamOrNull("id") ?: throw RequestParamLostException("reviewId is lost!!")
val reviewTitle = request.queryParamOrNull("title") ?: throw RequestParamLostException("reviewTitle is lost!!")
val reviewMono = shopReviewRepository.findShopReviewByIdAndTitleWithCaching(reviewId, reviewTitle)
val reviewDeferred = CoroutinesUtils.monoToDeferred(reviewMono)
return@coroutineScope withContext(Dispatchers.IO) {
reviewDeferred.await()
}?.let {
ok().contentType(MediaType.APPLICATION_JSON)
.bodyValueAndAwait(resultFactory.getSingleResult(toSimpleReadDto(it)))
} ?: throw ShopReviewNotFoundException("Shop review is not found!!")
}
suspend fun getReviewListByShopIdAndName(request: ServerRequest): ServerResponse = coroutineScope {
// query param으로부터 shopId와 shopName을 받아오고, 없으면 예외처리
val shopId = request.queryParamOrNull("shop-id") ?: throw RequestParamLostException("shopId is lost!!")
val shopName = request.queryParamOrNull("shop-name") ?: throw RequestParamLostException("shopName is lost!!")
val reviewDtoList = mutableListOf<ShopReviewSimpleReadDto>()
val reviewFlow = shopReviewRepository.getShopReviewListFlowByShopIdAndNameWithCaching(shopId, shopName)
// 비동기적으로 reviewDtoList에 원소를 담는다
withContext(Dispatchers.IO) {
reviewFlow.buffer()
.collect {
val reviewDeferred = CoroutinesUtils.monoToDeferred(it)
val review = reviewDeferred.await()!!
reviewDtoList.add(toSimpleReadDto(review))
}
}
// review가 하나도 없다면 예외 처리
check(reviewDtoList.size != 0) {
throw ShopReviewNotFoundException("shopReview is not found!!")
}
return@coroutineScope ok().contentType(MediaType.APPLICATION_JSON)
.bodyValueAndAwait(resultFactory.getMultipleResult(reviewDtoList))
}
private fun toSimpleReadDto(shopReview: ShopReview) = ShopReviewSimpleReadDto(
reviewId = shopReview.reviewId,
reviewTitle = shopReview.reviewTitle,
shopId = shopReview.shopId,
shopName = shopReview.shopName,
reviewContent = shopReview.reviewContent,
reviewScore = shopReview.reviewScore,
reviewPhotoList = shopReview.reviewPhotoList,
createdAt = shopReview.createdAt,
updatedAt = shopReview.updatedAt
)
}저는 이 중에서 getReviewListByShopIdAndName(request: ServerRequest)만을 리뷰하도록 하겠습니다.
이 메소드 코드만 뜯어서 다음 단락부터 소개하겠습니다!
2️⃣ getReviewListByShopIdAndName(request: ServerRequest) 코드를 살펴봅시다
우선 코드만 뜯어서 소개하겠습니다.
suspend fun getReviewListByShopIdAndName(request: ServerRequest): ServerResponse = coroutineScope {
// query param으로부터 shopId와 shopName을 받아오고, 없으면 예외처리
val shopId = request.queryParamOrNull("shop-id") ?: throw RequestParamLostException("shopId is lost!!")
val shopName = request.queryParamOrNull("shop-name") ?: throw RequestParamLostException("shopName is lost!!")
val reviewDtoList = mutableListOf<ShopReviewSimpleReadDto>()
val reviewFlow = shopReviewRepository.getShopReviewListFlowByShopIdAndNameWithCaching(shopId, shopName)
// 비동기적으로 reviewDtoList에 원소를 담는다
withContext(Dispatchers.IO) {
reviewFlow.buffer()
.collect {
val reviewDeferred = CoroutinesUtils.monoToDeferred(it)
val review = reviewDeferred.await()!!
reviewDtoList.add(toSimpleReadDto(review))
}
}
// review가 하나도 없다면 예외 처리
check(reviewDtoList.size != 0) {
throw ShopReviewNotFoundException("shopReview is not found!!")
}
return@coroutineScope ok().contentType(MediaType.APPLICATION_JSON)
.bodyValueAndAwait(resultFactory.getMultipleResult(reviewDtoList))
}
어지럽다 그죠?
위의 로직은 처리 순서가 다음과 같습니다.
1. query Parameter를 일단 가져온다
2. reviewDto를 쌓을 list를 하나 정의한다
3. repository로 부터 flow 변수를 가져온다
4. IODiapatcher를 이용해서 review를 db로부터 (혹은 redis에 히팅을 통해서 가져오거나) 가져와서 적재한다
5. dto 리스트를 공통 인터페이스로 포장을 해서 반환해준다
그런데 여기서 왜 IODispatcher를 이용해서 flow를 처리하는지 궁금할겁니다. 그 이유는 다음과 같은데요, 우선 위의 메소드 자체는 상위 코루틴이 존재하지 않기 때문에 일반적인 coroutineScope 라는 코루틴 빌더를 이용해서 열어주고, 데이터베이스에서 데이터를 가져오는 부분만 I/O에 특화된 Dispatcher인 IODispatcher를 이용하기 위함입니다.
위의 구현을 통해서 IODiapatcher를 사용하면서도 부모의 coroutine context와 일관성을 가지면서 비동기 처리를 구현할 수 있게됩니다. (CPS에 다른 흐름이 발생하지 않는다는 뜻이죠?)
그리고 flow를 buffer를 이용해서 구독을 하는 모습도 확인할 수 있는데요, flow에서 발행해주는 데이터들을 buffer에 적재하였다가 적재되는대로 바로 처리를 하기 위해서 buffer를 사용해보았습니다. 실제로도 buffer를 이용해서 flow의 속도 조절을 해줘야만 처리 속도가 더 빨라지기도 하구요.
사실 아직 Coroutines Flow을 공부 안해서 잘 몰라요...나중에 공부해보고 수정해드릴게요
따라서 IODispatcher를 이용해서 flow로부터 데이터를 뽑아오는 로직은 아래와 같아집니다.
// 비동기적으로 reviewDtoList에 원소를 담는다
withContext(Dispatchers.IO) {
reviewFlow.buffer()
.collect {
val reviewDeferred = CoroutinesUtils.monoToDeferred(it)
val review = reviewDeferred.await()!!
reviewDtoList.add(toSimpleReadDto(review))
}
}이 부분에서는 collect 전체 블록이 suspend point, 그리고 review를 가져오는 부분이 suspend point가 되기 때문에 reviewDtoList의 경우 모든 review가 다 적재될 때 까지 아래 라인의 로직을 타지 못할겁니다. 따라서 reviewDtoList는 모든 review를 품은 상태로 반환이 될수 있게되는겁니다!
3️⃣ 위의 Handler 메소드는 문제가 없을까?
사실 하나의 문제가 있습니다. Business logic이 Handler에서 묶여있기 때문에 엄밀한 단위 테스트가 힘들다는 점이 있습니다.
저도 아직 이 Business logic을 실제로 분리를 하지 않았는데요, 만약에 분리를 하게 된다면, handler에서 shopId, shopName을 검증하고 IODiapatcher를 이용해서 dtoList를 적재하는 로직을 Service layer로 내려버릴 것 같습니다.
이렇게 분리 구현을 하게 되면 비지니스 로직만을 service에서 단위 테스트가 가능해지기 때문에 아주 좋을겁니다.
다음으로, Router 부분을 테스트하도록 하겠습니다.
🔥 Router 구현을 보여드리겠습니다.
Router는 아주 간단합니다.
🔨 ShopReviewRouter.kt
@Configuration
class ShopReviewRouter(
private val shopReviewHandler: ShopReviewHandler
) {
@Bean
fun shopReviewRoutes() = coRouter {
"/v2/shop-review/simple".nest {
GET("", shopReviewHandler::findReviewByIdAndTitle)
GET("/list", shopReviewHandler::getReviewListByShopIdAndName)
}
}
}이거는 매우 간단하기 때문에 설명은 생략하도록 하겠습니다. 그저 shopReviewHandler를 Router에서 주입받아서 router를 작성했을 뿐이니까요.
🌲 글을 마치며
이번 포스트까지는 CQRS 패턴 중에서 Query 부분만을 다뤘습니다. 이제는 Command 모듈을 구현할 차례가 되었습니다. Command Module의 경우 다음 포스트부터 다룰 예정입니다!
다음 포스트에서 뵙겠습니다. 감사합니다!
🚀 References
토리맘의 한글라이즈 프로젝트 - Spring Webflux
레진 기술 블로그 - Kotlin과 Webflux 기반으로 컨텐츠 인증 서비스 개발 후기
Spring WebFlux - Functional Endpoint With Coroutine Tutorial
배민 광고리스팅 개발기
