날샘 코딩

• 리액티브 프로그래밍 기초 1 - 옵저버 패턴과 이터레이터 패턴
• 리액티브 프로그래밍 기초 2 - 비동기 프로그래밍
• 리액티브 프로그래밍 기초 3 - 리액티브 스트림

리액티브 프로그래밍(Reactive Programming)이란 데이터 또는 이벤트의 변경이 발생하면 이에 반응해 처리하는 프로그래밍 기법을 말합니다. 리액티브 프로그래밍은 2010년 에릭 마이어라는 프로그래머에 의해  마이크로소프트의 .NET 에코 시스템으로 정의되었습니다. 리액티브 프로그래밍의 특징은 비동기 프로그래밍을 처리하는 새로운 접근 방식이라는 것입니다. 리액티브 프로그래밍은 데이터의 통지, 완료, 에러에 대한 처리를 옵저버 패턴에 영감을 받아 설계되었고 데이터의 손쉬운 비동기 처리를 위해 함수형 언어의 접근 방식을 사용합니다.

리액티브 프로그래밍이 나오기 전 비동기 프로그래밍은 대부분 콜백(Callback) 기반의 비동기 처리 방식을 사용했습니다. 간단한 비동기 처리는 콜백 기반의 비동기 프로그래밍으로도 이해하기 쉬운 코드를 작성할 수 있지만 콜백이 많아져서 발생하는 일명 콜백 헬(Callback Hell)로 인해 코드의 복잡도가 늘어나서 가독성과 유지 보수성이 떨어지게 됩니다. 그동안 콜백 헬을 해결하기 위해 다양한 해결 방법들이 제시되었고 리액티브 프로그래밍을 적용하면 콜백 헬 없이 읽기 쉬운 코드를 작성할 수 있기 때문에 서버 애플리케이션 관점에서 리액티브 프로그래밍은 비동기 기반의 논-블로킹(Non-Blocking)과 이벤트 드리븐(Event-Driven) 애플리케이션 구현에 유리하여 최근 리액티브 프로그래밍을 적용한 사례가 점점 늘어나고 있습니다.

리액티브 프로그래밍은 비동기 처리 외에도 백프레셔(Back-Pressure)이라는 특징을 포함합니다. 백프레셔는 데이터를 소비하는 측에서 처리 가능한 만큼의 양을 데이터 제공자 측에 역으로 요청하는 기능입니다. 데이터 제공자와 데이터 소비자가 다른 스레드에서 비동기로 처리될 경우 주로 사용되며 제공자 측에서 너무 빠르게 데이터를 제공하면 소비하는 측에선 빠르게 처리하지 못해서 데이터가 점점 쌓이면 시스템에 장애 요소가 될 수 있는데 백프레셔는 이런 문제를 해결할 수 있습니다.

리액티브 프로그래밍과 디자인 패턴

리액티브 프로그래밍을 설명할 때 빠지지 않고 등장하는 디자인 패턴이 옵저버 패턴(Observer Pattern)과 이터레이터 패턴(Iterator Pattern)입니다. 리액티브 프로그래밍은 옵저버 패턴 그리고 이터레이터 패턴과 자주 비교되며 또한 영감을 받은 걸로 알려져 있습니다. 이번 절에 선 두 개의 패턴과 리액티브 프로그래밍 간에 어떤 차이점이 있는지 알아보겠습니다.

옵저버 패턴

옵저버 패턴이란 GoF가 소개한 디자인 패턴 중 하나로 관찰 대상이 되는 객체가 변경되면 대상 객체를 관찰하고 있는 옵저버(Observer)에게 변경사항을 통지(notify) 하는 디자인 패턴을 말합니다. 옵저버 패턴을 사용하면 객체 간의 상호작용을 쉽게 하고 효과적으로 데이터를 전달할 수 있습니다.

옵저버 패턴은 관찰 대상인 서브젝트(Subject)와 서브젝트를 관찰하는 옵저버로 이뤄져 있습니다. 하나의 서브젝트에는 하나 또는 여러 개의 옵저버를 등록할 수 있고 상태가 변경되면 자신을 관찰하는 옵저버들에게 변경 사항을 통지합니다. 서브젝트로부터 통지를 받은 옵저버는 부가적인 처리를 합니다. 

그림 1.1 옵저버 패턴의 구조

일반적으로 옵저버 패턴은 서브젝트와 옵저버를 상속하는 구체화(concrete) 클래스가 있습니다. 구체화 클래스는 서브젝트와 옵저버의 시그니처에 대한 상세 구현을 작성합니다. 구체화 클래스를 작성하기 위해 알아야 할 서브젝트와 옵저버의 필수 함수를 살펴보겠습니다.

표 1.1 서브젝트의 함수

표 1.2 옵저버의 함수

옵저버 패턴에 대해 간략히 알아봤으니 이번엔 JDK는 1.0 부터 포함된 Observable 클래스와 Observer 인터페이스를 사용해 간단한 커피 주문 예제를 만들어 보겠습니다. Observable은 옵저버 패턴의 서브젝트와 동일합니다.

예제 1.1 옵저버 패턴으로 구현한 커피 주문 예제

import java.util.*
 
class Customer : Observer {
 
    fun orderCoffee() = "Iced Americano"
 
    override fun update(o: Observable?, arg: Any?) {
         val coffee = arg as Coffee
         println("I got a cup of ${coffee.name}")
    }
}
 
class Coffee(val name: String)
 
class Barista : Observable() {
 
    private fun makeCoffee(name: String) = Coffee(name)
 
    fun serve(name: String) {
         setChanged()
         notifyObservers(makeCoffee(name))
    }
}
 
fun main(args: Array<String>) {
    val barista = Barista()
    val customer = Customer()
 
    barista.addObserver(customer)
    barista.serve(customer.orderCoffee())
}
 
--------------------
출력 결과)
--------------------
I got a cup of Iced Americano

• Customer 클래스는 Observer 인터페이스를 구현하여 Barista 클래스가 커피를 완성하면 통지를 받아서 update 함수에서 처리한다.
• Barista 클래스는 Observable 클래스를 상속하여 고객이 주문한 커피가 만들어지면 notifyObservers로 고객에게 만들어진 Coffee 객체를 전달한다. 이때 setChanged를 먼저 호출하여 변경 여부를 내부에 저장한다. 
• Customer 클래스가 Barista 클래스를 관찰하기 위해 addObserver로 등록한다.

예제 1.1과 같이 옵저버 패턴을 사용하지 않았다면 고객은 일정 간격으로 커피가 완성됐는지 바리스타에게 확인하는 처리가 있어야 합니다. 간격이 너무 짧으면 변경된 상태를 빠르게 확인할 수 있지만 매번 불필요한 호출이 발생하므로 성능상 문제가 발생할 수 있습니다. 또 간격이 너무 길면 변경된 상태를 즉시 확인할 수 없으므로 실시간성이 떨어질 수 있습니다. 옵저버 패턴은 관찰자인 옵저버가 서브젝트의 변화를 신경 쓰지 않고 상태 변경의 주체인 서브젝트가 변경사항을 옵저버에게 알려줌으로써 앞서 언급한 문제를 해결할 수 있습니다.

옵저버 패턴에서 서브젝트와 옵저버는 관심사에 따라 역할과 책임이 분리되어 있습니다. 서브젝트는 옵저버가 어떤 작업을 하는지 옵저버의 상태가 어떤지에 대해 관심을 가질 필요가 없습니다. 오직 변경 사항을 통지하는 역할만 수행하고 하나 혹은 다수의 옵저버는 각각 맡은 작업을 스스로 하기 때문에 옵저버가 하는 일이 서브젝트에 영향을 끼치지 않고 옵저버는 단순한 데이터의 소비자로서 존재하게 됩니다.

리액티브 프로그래밍은 옵저버 패턴의 서브젝트와 옵저버의 개념과 유사한 발행자와 구독자를 사용해 데이터를 통지하고 처리합니다. 데이터를 제공하는 측에서 데이터를 소비하는 측에 통지하는 방식을 일반적으로 푸시 기반 (Push-Based)이라고 부릅니다. 리액티브 프로그래밍은 옵저버 패턴의 단순한 데이터 통지 기능에 더해서 백프레셔라는 특징을 통해 데이터를 처리하는 측에서 데이터를 전달받는 개수를 역으로 요청할 수 있습니다. 그러므로 데이터를 통지하는 주체의 통지 속도를 제어하고 처리하는 측에서는 자신이 처리 가능한 만큼의 데이터만 받아서 처리할 수 있게 됩니다.

이터레이터 패턴

이터레이터 패턴은 옵저버 패턴과 마찬가지로 GoF가 소개한 디자인 패턴 중 한 가지입니다. 이터레이터 패턴은 데이터의 집합에서 원소라고 불리는 데이터를 순차적으로 꺼내기 위해 만들어진 디자인 패턴입니다. 이터레이터 패턴을 사용하면 다른 컬렉션을 사용해도 동일한 인터페이스를 사용해 데이터를 꺼내올 수 있기 때문에 컬렉션이 변경되더라도 사용하는 쪽에서는 변경사항이 발생하지 않습니다. 이러한 장점 때문에 이터레이터 패턴을 사용하면 코드를 좀 더 유연하고 확장성 있게 만듭니다.

그림 1.2 이터레이터 패턴의 구조

그림 1.2의 에그리게잇(Aggregate)은 자료구조의 리스트, 맵, 세트와 같은 구조가 집합체에 해당합니다. 이터레이터(Iterator)는 집합체 내부에 구현된 iterator를 이용해 생성합니다. 클라이언트는 생성된 이터레이터를 사용해 hasNext 함수에서  내부에 데이터가 존재하는지를 확인하고 next 함수를 사용해 데이터를 꺼내옵니다.

표 1.3  에그리게잇의 함수

표 1.4 이터레이터의 함수

이번에는 코틀린 컬렉션 라이브러리에 포함된 Iterable과 Iterator를 사용해 간단한 이터레이터 패턴을 구현해 보겠습니다.

예제 1.2 이터레이터 패턴 예제

package iterator
 
data class Car(val brand: String)
 
class Cars(val cars: List<Car> = listOf()) : Iterable<Car> {
    override fun iterator() = CarsIterator(cars)
}
 
class CarsIterator(val cars: List<Car> = listOf(), var index: Int = 0) : Iterator<Car> {
    override fun hasNext() = cars.size > index
    override fun next() = cars[index++]
}
 
fun main(args: Array<String>) {
    val cars = Cars(listOf(Car("Lamborghini"), Car("Ferrari")))
    val iterator = cars.iterator()
    while (iterator.hasNext()) {
         println("brand : ${iterator.next()}")
    }
}
 
--------------------
출력 결과)
--------------------
brand : Car(brand=Lamborghini)
brand : Car(brand=Ferrari)

• Cars 클래스는 Iterable 인터페이스를 구현하여 CarsIterator를 리턴하는 iterator 함수를 오버라이드 한다.
• CarsIterator 클래스는 Iterator 인터페이스를 구현하여 데이터가 존재하는지 확인하는 hasNext와 데이터가 존재하면 데이터를 가져오는 next 함수를 오버라이드 한다.
• while문 내부에선 hasNext를 사용하여 데이터를 모두 가져올때까지 반복하고 데이터를 출력한다.

데이터를 제공한다는 관점에서 이터레이터 패턴과 리액티브 프로그래밍은 유사하지만 이터레이터 패턴은 에그리게잇이 내부에 데이터를 저장하고 이터레이터를 사용해 데이터를 순차적으로 당겨오는 방식이기 때문에 풀 기반(Pull-Based)입니다.  이에 반해 리액티브 프로그래밍은 기본적으로 옵저버 패턴 처럼 데이터 제공자가 소비하는 측에 데이터를 통지하는 푸시 기반이기 때문에 분명한 차이점이 있습니다. 다만 구현체의 사양에 따라서 풀과 푸시를 모두 제공하는 경우도 있습니다.  

문제 확인

r2dbc-mysql을 사용해 개발한 애플리케이션에서 netstat | grep 3306으로 조회시 ESTABLISHED로 조회되는 개수가 커넥션 풀에 설정한 값과 상이한것을 확인함 .

r2dbc 풀 설정시 initSize가 20, maxSize가 50인데 아래와 같이 설정과는 다르게 커넥션이 생성되 있는 것을 확인함.

• A 인스턴스는 2개
• B 인스턴스는 10개
• C 인스턴스는 20개

테스트 환경

• Kotlin:1.4.21
• spring boot:2.3.7.RELEASE
• dev.miku:r2dbc-mysql:0.8.2.RELEASE

원인 파악

as-is

import io.r2dbc.pool.ConnectionPool
import io.r2dbc.pool.ConnectionPoolConfiguration
import io.r2dbc.spi.ConnectionFactories
import io.r2dbc.spi.ConnectionFactory
import io.r2dbc.spi.ConnectionFactoryOptions.*

@Configuration
@EnableR2dbcRepositories
class DatabaseConfiguration: AbstractR2dbcConfiguration() {

    override fun connectionFactory(): ConnectionFactory {

        val connectionFactory = ConnectionFactories.get(
            builder()
                .option(DRIVER, "mysql")
                .option(HOST, host)
                .option(USER, userName)
                .option(PORT, port)
                .option(PASSWORD, passWord)
                .option(DATABASE, db)
                .build())

        val configuration = ConnectionPoolConfiguration.builder(connectionFactory)
            .maxIdleTime(Duration.ofSeconds(maxIdleTime))
            .maxCreateConnectionTime(Duration.ofSeconds(maxCreateConnectionTime))
            .maxLifeTime(Duration.ofMinutes(maxLife))
            .initialSize(initialSize)
            .maxSize(maxSize)
            .build()

        return ConnectionPool(configuration)
    }

}    

분명 ConnectionPoolConfiguration에 initialSize와 maxSize를 넣었는데 비정상적으로 동작하는 것을 확인.

to-be

option(DRIVER, "pool")을 추가하였음. 이 설정은 추가하는 이유는 github.com/r2dbc/r2dbc-pool를 보면 Supported ConnectionFactory Discovery Options섹션에 아래와 같이 무조건 pool(Must be pool)로 설정하라고 되어 있음

import io.r2dbc.pool.PoolingConnectionFactoryProvider.*
import io.r2dbc.spi.ConnectionFactories
import io.r2dbc.spi.ConnectionFactory
import io.r2dbc.spi.ConnectionFactoryOptions.*

@Configuration
@EnableR2dbcRepositories
class DatabaseConfiguration: AbstractR2dbcConfiguration() {

   override fun connectionFactory(): ConnectionFactory =
        ConnectionFactories.get(
            builder()
                .option(DRIVER, "pool")
                .option(PROTOCOL, "mysql")
                .option(HOST, readPoolProperties.host)
                .option(USER, readPoolProperties.username)
                .option(PORT, readPoolProperties.port)
                .option(PASSWORD, readPoolProperties.password)
                .option(DATABASE, readPoolProperties.db)
                .option(MAX_SIZE, r2dbcPoolProperties.maxSize)
                .option(INITIAL_SIZE, r2dbcPoolProperties.initialSize)
                .option(MAX_IDLE_TIME, Duration.ofSeconds(r2dbcPoolProperties.maxIdleTime))
                .option(MAX_CREATE_CONNECTION_TIME, Duration.ofSeconds(r2dbcPoolProperties.maxCreateConnectionTime))
                .option(MAX_LIFE_TIME, Duration.ofMinutes(r2dbcPoolProperties.maxLife))
                .build()
        )

}        

netstat으로 조회시 초기값에 맞춰서 ESTABLISHED된 것을 확인함.

r2dbc-pool 소스를 디버그해보니 option(DRIVER, "pool")이 있고 없고에 따라서 ConnectionFactoryProvider가 달라지는 것을 확인 함

option(DRIVER, "mysql")로 설정시 dev.miku.r2dbc.MySqlConnectionFactoryProvider가 동작하고
option(DRIVER, "pool")로 설정시 io.r2dbc.pool.PoolingConnectionFactoryProvider가 동작하는데
dev.miku.r2dbc.MySqlConnectionFactoryProvider에는 initSize, maxSize에 대한 설정이 없고(github.com/mirromutth/r2dbc-mysqlgithub.com/mirromutth/r2dbc-mysql) 풀링되는 커넥션 팩토리가 아니기 때문에 그때그때마다 필요한 커넥션을 생성하는 방식이었음.

결론

r2dbc-pool을 사용해 커넥션 풀링을 정상적으로 이용하고 싶다면 option(DRIVER, "pool") 옵션은 필수이다.

이 글은 제가 레진 기술블로그에 동명의 제목으로 공유한 글입니다.

제가 일하고 있는 서비스 개발팀은 레진코믹스의 백엔드 서비스를 책임지고 있는 팀입니다. 저희는 작년부터 Kotlin과 Spring WebFlux를 메인 스택으로 선정하여 개발하고 있습니다. 이 글에선 WebFlux 기반의 컨텐츠 인증 서비스를 개발하면서 경험한 이슈들을 공유하려 합니다.

이 글은 WebFlux 또는 리액티브 프로그래밍에 대한 기초 지식을 다루진 않으므로 다소 불친절하게 느껴질 수 있는 점 양해 바랍니다. WebFlux에 대한 기초적인 내용은 제 블로그의 Spring WebFlux와 Kotlin으로 만드는 Todo 서비스 - 1편을 참고하시기 바랍니다.

이 글의 예제는 저희가 서비스하고 있는 코드를 공개용으로 변경했기 때문에 정상 동작을 보장하지 않습니다. 참고 부탁드립니다

이 글에서 다루고 있는 내용

• R2DBC
• Reactive Redis를 사용한 캐시 적용
• RestTemplate 대신 WebClient
• @Scheduled 대신 Flux.interval
• 쿠버네티스 환경 모니터링 시스템 구축

컨텐츠 인증 서비스

컨텐츠 인증 서비스는 유저가 레진코믹스 컨텐츠의 구매 여부를 판별하는 서비스입니다. 기본적으로 이미지 서버와 통신하며 유료 작품이라면 유저가 해당 작품을 구매했는지 판별하거나 무료 작품이라면 회원인지 비회원인지 판별하여 컨텐츠 접근 여부를 응답하는 단순하지만 없어선 안되는 서비스입니다.

컨텐츠 인증 서비스 기술 스택

• Kotlin 1.3.x
• Spring Boot 2.3.x
• Spring WebFlux
• Spring Data R2DBC 1.1.x
• Spring Data Redis Reactive
• Spring Boot Actuator

시스템 구성도

이해를 돕기 위해 작성한 간략한 시스템 구성도입니다.

R2DBC(Reactive Relational Database Connectivity)

컨텐츠 인증 서버에 앞서 전에도 WebFlux 기반의 서비스를 출시했었지만 JPA를 사용해 데이터베이스를 엑세스하고 있었습니다. JPA는 내부적으로 JDBC를 사용하기 때문에 DB I/O가 블로킹으로 동작합니다. 기본적으로 리액티브 스택은 블로킹 되는 구간이 있다면 전통적인 MVC 방식에 비해 얻는 이점이 거의 없기 때문에 리액터 공식 문서에서 설명하는 것처럼 아래와 같은 패턴을 사용해 해결하고 있었습니다.

Mono blockingWrapper = Mono.fromCallable(() -> { 
    return /* make a remote synchronous call */ 
});
blockingWrapper = blockingWrapper.subscribeOn(Schedulers.boundedElastic());

하지만 원칙적으로 리액티브 스택은 비동기-논블로킹 형태로 개발하는 것이 자연스럽고 최적의 성능을 보여줍니다. 그리고 때마침 리액티브 기반의 비동기-논블로킹을 지원하는 R2DBC의 GA 버전이 릴리즈 되면서 새로운 서비스에 적용하게 되었습니다.

Spring Data R2DBC 적용

R2DBC를 스프링 환경에서 쉽게 사용하는 방법은 Spring Data R2DBC를 사용하는 것입니다. Spring Data R2DBC 공식 깃헙을 보면 This is Not an ORM 이라는 문구가 가장 먼저 눈에 띕니다. 이것은 JPA가 ORM 프레임 워크임을 어필하는 것과 대조적인 부분입니다.
Spring Data R2DBC는 단순함을 지향하여 일반적인 ORM 프레임 워크가 지원하는 caching, lazy loading, write behind 등을 지원하지 않습니다.

ReactiveCrudRepository를 상속하는 리파지토리 인터페이스

Spring Data JPA나 Spring Data JDBC을 이미 사용하고 있어, Spring Data R2DBC를 적용할때 큰 어려움이 없었습니다.
Spring Data 프로젝트는 리액티브 패러다임을 지원하는 ReactiveCrudRepository를 제공하므로 ReactiveCrudRepository 인터페이스를 상속하는 인터페이스를 만들어 주면 쉽게 Spring Data R2DBC를 사용할 수 있습니다.

interface ContentRepository : ReactiveCrudRepository<Content, Long> {
    fun findFirstByUserIdAnContentIdOrderByIdDesc(userId: Long, contentId:Long) : Mono<Content>
}

멀티 데이터 소스 구현

실무에서 서비스를 개발하다 보면 하나의 서버에서 여러 데이터 소스에 대한 접근이 필요할 수 있습니다. Spring Data R2DBC는 이런 경우에 사용할 수 있도록 AbstractRoutingConnectionFactory라는 추상 클래스를 지원합니다.

MultiTenantRoutingConnectionFactory

MultiTenantRoutingConnectionFactory는 AbstractRoutingConnectionFactory를 상속받아서 determineCurrentLookupKey 함수를 오버라이드 하였습니다.

class MultiTenantRoutingConnectionFactory : AbstractRoutingConnectionFactory() {

    override fun determineCurrentLookupKey(): Mono<Any> {
        return TransactionSynchronizationManager.forCurrentTransaction().map {
            if (it.isActualTransactionActive) {
                if (it.currentTransactionName?.contains(SecondaryDataSourceProperties.BASE_PACKAGE)!!) {
                    SecondaryDataSourceProperties.KEY
                } else {
                    PrimaryDataSourceProperties.KEY
                }
            } else {
                PrimaryDataSourceProperties.KEY
            }
        }
    }
}

determineCurrentLookupKey에선 현재 트랜잭션의 이름을 읽어서 @Transactional을 선언한 서비스의 패키지 기준으로 분기 처리하였습니다. determineCurrentLookupKey의 상세 구현은 개발자가 다양한 방법을 사용할 수 있으므로 이것과 같은 방법을 사용하지 않아도 됩니다.예를 들면 @Transactional(readOnly=true)인 경우엔 리플리케이션으로 라우팅하는 경우도 있을 수 있습니다. 이런 경우엔 현재 트랜잭션이 readOnly인지 판단하는 로직을 내부에 구현해주면 됩니다.

application.yml

application.yml에는 두개의 데이터 소스 설정을 추가하였습니다.

datasource:
  primary:
    url:
    username: 
    password: 
  secondary:
    url:
    username:
    password:

DatasourceProperties

각 데이터 소스의 설정 정보는 DatasourceProperties라는 이름으로 추가하였습니다.@ConfigurationProperties는 application.yml에 지정한 설정 정보의 prefix를 기준으로 동일한 이름의 프로퍼티가 존재하면 자동으로 값을 세팅해 줍니다.@ConstructorBinding은 setter가 아닌 생성자를 사용해 값을 바인딩 해주는 애노테이션입니다. 코틀린의 Data class와 조합하면 설정 클래스를 Immutable 하게 사용할 수 있습니다.

@ConstructorBinding
@ConfigurationProperties(prefix = "datasource.primary")
data class PrimaryDataSourceProperties(val username: String, val password: String, val url: String) {

    companion object {
        const val KEY = "primary"
        const val BASE_PACKAGE = "com.lezhin.backend.service.primary"
        const val TRANSACTION_MANAGER = "primaryTransactionManager"
    }
}

@ConstructorBinding
@ConfigurationProperties(prefix = "datasource.secondary")
data class SecondaryDataSourceProperties(val username: String, val password: String, val url: String) {

    companion object {
        const val KEY = "secondary"
        const val BASE_PACKAGE = "com.lezhin.backend.service.secondary"
        const val TRANSACTION_MANAGER = "secondaryTransactionManager"
    }
}

R2dbcConfig

그 다음 R2dbc 설정 클래스를 만들어서 connectionFactory빈을 생성해 주면 됩니다.

@Configuration
@EnableR2dbcRepositories
class R2dbcConfig(val primaryDataSourceProperties: PrimaryDataSourceProperties,
                  val secondaryDataSourceProperties: SecondaryDataSourceProperties) : AbstractR2dbcConfiguration() {
    @Bean
    override fun connectionFactory(): ConnectionFactory {
        val multiTenantRoutingConnectionFactory = MultiTenantRoutingConnectionFactory()

        val factories = HashMap<String, ConnectionFactory>()
        factories[PrimaryDataSourceProperties.KEY] = primaryConnectionFactory()
        factories[SecondaryDataSourceProperties.KEY] = secondaryConnectionFactory()

        multiTenantRoutingConnectionFactory.setDefaultTargetConnectionFactory(primaryConnectionFactory())
        multiTenantRoutingConnectionFactory.setTargetConnectionFactories(factories)
        return multiTenantRoutingConnectionFactory
    }

    @Bean
    fun primaryConnectionFactory() =
        parseAndGet(Triple(primaryDataSourceProperties.url, primaryDataSourceProperties.username, primaryDataSourceProperties.password))

    @Bean
    fun secondaryConnectionFactory() =
        parseAndGet(Triple(secondaryDataSourceProperties.url, secondaryDataSourceProperties.username, secondaryDataSourceProperties.password))

    private fun parseAndGet(propertiesAsTriple: Triple<String, String, String>): ConnectionFactory {
        val (url,username,password) = propertiesAsTriple

        val properties = URLParser.parseOrDie(url)
        return JasyncConnectionFactory(MySQLConnectionFactory(
            com.github.jasync.sql.db.Configuration(
                username = username,
                password = password,
                host = properties.host,
                port = properties.port,
                database = properties.database,
                charset = properties.charset,
                ssl = properties.ssl
            )))
    }

    @Bean
    fun primaryTransactionManager(@Qualifier("primaryConnectionFactory") connectionFactory: ConnectionFactory?) =
        R2dbcTransactionManager(connectionFactory!!)

    @Bean
    fun secondaryTransactionManager(@Qualifier("secondaryConnectionFactory") connectionFactory: ConnectionFactory?) =
        R2dbcTransactionManager(connectionFactory!!)
}

Reactive Redis를 사용한 캐시 적용

저희는 인프라에 따라서 Redis와 Memcached를 모두 사용하고 있습니다. 이번 컨텐츠 인증 서비스에선 Redis를 사용해 DB에서 읽어온 데이터를 캐시하였습니다. WebFlux를 적용하면서 @Cacheable 사용이 불가하여 별도로 Reactor 호환 캐시 유틸을 만들어서 사용했었습니다만 이번에는 Spring Data Redis Reactive를 직접 사용하는 방법으로 구현하게 되었습니다.

RedisConfig

Redis 클라이언트로 사용될 Lettuce 설정과 ReactiveRedisTemplate에 대한 설정 입니다.

@Configuration
class RedisConfig(@Value("\${spring.redis.host}") val host: String,
                   @Value("\${spring.redis.port}") val port: Int,
                   @Value("\${spring.redis.database}") val database: Int) {

    @Primary
    @Bean("primaryRedisConnectionFactory")
    fun connectionFactory(): ReactiveRedisConnectionFactory? {
        val connectionFactory = LettuceConnectionFactory(host, port)
        connectionFactory.database = database
        return connectionFactory
    }

    @Bean
    fun contentReactiveRedisTemplate(factory: ReactiveRedisConnectionFactory?): ReactiveRedisTemplate<String, Content>? {
        val keySerializer = StringRedisSerializer()
        val redisSerializer = Jackson2JsonRedisSerializer(Content::class.java)
            .apply {
                setObjectMapper(
                    jacksonObjectMapper()
                        .registerModule(JavaTimeModule())
                )
            }
        val serializationContext = RedisSerializationContext
            .newSerializationContext<String, Content>()
            .key(keySerializer)
            .hashKey(keySerializer)
            .value(redisSerializer)
            .hashValue(redisSerializer)
            .build()
        return ReactiveRedisTemplate(factory!!, serializationContext)
    }
}

ContentService

구현은 일반적인 캐시처리 방식으로 cache hit 여부를 판단하여 fallback 처리 하는 구조입니다.

@Service
class ContentService(val contentRepository: ContentRepository,
                      val contentRedisOps: ReactiveRedisOperations<String, Content>) {

    companion object {
        const val DAYS_TO_LIVE = 1L
    }

    @Transactional("primaryTransactionManager")
    fun findById(id: Long): Mono<Content> {
        val key = "content:${id}"
        return contentRedisOps.opsForValue().get(key).switchIfEmpty {
            contentRepository.findById(id).doOnSuccess {
                    contentRedisOps.opsForValue()
                        .set(key, it, Duration.ofDays(DAYS_TO_LIVE))
                        .subscribe()
            }.onErrorResume {
                Mono.empty()
            }
        }
    }
}

1. Redis에 캐시된 데이터가 있는지 확인하여 존재하지 않으면 switchIfEmpty 내부 코드 동작
   • 이때 사용하는 switchIfEmpty는 Mono.defer로 감싸진 Reactor 코틀린 확장 함수이므로 Mono.defer를 직접 사용하지 않아도 됩니다.
   • fun <T> Mono<T>.switchIfEmpty(s: () -> Mono<T>): Mono<T> = this.switchIfEmpty(Mono.defer { s() })
2. switchIfEmpty 내부 코드에는 우선 DB에서 조회하여 데이터가 존재하면 에서 내부 코드 동작
3. 데이터가 존재하면 Redis에 캐시위와 같은 방식으로 캐시 레이어를 리액티브하게 구현할 수 있었습니다.

RestTemplate 대신 WebClient

외부 서비스와 HTTP로 통신해야 하는 경우 가장 흔한 방법은 RestTemplate을 사용하는 것입니다. RestTemplate은 Spring 애플리케이션에서 가장 일반적인 웹 클라이언트지만 블로킹 API이므로 리액티브 기반의 애플리케이션에서 성능을 떨어트리는 원인이 될 수 있습니다.
이런 이유로 Spring5에서 추가된 WebClient를 사용해 리액티브 기반의 비동기-논블로킹 통신을 구현하였습니다.

UserTokenRepositoryImpl

UserTokenRepositoryImpl은 유저의 토큰을 가지고 WebClient를 사용해 유저 서비스를 호출해 받아온 결과를 유저 객체 매핑합니다. 이러한 동작은 모두 리액티브 하게 비동기-논블로킹 형태로 동작합니다.

@Repository
class UserTokenRepositoryImpl(val webClientBuilder: WebClient.Builder) : UserTokenRepository {

    private lateinit var webClient: WebClient

    @PostConstruct
    fun setWebClient() {
        webClient = webClientBuilder.build()
    }

    override fun fetchUser(uri: String, accessToken: String) =
        webClient.get()
            .uri(uri)
            .header("Authorization", "Bearer $accessToken")
            .retrieve()
            .onStatus(HttpStatus::isError) {
                Mono.error(ForbiddenException("Connection Failed"))
            }
            .bodyToMono(User::class.java)
}

@Scheduled 대신 Flux.interval

WebFlux로 서비스를 만들다 보면 기존에는 당연하게 사용하던 기능이 지원되지 않는 경우가 많습니다. @Scheduled도 그중 하나입니다. 만약 일정 간격으로 처리할 작업이 있는 경우 @Scheduled의 대안으로 Flux.interval을 사용할 수 있습니다.

Flux.interval 예시

아래 예제는 90초 간격으로 로그를 출력하도록 만든 예제입니다.

    @PostConstruct
    fun doProcess() = Flux.interval(Duration.ofSeconds(90))
            .map {
                // 구현
                logger.info("hello world")
            }
            .subscribe()

쿠버네티스 환경 모니터링 시스템 구축

저희 레진은 새로 개발하는 서비스는 모두 쿠버네티스 환경에서 운영 중 이고 prometheus를 쿠버네티스 환경에서 기본 모니터링 툴로 사용하고 있습니다.
prometheus는 풀 방식 으로 데이터를 수집합니다. Spring Boot2 부터 메트릭 표준이 된 micrometer를 사용해 prometheus를 지원하는 메트릭을 추가하였습니다.

build.gradle.kts

우선 메트릭 생성에 필요한 의존성을 build.gradle.kts에 추가하였습니다.

dependencies {
    implementation("org.springframework.boot:spring-boot-starter-actuator")
    implementation(group = "io.micrometer", name = "micrometer-registry-prometheus", version = "1.5.1")
}

application.yml

그다음 아래와 같은 형태로 메트릭 엔드포인트를 추가해줍니다. 기본적으로 Spring Boot Actuator는 include에 포함된 엔드포인트만 엑세스가 가능합니다.

management:
  endpoints:
    web:
      exposure:
        include: metrics, prometheus

더 많은 엔드포인트에 대한 정보는 엔드포인트 공식 문서에서 확인할 수 있습니다.

prometheus 엔드포인트

웹 브라우저에서 /actuator/prometheus로 접속해보면 이미지와 같이 prometheus 형식의 메트릭 정보가 출력되는 것을 확인 할 수 있습니다.

)

Grafana 대쉬보드

최종적으로 엔드포인트에서 출력된 정보를 prometheus에서 10초 간격으로 데이터를 풀링해가고 Grafana 대쉬보드에서 시각화된 정보를 관측할 수 있게되었습니다.

)

회고

이번 컨텐츠 인증 서버 개발은 리액티브 프로그래밍과 WebFlux에 대해 좀 더 익숙해지는 기회가 되었습니다. 또 Spring Data R2DBC의 GA 버전이 릴리즈되면서 모든 구간에서 리액티브 스택을 적용할 수 있었습니다.
이외에도 WebFlux와 R2DBC에 대해 공유하고 싶은 주제들이 더 있지만 한편의 블로그로 작성하기엔 너무 길어지는 내용이 될 것 같아서 생략하였습니다.

원문

개요

API를 개발하다보면 처리에 따라서 응답에 특정한 헤더를 내려준다던지, 스테이터스 코드를 조작하여 응답하는 경우가 있다. 이럴때 전통적인 Spring MVC를 사용하면 문제가 되지 않지만 Spring WebFlux로 개발하게 되면 기본적으로 Mono 또는 Flux로 리턴하는것이 원칙이기 때문에 어떻게 처리해야할지 난감스러울 수 있다. 

Spring WebFlux는 함수형 방식과 애노테이션 방식 이렇게 2가지 프로그래밍 모델을 지원한다. 함수형 방식은 ServerResponse를 사용해서 부가적인 응답을 처리할 수 있지만, 애노테이션 방식으로 개발을 하면서 ServerResponse를 사용하게 되면
Caused by: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: No serializer found for class org.springframework.web.reactive.function.server.DefaultServerResponseBuilder$WriterFunctionResponse and no properties discovered to create BeanSerializer (to avoid exception, disable SerializationFeature.FAIL_ON_EMPTY_BEANS)과 같은 에러를 볼수있다.

이에 대한 해결책으로 Spring MVC에서 사용하던 ResponseEntity를 그대로 사용하면된다. 아래 코드를 보고 확인해보자.

예제

아래 코드는 사용자가 성인인지 체크한 후 맞다면 "X-User-Adult" 헤더와 스테이터스 코드 200 Ok를 응답으로 내려주고, 성인이 아니라면 403 Fobidden을 응답하도록 만든 예제이다.

결론

예제처럼 Mono 또는 Flux로 ResponseEntity를 감싸면 쉽게 헤더 및 스테이터스 코드 조작이 가능하다.