들어가며

스타크래프트2의 프로토스 종족에는 '시간 증폭'이라는 독특한 메커니즘이 있다. 연결체(Nexus)에서 프로브를 생산할 때 시간 증폭을 사용하면 일정 시간 동안 생산 속도가 빨라지는 기능이다. 이런 상태 관리와 비동기 처리가 필요한 시스템을 Kotlin의 Coroutine Flow를 활용해 구현해보자.

https://github.com/waterfogSW/starcraft-time-amplification

 

시스템 요구사항

  1. 연결체는 프로브를 생산할 수 있다
  2. 생산 큐는 최대 5개까지 가능
  3. 시간 증폭은 10초 동안 지속되며, 적용 시 생산 속도가 3배로 증가
  4. 생산 진행 상태를 실시간으로 관찰 가능
  5. 생산 중인 항목을 취소할 수 있음

 

연결체 프로브 생산 프로세스

 

1. 전통적인 방식 (Observer 패턴)

직관에 따라 Nexus(연결체) 를 구현한다면 Variable과 Observer 패턴을 사용하는 방식을 떠올릴 수 있다.

class Nexus {
    private var productionState: ProductionState = ProductionState.Idle
    private var probeCount: Int = 0
    private val observers = mutableListOf<ProductionObserver>()
    
    // 메모리 누수 위험이 있는 observer 등록/해제
    fun addObserver(observer: ProductionObserver) {
        observers.add(observer)
    }
    
    fun removeObserver(observer: ProductionObserver) {
        observers.remove(observer)
    }
    
    // 스레드 안전성을 위해 모든 메서드에 동기화 필요
    @Synchronized
    fun updateState(newState: ProductionState) {
        productionState = newState
        observers.forEach { it.onStateChanged(newState) }
    }
    
    // 여러 상태를 변경할 때 데드락 위험
    @Synchronized
    fun completeProduction() {
        productionState = ProductionState.Complete
        probeCount++
        observers.forEach { 
            it.onStateChanged(productionState)
            it.onProbeCountChanged(probeCount)
        }
    }
}

// Activity/Fragment에서 메모리 누수 발생 가능
class NexusActivity : Activity(), ProductionObserver {
    private val nexus = Nexus()
    
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        nexus.addObserver(this) // 등록은 했지만...
    }
    
    // onDestroy에서 removeObserver를 호출하지 않으면 메모리 누수
}

전통적인 Observer 패턴 기반의 상태 관리 방식은 근본적인 문제점을 가지고 있다. 개발자가 직접 Observer를 등록하고 해제하는 과정에서 메모리 누수가 발생하기 쉽다. Activity나 Fragment의 생명주기와 Observer의 생명주기를 수동으로 동기화해야 하는데, 이는 실수하기 쉬운 작업이다.

멀티스레드 환경에서의 안전성도 보장하기 어렵다. 상태 변경 메소드마다 @Synchronized 어노테이션을 붙여야 하며, 이는 성능 저하를 일으킨다. 여러 상태를 동시에 변경할 때는 데드락이 발생할 위험도 있다. 결과적으로 복잡한 보일러플레이트 코드가 생기고 유지보수가 어려워진다.

 

2. LiveData로의 진화

그럼 LiveData는 어떤가?, LiveData는 이러한 문제를 상당 부분 해결한다. 생명주기를 자동으로 관리하여 메모리 누수를 방지하고, 메인 스레드 안전성을 보장한다. Observer 등록과 해제를 수동으로 관리할 필요가 없어졌다.

class Nexus {
    // 안드로이드 플랫폼 종속적
    private val _productionState = MutableLiveData<ProductionState>()
    val productionState: LiveData<ProductionState> = _productionState
    
    private val _probeCount = MutableLiveData<Int>()
    val probeCount: LiveData<Int> = _probeCount
    
    // 메인 스레드에서만 값 변경 가능
    fun updateState(newState: ProductionState) {
        _productionState.value = newState
    }
    
    // 빠른 연속 업데이트 처리 불가능
    fun startFastUpdates() {
        viewModelScope.launch {
            repeat(1000) { // 빠른 업데이트 발생
                _productionState.value = ProductionState.Producing(it / 1000f)
                delay(16) // 16ms마다 업데이트
                // 백프레셔 처리 메커니즘 부재로 프레임 드롭 발생
            }
        }
    }
}

그러나 LiveData는 안드로이드 플랫폼에 종속되어 있어 서버사이드에서는 사용할 수 없다. 역시 순수 Kotlin 모듈에서는 사용이 불가능하며, 이로 인해 멀티플랫폼 개발과 테스트가 제한된다.

더 큰 문제는 연속적인 데이터 업데이트 처리에 취약하다는 점이다. 센서 데이터와 같이 빠르게 들어오는 데이터 스트림을 처리할 때 데이터 손실이 발생하거나 UI 성능이 저하된다. 이는 LiveData가 백프레셔 처리 메커니즘을 가지고 있지 않기 때문이다.

 

3. Flow의 도입과 해결

Flow는 이전 방식들의 문제를 해결하고 더 강력한 기능을 제공한다.

class Nexus {
    private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
    
    // 플랫폼 독립적이며 상태 관리가 용이한 StateFlow
    private val _productionState = MutableStateFlow<ProductionState>(ProductionState.Idle)
    val productionState = _productionState.asStateFlow()
    
    private val _probeCount = MutableStateFlow(0)
    val probeCount = _probeCount.asStateFlow()
    
    // 백프레셔가 적용된 생산 프로세스
    fun startProduction() = flow {
        var progress = 0f
        while (progress < 1f) {
            progress += 0.1f
            emit(progress) // 백프레셔 자동 적용
            delay(16)
        }
    }.buffer(Channel.BUFFERED) // 생산자-소비자 분리
     .catch { e -> 
        _productionState.value = ProductionState.Idle
        emit(0f)
    }
    
    // 복잡한 상태 조합도 쉽게 처리
    val combinedState = combine(
        productionState,
        probeCount
    ) { state, count ->
        CombinedState(state, count)
    }.stateIn(scope, SharingStarted.Lazily, CombinedState())
    
    // 스레드 안전한 상태 업데이트
    fun updateState(newState: ProductionState) {
        _productionState.value = newState // 자동으로 스레드 안전
    }
    
    fun shutdown() {
        scope.cancel() // 리소스 정리도 간단
    }
}

// UI에서의 사용 (Compose)
@Composable
fun NexusScreen(nexus: Nexus) {
    val state by nexus.productionState.collectAsState()
    
    LaunchedEffect(Unit) {
        nexus.startProduction()
            .collect { progress ->
                // 백프레셔로 인해 UI는 버벅임 없이 부드럽게 업데이트
            }
    }
}

// 순수 Kotlin 모듈에서도 사용 가능
class PureKotlinModule {
    private val nexus = Nexus()
    
    fun processData() {
        // Flow는 플랫폼 독립적이므로 사용 가능
        nexus.productionState
            .map { ... }
            .filter { ... }
            .collect { ... }
    }
}

이처럼 Flow와 StateFlow는 이전 방식들의 한계를 모두 해결하고, 코루틴과의 자연스러운 통합, 백프레셔 지원, 플랫폼 독립성 등 다양한 이점을 제공한다. 특히 상태 관리에 있어서 스레드 안전성과 메모리 관리가 자동으로 이루어지며, 복잡한 비동기 처리도 간단하게 구현할 수 있다.

 

Flow와 StateFlow

Flow의 기본 개념

Flow는 코틀린에서 비동기적으로 계산될 수 있는 데이터 스트림을 나타내는 타입이다. Flow는 다음과 같은 특징을 가진다:

// Flow의 기본 구조
val flow = flow {
    for (i in 1..3) {
        delay(100) // 비동기 작업 시뮬레이션
        emit(i)    // 값 방출
    }
}

// Flow 수집
scope.launch {
    flow.collect { value ->
        println(value)
    }
}

Flow의 주요 특징

  1. Cold Stream: Flow는 collect를 호출할 때만 데이터를 방출한다
  2. 순차적 실행: 기본적으로 순차적으로 처리된다
  3. 취소 가능: 코루틴의 취소 메커니즘을 지원한다
  4. 백프레셔 지원: 데이터 생산과 소비 속도를 조절할 수 있다

 

StateFlow 이해하기

StateFlow는 Flow의 특별한 형태로, 항상 값을 가지고 있는 상태 홀더다. 우리의 연결체 구현에서 중요한 역할을 한다

// StateFlow 기본 예제
private val _state = MutableStateFlow(초기값)
val state: StateFlow<T> = _state.asStateFlow()

// 값 업데이트
_state.value = 새로운값
// 또는
_state.update { currentValue -> 
    // 새로운 값을 계산하고 반환
}

StateFlow의 특징

  1. Hot Stream: 수집하는 코루틴이 없어도 활성 상태를 유지
  2. 상태 보유: 항상 현재 값을 가짐
  3. 중복 제거: 동일한 값은 방출하지 않음
  4. 다중 구독자 지원: 여러 수집기가 동시에 값을 관찰할 수 있음

 

생산 프로세스, 시간 증폭 프로세스 구현

앞서 본 내용들을 바탕으로 연결체의 생산 프로세스와 시간 증폭 프로세스를 구현해 보자.

생산 프로세스

fun startProduction() {
    if (_productionQueue.value.size >= 5) return

    scope.launch {
        // 큐에 새 항목 추가
        _productionQueue.update { it + ProbeQueueItem() }

        // 생산이 진행중이지 않은 경우에만 시작
        if (_productionState.value is ProductionState.Idle) {
            startProductionProcess()
        }
    }
}

private fun startProductionProcess() {
    productionJob = scope.launch {
        while (_productionQueue.value.isNotEmpty()) {
            var accumulatedProgress = 0f
            var lastUpdateTime = System.currentTimeMillis()

            // 진행률 업데이트 루프
            while (accumulatedProgress < 1f) {
                val currentTime = System.currentTimeMillis()
                val deltaTime = currentTime - lastUpdateTime
                lastUpdateTime = currentTime

                // 시간 증폭 상태 확인
                val isCurrentlyBoosted = _chronoBoostState.value.isActive
                val progressIncrement = calculateProgress(deltaTime, isCurrentlyBoosted)
                
                accumulatedProgress = (accumulatedProgress + progressIncrement)
                    .coerceAtMost(1f)

                // 상태 업데이트
                updateProductionState(ProductionState.Producing(accumulatedProgress))
                updateQueueProgress(accumulatedProgress)
                
                delay(16) // ~60fps
            }

            // 생산 완료 처리
            completeProduction()
        }
    }
}

 

시간 증폭 프로세스

동일한 메커니즘으로 시간 증폭도 구현할 수 있다.

fun applyChronoBoost() {
    chronoBoostJob = scope.launch {
        updateChronoBoostState(
            ChronoBoostState(
                isActive = true,
                remainingTimeMillis = CHRONOBOOST_DURATION
            )
        )
        
        val startTime = System.currentTimeMillis()
        while (true) {
            val remainingTime = CHRONOBOOST_DURATION - (System.currentTimeMillis() - startTime)
            if (remainingTime <= 0) break
            
            updateChronoBoostState(
                _chronoBoostState.value.copy(remainingTimeMillis = remainingTime)
            )
            delay(100)
        }
    }
}

시간 증폭은 별도의 코루틴에서 관리되며, 상태를 통해 생산 속도에 영향을 준다. 생산 프로세스는 지속적으로 시간 증폭 상태를 확인하며 생산 속도에 반영해 결정하게 된다.

 

서버사이드에서의 응용

앞서 살펴본 Flow와 StateFlow를 활용한 생산 및 시간 증폭 프로세스 구현은 서버사이드 애플리케이션에서도 효과적으로 응용할 수 있다. 서버 환경에서는 다수의 클라이언트 요청을 효율적으로 처리하고, 실시간 상태 관리를 통해 시스템의 성능과 안정성을 유지하는 것이 중요하다. Kotlin의 Coroutine과 Flow는 이러한 요구사항을 충족시키는 강력한 도구를 제공한다.

실시간 게임 서버에서는 플레이어의 상태 업데이트, 게임 내 이벤트 처리, 자원 관리 등이 빈번하게 발생한다. Flow와 StateFlow를 활용하면 각 플레이어의 상태 변화를 비동기적으로 처리하고, 서버 자원의 효율적인 분배를 통해 높은 동시성을 유지할 수 있다. 예를 들어, 플레이어의 행동에 따른 자원 생산 속도를 조절하거나, 특정 이벤트 발생 시 일시적으로 생산 속도를 증가시키는 시간 증폭 메커니즘을 구현할 수 있다.

또한 실시간 데이터 스트리밍 이 필요한 금융 거래 시스템, IoT 데이터 수집, 실시간 분석 등 고속의 데이터 스트림을 처리해야 하는 서버 애플리케이션에서 Flow는 자연스러운 선택이 될 수 있다. 백프레셔 지원을 통해 데이터 생산자와 소비자 간의 속도 차이를 조절할 수 있으며, 상태 관리 기능을 통해 현재 데이터 처리 상태를 실시간으로 모니터링할 수 있다. 이는 데이터 손실을 방지하고 시스템의 안정성을 높이는 데 기여한다.

 

결론

Kotlin의 Coroutine과 Flow는 비동기 처리와 상태 관리를 효율적으로 구현할 수 있는 강력한 도구를 제공한다. 특히 서버사이드 애플리케이션에서는 높은 동시성과 실시간 처리가 요구되는 다양한 상황에서 Flow와 StateFlow의 장점을 극대화할 수 있다. 시간 증폭과 같은 복잡한 비동기 메커니즘을 손쉽게 구현하고, 안정적인 상태 관리를 통해 시스템의 성능과 신뢰성을 높일 수 있다. 따라서, 서버 개발자들은 Flow와 StateFlow를 적극적으로 활용하여 더욱 효율적이고 확장 가능한 서버 애플리케이션을 개발할 수 있을 것이다.

Spring Cloud OpenFeign이 feature complete 상태가 되었다. Spring Cloud OpenFeign 공식문서에서는 RestClient와 Http interface를 활용한 방식으로의 마이그레이션을 권장하고 있다. Spring 6에서 제공하는 HTTP Interface는 OpenFeign의 선언형 프로그래밍 방식을 대체할 수 있다. 이 글에서는 OpenFeign에서 HTTP Interface로 마이그레이션하는 방법을 설명한다.

 

의존성 구성

dependencies {
    implementation("org.springframework.boot:spring-boot-starter-web")
    implementation("org.springframework.retry:spring-retry")
    implementation("org.springframework:spring-aspects")
}

먼저 필요한 의존성을 설정해야 한다. Spring Boot 3.x 이상을 사용하면 별도의 HTTP Interface 관련 의존성은 필요하지 않다. OpenFeign에서 사용했던 재시도 기능을 HttpExchange에서 재현하기 위해 spring-retry만 추가하면 된다

 

OpenFeign에서 HTTP Interface 코드 비교

인터페이스 정의 방식의 변화

기존 OpenFeign에서는 다음과 같이 클라이언트를 정의했다:

@FeignClient(name = "user-service", url = "${user.service.url}")
interface UserClient {
    @GetMapping("/users/{id}")
    fun getUser(@PathVariable id: Long): User
    
    @PostMapping("/users")
    fun createUser(@RequestBody user: User): User
}

HTTP Interface에서는 이렇게 변경된다:

@HttpExchange
interface UserClient {
    @GetExchange("/users/{id}")
    fun getUser(@PathVariable id: Long): User

    @PostExchange("/users")
    fun createUser(@RequestBody user: User): User
    
    @PutExchange("/users/{id}")
    fun updateUser(
        @PathVariable id: Long,
        @RequestBody user: User
    ): User

    @DeleteExchange("/users/{id}")
    fun deleteUser(@PathVariable id: Long)
}

주요 변경사항을 살펴보면:

  1. @FeignClient 어노테이션이 @HttpExchange로 변경됐다
  2. HTTP 메서드 어노테이션들이 각각 대응되는 Exchange 어노테이션으로 변경됐다
    • @GetMapping → @GetExchange
    • @PostMapping → @PostExchange
    • @PutMapping → @PutExchange
    • @DeleteMapping → @DeleteExchange
  3. 기본적인 요청/응답 구조는 유사하게 유지된다

이러한 변경은 Spring의 기본 기능을 활용하면서도, OpenFeign의 직관적인 인터페이스 스타일을 유지하고 있다.

 

재시도 전략

OpenFeign에서 제공하던 exponential backoff 기능은 Spring Retry를 통해 구현할 수 있다. 기존 OpenFeign이 별도의 의존성 없이 재시도 기능을 지원했던 반면 Http Interface를 사용할때는 Srping Retry가 필요하다.

@Retryable(
    include = [RuntimeException::class],
    maxAttempts = 3,
    backoff = Backoff(delay = 1000)
)
@GetExchange("/users/retry/{id}")
fun getUserWithRetry(
    @PathVariable id: Long,
    @RequestParam("failCount", required = false) failCount: Int?
): User

주요 설정:

  • @EnableRetry: 애플리케이션 레벨에서 재시도 기능 활성화, 
  • @Retryable: 메서드 레벨에서 재시도 정책 설정
  • include: 재시도할 예외 타입 지정
  • maxAttempts: 최대 재시도 횟수
  • backoff: 재시도 간격 설정

 

에러 처리

OpenFeign의 ErrorDecoder와 달리, RestClient는 defaultStatusHandler를 통해 HTTP 상태 코드별 에러 처리를 구현할 수 있다. 상태 코드에 따라 적절한 예외를 던지도록 설정이 가능하다.

@Configuration
class RestClientConfig {

    @Bean
    fun defaultRestClientBuilder(): RestClient.Builder {
        return RestClient
            .builder()
            .defaultStatusHandler(HttpStatusCode::isError) { _, response ->
                when (response.statusCode) {
                    HttpStatus.NOT_FOUND -> throw RestClientException("Resource not found")
                    HttpStatus.UNAUTHORIZED -> throw RestClientException("Unauthorized")
                    HttpStatus.BAD_REQUEST -> throw RestClientException("Invalid request")
                    else -> throw RestClientException("HTTP error: ${response.statusCode}")
                }
            }
    }

}

 

요청/응답 로깅

class LoggingInterceptor : ClientHttpRequestInterceptor {
    override fun intercept(
        request: HttpRequest,
        body: ByteArray,
        execution: ClientHttpRequestExecution
    ): ClientHttpResponse {
        logger.info("=== Request ===")
        logger.info("URI: {}", request.uri)
        logger.info("Method: {}", request.method)
        logger.info("Headers: {}", request.headers)

        val response = execution.execute(request, body)

        logger.info("=== Response ===")
        logger.info("Status: {}", response.statusCode)

        return response
    }
}

// 적용
.requestInterceptor(LoggingInterceptor())

OpenFeign의 Logger.Level 설정과 비교하여, RestClient는 더 유연한 로깅 방식을 제공한다

타임아웃 설정

@Configuration
class RestClientConfig(
    @Value("\${rest.client.base-url}")
    private val baseUrl: String
) {
    @Bean
    fun defaultRestClientBuilder(): RestClient.Builder {
        return RestClient
            .builder()
            .baseUrl(baseUrl)
            .defaultHeaders { headers ->
                headers.setBearerAuth(generateToken())
            }
            // 타임아웃 설정
            .requestFactory {
                HttpComponentsClientHttpRequestFactory().apply {
                    setConnectTimeout(Duration.ofSeconds(5))
                    setConnectionRequestTimeout(Duration.ofSeconds(5))
                }
            }
    }
}

OpenFeign의 타임아웃 설정과 달리, RestClient는 HttpComponentsClientHttpRequestFactory를 통해 더 섬세한 타임아웃 제어가 가능하다:

  1. setConnectTimeout: 서버와의 연결을 맺는데 걸리는 최대 시간
    • TCP 연결 수립 시간 제한
    • 네트워크 문제나 서버 응답 지연 시 빠른 실패 처리 가능
  2. setConnectionRequestTimeout: 커넥션 풀에서 커넥션을 가져오는데 걸리는 최대 시간
    • 커넥션 풀이 고갈되었을 때의 대기 시간 제한
    • 서비스 과부하 상황에서의 타임아웃 처리
  3. 선택적으로 ReadTimeout 설정도 가능하다
.requestFactory {
    HttpComponentsClientHttpRequestFactory().apply {
        setConnectTimeout(Duration.ofSeconds(5))
        setConnectionRequestTimeout(Duration.ofSeconds(5))
        setReadTimeout(Duration.ofSeconds(10))  // 데이터 읽기 타임아웃
    }
}

 

Conclusion

Spring Cloud OpenFeign에 비교해 HTTP Interface의 경우 다음과 같은 이점이 있다.

 

  • Spring 네이티브 통합
    • Spring 6의 기본 기능을 활용하여 더 나은 Spring 생태계와 통합된다
    • 별도의 외부 라이브러리 의존성이 감소한다
    • Spring의 지속적인 개선사항이 자동으로 적용된다
  • openFeign과 유사한 선언적 프로그래밍 방식을 유지한다
    • OpenFeign과 유사한 선언적 프로그래밍 방식을 유지한다
    • 기존 코드 구조를 크게 변경하지 않고도 전환할 수 있다
  • 안정성
    • Spring의 최신 HTTP 클라이언트 스택을 활용할 수 있다
    • Spring의 지속적인 개선사항이 자동으로 적용된다

 

예시에 사용한 코드는 다음 링크에 있다.
https://github.com/waterfogSW/HttpInterfaceExample

Kafka Streams의 stateful 프로세싱은 다른 시간에 도착하는 관련 이벤트들을 그룹화하여 관리하고 저장할 수 있는 기능을 제공한다. State Store는 중간 상태를 로컬 또는 원격으로 저장하는 구조로, Kafka 체인지 로그 토픽을 기반으로 한 내결함성을 갖추고 있다. 이러한 구조 덕분에 Kafka Streams 애플리케이션은 인스턴스 간 데이터 일관성을 유지하면서도 효율적인 스케일 아웃을 달성할 수 있다. 이번 글에서는 Kafka Streams의 state store 분산 방식, 데이터 일관성 유지 방법, 그리고 로컬 및 원격 상태 조회 방식을 중점으로 다룬다.

 

1. State Store의 역할과 필요성

Kafka Streams에서 State Store는 스트림 처리 중 발생하는 중간 상태를 로컬에 저장한다. 예를 들어 사용자 행동 데이터를 처리할 때, 각 사용자의 총 구매 횟수를 집계한다고 가정해보자. 각 이벤트가 들어올 때마다 사용자의 기존 구매 횟수를 가져와 업데이트해야 하는데, 이 중간 데이터는 빠르게 접근할 수 있는 로컬 저장소에 저장하는 것이 효율적이다.

예시 : 쇼핑 애플리케이션의 구매 집계 시스템

 

  • 사용자의 구매 이벤트가 Kafka 토픽으로 수신될 때마다, Kafka Streams 애플리케이션은 구매 이벤트를 기반으로 사용자의 총 구매 횟수를 누적하여 업데이트한다.
  • 각 Kafka Streams 인스턴스는 사용자 ID를 기준으로 파티셔닝된 데이터를 관리하며, 해당 파티션에 속한 사용자의 구매 횟수를 state store에 저장하여 빠르게 접근할 수 있다.

 

2. 스케일아웃 상황에서 State Store와 Changelog Topic의 동작 방식

Kafka Streams의 State Store와 Changelog Topic은 인스턴스 간 분산 처리와 데이터 일관성을 지원하기 위해 다음과 같은 방식으로 동작한다.

2-1. 파티셔닝과 할당

Kafka Streams는 데이터를 파티셔닝하여 관리하며, 각 인스턴스는 특정 파티션을 처리하는 방식으로 상태를 유지한다. 스케일아웃 시 인스턴스가 추가되면, 기존 인스턴스에 할당된 파티션을 새 인스턴스에 자동으로 재할당하여 데이터 처리가 더욱 분산된다. 각 인스턴스는 할당된 파티션에 대한 독립적인 State Store를 유지하게 된다.

2-2. State Store 데이터 복제와 일관성 유지

각 인스턴스의 State Store는 Changelog Topic에 변경 사항을 기록하여 다른 인스턴스에서도 해당 상태를 복구할 수 있도록 한다. 인스턴스가 확장되어 파티션이 새로운 인스턴스에 재배치될 때, 새로운 인스턴스는 Changelog Topic을 구독하여 필요한 데이터를 로드하고, 이후 업데이트 사항을 받아 데이터 일관성을 유지한다.

2-3. 스케일아웃 과정에서의 성능 최적화와 복구 지연 완화

스케일아웃 중 State Store의 데이터 크기가 클 경우 복구에 시간이 지연될 수 있다. Kafka Streams는 이를 완화하기 위해 상태 스냅샷(State Snapshot)상태 저장소 압축(State Store Compaction) 기법을 제공한다. 상태 스냅샷을 통해 특정 시점의 데이터를 저장하여 복구 시 전체 데이터를 다시 적용할 필요 없이 빠르게 복구할 수 있다. 압축 기법을 통해 불필요한 데이터를 정리하여 복구 시간을 단축할 수 있다.

2-4. 분산 환경에서의 데이터 일관성 보장

Kafka Streams는 분산 환경에서도 exactly-once 또는 at-least-once 처리 방식을 통해 데이터 일관성을 보장한다. State Store는 트랜잭션 방식으로 이벤트를 처리하며, 이벤트 처리 후 상태를 Changelog Topic에 커밋하여 장애 시 중복 처리를 방지한다. 이를 통해 스케일아웃 상황에서도 데이터 일관성을 유지할 수 있다.


예시 상황

쇼핑 애플리케이션에서 사용자 구매 이벤트를 처리하는 Kafka Streams 애플리케이션을 가정한다.

  1. 초기 인스턴스 구성
    사용자의 구매 횟수를 집계하는 애플리케이션이 초기에는 두 개의 인스턴스를 사용한다. 각 인스턴스는 사용자 ID를 기준으로 파티셔닝된 데이터를 할당받아 각 사용자의 총 구매 횟수를 State Store에 기록한다.
  2. 스케일아웃으로 인스턴스 추가
    트래픽이 증가하면서 새로운 인스턴스를 추가하여 스케일아웃한다. Kafka Streams는 기존 두 인스턴스에 할당된 파티션 중 일부를 새 인스턴스에 자동으로 재할당한다. 새 인스턴스는 할당된 파티션의 Changelog Topic을 구독하여 해당 파티션의 모든 상태 데이터를 로드하고, 이후 들어오는 구매 이벤트를 처리한다.

 

3. State Store 데이터 일관성 유지 및 장애 복구

Kafka Streams 애플리케이션은 상태가 필요한 스트림 처리 작업을 수행할 때 State Store를 사용한다. 이때 State Store에 저장된 데이터는 로컬에 유지되기 때문에, 특정 인스턴스에 장애가 발생하면 해당 인스턴스의 상태 정보를 잃어버릴 위험이 있다. Changelog Topic은 이러한 문제를 해결하기 위해 상태 변경 사항을 Kafka의 특별한 토픽에 기록해 두며, 다른 인스턴스가 해당 파티션을 맡게 되었을 때 이 기록을 바탕으로 State Store를 복구할 수 있게 해준다.

예시 :  사용자 클릭 집계에서 장애 상황 발생 시 데이터 복구

  • 웹사이트에서 발생하는 사용자 클릭 이벤트를 집계하는 Kafka Streams 애플리케이션을 가정한다. 각 인스턴스는 특정 파티션의 데이터를 처리하며, 클릭 수 집계를 state store에 기록한다.
  • 인스턴스 장애 시, 다른 인스턴스가 해당 파티션을 할당받아 state store를 복구하는데, 이때 필요한 데이터는 Changelog Topic에서 불러와 복구한다. 이를 통해 중단 없이 데이터 일관성을 유지하면서 집계를 이어갈 수 있다.

 

Changelog Topic은 Kafka의 파티션 구조를 활용해 각 State Store의 변경 사항을 토픽 파티션에 저장한다. Kafka Streams 애플리케이션이 여러 파티션을 사용하는 경우, 각 파티션별로 Changelog Topic도 같은 수의 파티션을 가지게 된다. 이를 통해 Changelog Topic의 데이터는 애플리케이션 파티션과 동일한 구조로 분산 저장되어 있어 복구할 때 효율적이다.

  • 데이터 일관성 유지: State Store는 기본적으로 RocksDB 같은 로컬 데이터베이스에 저장된다. State Store가 변경될 때마다 로그 커밋이 Changelog Topic으로 전송되어 데이터 일관성이 보장된다. Kafka Streams의 at-least-once 또는 exactly-once 처리 보장에 따라, 중복 데이터가 발생할 수 있는 상황에서도 Changelog Topic을 통해 마지막으로 커밋된 상태를 기준으로 일관성을 유지할 수 있다.

 

Kafka Streams의 인스턴스가 장애로 인해 중단될 때, Kafka는 해당 인스턴스가 맡고 있던 파티션을 다른 인스턴스에 할당한다. 이 새로운 인스턴스는 Changelog Topic에 저장된 데이터를 기반으로 State Store를 복구한다.

  • 복구 과정
    • 새 인스턴스가 할당된 파티션의 Changelog Topic을 구독하여 모든 변경 기록을 가져온다.
    • 각 이벤트를 순차적으로 적용해 State Store를 다시 생성한다.
  • 예를 들어, A 인스턴스가 클릭 수를 관리하던 도중 장애가 발생했다면, B 인스턴스가 해당 파티션을 할당받고, Changelog Topic에서 A 인스턴스의 상태 변경 사항을 받아 State Store를 복구해 클릭 수 집계를 이어간다.

 

4. Changelog Topic의 장점과 고려 사항

장점:

  • 내결함성(Fault Tolerance): Changelog Topic은 State Store의 상태를 Kafka 클러스터에 백업하는 역할을 하기 때문에, 로컬 데이터 손실이 발생해도 안전하게 복구할 수 있다.
  • 데이터 일관성: Kafka Streams는 장애 시 재처리를 통해 일관성을 유지하기 위해 Changelog Topic을 사용하며, 정확히 한 번(exactly-once) 또는 적어도 한 번(at-least-once) 처리 옵션을 설정할 수 있다.

고려 사항:

  • 디스크 사용량: Changelog Topic은 모든 State Store 변경 사항을 저장하므로, 대규모 데이터를 다루는 애플리케이션에서는 저장 용량이 커질 수 있다. 필요 시 TTL(Time-to-Live) 설정을 통해 오래된 데이터를 삭제하거나, 압축을 통해 용량을 줄일 수 있다.
  • 복구 지연 시간: 대량의 데이터를 가진 State Store를 복구할 때 지연이 발생할 수 있다. 복구 시간이 중요한 경우, State Store를 주기적으로 백업하거나 상태 스냅샷 기능을 사용해 복구 시간을 단축하는 것이 도움이 된다.

 

다음 코드의 실행 결과를 한번 예측해보자.

fun main() {
    val a: Int? = 128
    val b: Int? = 128
    println(a === b)

    val c: Int? = 1
    val d: Int? = 1
    println(c === d)
}

이 코드의 실행 결과는 false, true이다. 이러한 결과가 나오는 원인은 Kotlin(그리고 그 기반이 되는 Java)의 정수 객체 캐싱 메커니즘에 있다. 본 글에서는 이 현상의 원인과 그 배경에 있는 객체 캐싱 메커니즘에 대해 상세히 분석한다.

 

1. Java의 Autoboxing과 객체 캐싱

Java에서는 기본 타입(primitive type)과 그에 대응하는 래퍼 클래스(wrapper class)가 존재한다. 예를 들어, int에 대응하는 래퍼 클래스는 Integer이다. Java 5부터 도입된 Autoboxing 기능은 기본 타입과 래퍼 클래스 간의 자동 변환을 지원한다.

Integer a = 100; // Autoboxing: int -> Integer
int b = a; // Unboxing: Integer -> int

Java는 성능 최적화를 위해 특정 범위의 정수값에 대해 객체 캐싱을 수행한다. 기본적으로 -128부터 127까지의 정수값에 대해서는 미리 객체를 생성하여 캐시에 저장한다.

다음 Java 코드를 통해 이 동작을 확인할 수 있다:

public class IntegerCacheTest {
    public static void main(String[] args) {
        Integer a = 127;
        Integer b = 127;
        System.out.println(a == b); // true

        Integer c = 128;
        Integer d = 128;
        System.out.println(c == d); // false

        int e = 128;
        int f = 128;
        System.out.println(e == f); // true
    }
}

이 코드에서 a == b는 true를 반환하지만, c == d는 false를 반환한다. 이는 127이 캐시 범위 내에 있어 같은 객체를 참조하지만, 128은 캐시 범위를 벗어나 새로운 객체가 생성되기 때문이다. 반면 e == f는 기본 타입 int의 비교이므로 true를 반환한다.

 

2. Kotlin에서의 정수 비교

Kotlin은 Java의 이러한 특성을 기반으로 하면서도, 몇 가지 추가적인 특징을 제공한다. Kotlin에서 === 연산자는 참조 동등성을 비교한다. 즉, 두 객체가 메모리상에서 같은 객체인지를 확인한다.

앞서 제시한 Kotlin 코드의 결과를 분석하면 다음과 같다:

  1. 1은 -128에서 127 사이의 값이므로 캐시된 객체를 사용한다. 따라서 c와 d는 같은 객체를 참조하게 되어 true가 반환된다.
  2. 128은 캐시 범위를 벗어나는 값이므로 새로운 객체가 생성된다. 따라서 a와 b는 서로 다른 객체를 참조하게 되어 false가 반환된다.

 

3. Java의 래퍼 클래스와 캐싱

Java에서 이러한 캐싱 메커니즘은 여러 래퍼 클래스에 적용된다:

  • Boolean: true와 false
  • Byte: 모든 값 (-128 to 127)
  • Short: -128 to 127
  • Integer: -128 to 127 (기본값, 변경 가능)
  • Long: -128 to 127
  • Character: 0 to 127

다음 Java 코드를 통해 다양한 래퍼 클래스의 캐싱 동작을 확인할 수 있다:

public class WrapperCacheTest {
    public static void main(String[] args) {
        Boolean bool1 = true;
        Boolean bool2 = true;
        System.out.println("Boolean: " + (bool1 == bool2)); // true

        Byte byte1 = 127;
        Byte byte2 = 127;
        System.out.println("Byte: " + (byte1 == byte2)); // true

        Short short1 = 127;
        Short short2 = 127;
        System.out.println("Short: " + (short1 == short2)); // true

        Integer int1 = 127;
        Integer int2 = 127;
        System.out.println("Integer: " + (int1 == int2)); // true

        Long long1 = 127L;
        Long long2 = 127L;
        System.out.println("Long: " + (long1 == long2)); // true

        Character char1 = 127;
        Character char2 = 127;
        System.out.println("Character: " + (char1 == char2)); // true
    }
}

 

4. -XX:AutoBoxCacheMax 옵션

Java (그리고 Kotlin)에서는 -XX:AutoBoxCacheMax JVM 옵션을 통해 Integer 캐시의 최대값을 조정할 수 있다. 기본값은 127이지만, 이를 변경하여 더 큰 범위의 정수에 대해서도 캐싱을 적용할 수 있다.

java -XX:AutoBoxCacheMax=1000 YourProgram

이 옵션을 사용하면 지정된 값까지의 Integer 객체가 캐시되어, 해당 범위 내의 정수 비교 시 == 연산자 (Java) 또는 === 연산자 (Kotlin)가 true를 반환하게 된다.

다음은 이 옵션을 적용한 Java 프로그램의 예시이다:

public class AutoBoxCacheMaxTest {
    public static void main(String[] args) {
        Integer a = 1000;
        Integer b = 1000;
        System.out.println(a == b); // true (if -XX:AutoBoxCacheMax=1000 is set)
    }
}

 

5. Kotlin의 특징

Kotlin은 Java의 이러한 특성을 기반으로 하면서도, 몇 가지 추가적인 특징을 제공한다:

  1. == 연산자: Kotlin에서 ==는 구조적 동등성을 비교한다. 이는 내부적으로 .equals() 메소드를 호출하는 것과 동일하다.
  2. === 연산자: 참조 동등성을 비교한다. Java의 ==와 유사한 역할을 한다.

다음 Kotlin 코드를 통해 이러한 특징을 확인할 수 있다:

fun main() {
    val a: Int? = 128
    val b: Int? = 128
    println(a == b)  // true (구조적 동등성)
    println(a === b) // false (참조 동등성)

    val c: Int? = 127
    val d: Int? = 127
    println(c == d)  // true
    println(c === d) // true (캐시된 객체)
}

 

6. 주의사항

  1. 이 캐싱 메커니즘은 성능 최적화를 위한 기능이므로, 코드의 정확성을 이 동작에 의존해서는 안 된다.
  2. Java에서 값 비교를 위해서는 .equals() 메소드를 사용하는 것이 안전하다. Kotlin에서는 == 연산자를 사용하면 된다.
  3. 이 캐싱 메커니즘은 Integer/Int 외의 다른 숫자 타입(Byte, Short, Long 등)에도 적용되지만, 범위가 다를 수 있다.
  4. Float와 Double은 캐싱되지 않는다.

 

7. 결론

Java와 Kotlin에서의 정수 비교는 표면적으로는 단순해 보이지만, 내부적으로 복잡한 메커니즘이 작동하고 있다. 객체 캐싱은 성능 최적화를 위한 중요한 기능이지만, 개발자는 이에 의존하기보다는 항상 명확하고 안전한 비교 방법을 사용해야 한다.

이러한 내부 동작을 이해함으로써, 더 효율적이고 버그 없는 코드를 작성할 수 있다. 또한, 이는 Java와 Kotlin의 내부 동작 방식에 대한 깊이 있는 이해를 제공하여, 더 나은 프로그래밍 실력 향상에 기여할 수 있다.

마지막으로, 이러한 세부사항을 알고 있는 것은 중요하지만, 일반적인 애플리케이션 개발에서는 == (Kotlin) 또는 .equals() (Java)를 사용하여 값을 비교하는 것이 가장 안전하고 명확한 방법임을 인지해야 한다.

WEAVE라는 대학생 미팅서비스를 개발하고 출시하기 까지 과정에 대한 회고글 입니다.

퇴사후 공백기

퇴사후 잠깐 공백기가 생기게 되었는데, 뭐라도 만들고 싶어서 미칠것 같았다. 그래서 사이드 프로젝트 개발을 계속 시도해 왔는데, 금방 흥미가 떨어져 좌절되곤 했다.

여태까지 중간에 그만두었거나, 더발전시키지 못한 사이드 프로젝트들을 되돌아 보면 다음과 같은 이유들이 있을것 같다.

  • 페인포인트에 공감하지 못했다.
  • 혼자하느라 금방 흥미가 떨어졌다.

이런 와중에 이전에 부트캠프에서 함께했던 도진님이 같이 사이드 프로젝트를 해보지 않겠냐고 연락 주셔서 무슨 서비스인지 듣지도 않고 바로 하겠다고 했다.

MVP 개발 1월

대학생 미팅 서비스를 개발하기로 했다. 디자이너 한분, AOS 개발자 한분, 서버 두명이 모여 개발을 진행했다. 회의는 월, 수, 금 짧게, 일요일 길게 진행했는데, 자주하는만큼 팀원들의 진행사항을 쉽게 파악할 수 있었고, 자주하는 만큼 일상속에 사이드 프로젝트가 잘 녹아들어갔다.

이전에 진행했던 사이드 프로젝트들의 경우 디자인이나 기획이 주어진대로 그대로 따라가는 형태가 대부분이었는데, 왜 이 기능이 필요하지 라는 의문이 드는 케이스가 있어서 아쉬웠었다. WEAVE에서는 기획단계부터 참여했던 만큼, MVP기능을 선정하고 디자인 하는 과정에서도 의견을 많이 낼 수 있어서 동기부여가 잘됐다.

MVP 개발 2월

기술스택과 아키텍처를 협의하고 본격적으로 개발에 들어갔다. 기존에 사이드 프로젝트를 진행하면서 만들어 두었던 코드 템플릿을 토대로 서버 개발을 진행했다. 도진님은 회사일 때문에 바쁘셔서 많이 참여는 못하셨지만 리뷰는 빠르고 꼼꼼하게 남겨주셔서 놓친부분들을 쉽게 파악할 수 있었다.

IOS개발자 두분이 중간에 합류해 주셨는데, 싱크를 맞추는 과정에서 내가 API명세에서 놓쳤던 부분들을 많이 알 수 있었다. 에러코드 관리와 명세가 아쉬웠고, 다양한 케이스들에 대한 테스트베드를 Swagger상에서 제공할 수 있다는 점을 처음 알았다.

JIRA를 사용하는데 있어서 테스크가 유스케이스 별로 관리되는것이 아니라 각자 파트 별로 단순하게 관리되는 점이 아쉬웠다. 어떤 작업들을 진행하고 있는지 회의로 한번더 파악해야 했었고, JIRA는 유명무실하게 사용되고 있었다. 사실 회의 주기가 짧고 비동기 소통이 꽤나 잘되고 있어서 크게 문제가 되진 않았는데, JIRA의 효용성에 대한 의문이 있었다.

MVP 개발 3월

본격적으로 출시를 앞두면서 MVP기능 개발은 대부분 완료가 되었다. AOS심사가 들어갔고, 개발서버와 별개로 배포서버 인프라를 구축했다. 대학생 서버 개발자 한분이 합류하셨다.

Terraform과 Terraform cloud를 통해 인프라를 배포했는데, 충분히 팀원들에게 Terraform의 활용법에 대해 공유하지 못한부분이 아쉬웠다. ELB가 unmanaged resource로 배포되었다.

AWS비용이 발생하기 시작했는데, 6~70달러를 웃돌았다. 최대한 프리티어로 맞추었지만 추가적인 EC2비용과 ELB로 인한 비용이 컸다. 해외 IP를 차단하기 위해 WAF를 사용했다.

3월 말에는 AOS 심사가 지연되었고, IOS도 심사에 들어갔다.

MVP 개발 4월

앱스토어, 플레이스토어에 모두 출시가 되었다. IOS 한분과 이전에 합류했던 서버팀원 한분이 나가셔서 친한 서버 개발자 지인을 데려왔다.

홍보를 했지만, 효과는 미미했고, 유저들을 확보하기 어려웠다.

팀원들과 몇가지 고민을 했다.

  • 홍보가 미약했다. (인스타를 활용 안함)
  • 유사 제품들과 차별점이 뭔지 모르겠다.
  • 유저를 잘 모른다 (대학생 서비스인데 팀에 현직 대학생이 한명밖에 없음.

고민끝에 잠깐 서비스 개발을 중단하고 마침 디자이너 분도 학업때문에 이탈하게 되셔서 새로운 팀원을 모집하는겸 시간을 갖기로 했다.

Liked : 좋았던 점은 무엇인가?

짧은 회의주기

사이드 프로젝트는 동기부여가 가장 중요하다고 생각한다. 대부분 각자 학업, 현업이 있는 만큼 짬내서 작업을 해야 하는데, 동기부여 없이는 아주 힘들다.

처음에는 성대한 동기로 시작하지만 중간 중간 위기를 겪는다. 짧은 회의 주기(월,수,금,일)는 작은 성취를 팀원들과 공유하면서 지속적인 동기부여를 주기에 아주 좋은 방법이라고 생각한다.

코드 리뷰

나는 필요할때는 코드리뷰를 매우 적극적으로 하는 편이다. 몇몇 PR은 하나에 수십개가 넘는 코멘트를 주고받기도 했는데 도진님도 비슷한 성향이셔서 잘 맞았다. 주관을 개진할때는 확신을 가지고 하되, 상대방이 더 합리적이라면 누그러뜨리고 수용한다. 상대방에 대한 충분한 신뢰가 있었기에 가능했다고 생각한다.

Lacked : 아쉬웠던 점, 부족한 점은 무엇인가?

API 명세
에러코드, 다양한 케이스에 대한 테스트 베드를 Swagger상에 명세하지 못했다. Discord 서버 채널로 클라이언트 분들의 관련 문의가 들어올때마다 죄송한 마음이 들었다.

Swagger에 여러 케이스별로 테스트 베드를 만들어 내 명세할 수 있다는점을 나중에 알았다. 클라이언트 분들이 추가적인 문의 없이 개발할 수 있을 만큼 충분한 명세서를 만들자

빈약한 DDD

도메인 엔터티와 JPA 엔터티를 분리한 만큼 도메인에서 최대한 로직을 처리할 수 있게 설계할 수도 있었다. 초창기에는 그러지 못했고, Aggregate Root말고 모든 객체에 대해 Repository를 만들어 졌는데, 복잡도가 많이 높아졌고, Application 레이어로 도메인 로직들이 새어나갔다. 리팩토링을 통해 바로잡았는데, 다시 설계한다면 비슷한 우를 범하진 않을것 같다.

이외에도 기술적으로는 아쉬운점들은 많았는데, 사소하거나 방향을 조금만 바꾸면 해결할 수 있는것들이어서 생략해도 될것 같다.

도메인 명세

도메인 지식을 명세하는 문서가 있긴 했는데, 유명무실하게 관리되었다. 모두가 와이어 프레임만 보고 기능을 개발하다 보니 제때 수정사항을 문서에 반영하지 못한 부분도 있었고 와이어 프레임에 나타나지 않는 수정사항들을 파악하는데 추가적인 소요가 발생했다. 도메인 문서는 백엔드가 챙겨야할 중요한 부분중 하나라는 생각이다

Longed for : 앞으로 바라는 것은 무엇인가?

서비스 개발을 중단한 기간동안 여러 스타트업들의 아이디어 검증 방법, 성공사례, 실패사례 등을 살펴보았다. 공통적으로 아이디어 검증을 충분히 진행한 이후에야 개발을 진행한 케이스들이 많았다.

하지만 사이드 프로젝트들의 경우 자신만의 명확한 페인포인트를 해결하는데 집중한 케이스들이 많았다. 덕분에 규모가 크지 않았고, 타겟층이 확실한 만큼 유저도 수월하게 확보한 것 같았다.

위의 두가지 포인트를 잘 논의하면서 어떤 방법이 좋은 방향인지 논의해 봐야 할것 같다. 잠깐 서비스는 중단했지만, 여태 진행했던 사이드 프로젝트들 중 가장 완성도 높은 프로젝트였고, 좋은 팀원분들과 함께 했기에 괜찮은 결과물을 만들 수 있었다.

이제는 실제 유저를 확보하고 지속적으로 운영할수 있을만큼의 수익도 발생 시킬 수 있는 서비스 까지 가기 위해 고민해보고 방향을 더 논의해봐야 할 것 같다. 정말 필요한 기능만 린하게 개발할수 있게 충분히 아이디어를 검증하고, 이전에 아쉬웠던 점들을 보완해 좀더 견고하게 설계 할 수 있어야한다.

Apache Lucene에서 사용된 Incremental indexing기법의 기반이 된 Doug Cutting의 Optimizations for dynamic inverted index maintenance. 논문을 읽고 정리한 글입니다.
https://dl.acm.org/doi/10.1145/96749.98245

Apache Lucene

Apache Lucene은 오픈소스 정보 검색 라이브러리로, 트위터, 위키피디아, 링크드인, 넷플릭스, 아이튠즈 스토어, 세일즈포스 등 구글과 페이스북을 제외한 대부분의 검색 엔진에서 사용되며, 솔라(Solr), 엘라스틱 서치(ElasticSearch) 등의 검색엔진이 Apache Lucene을 기반으로 합니다.

Apache Lucene은 문서를 빠르게 색인하고 검색할 수 있는 기능을 제공하는데, 특히 증분식 색인(incremental indexing) 기법을 통해 문서가 추가될 때마다 효율적으로 색인을 갱신하는 방식을 사용하고 있습니다.

여기서 증분식 색인(incremental indexing) 기법은 기존 문서 집합에 새로운 문서가 추가될 때, 전체 문서를 다시 색인하지 않고 추가된 문서만 색인하는 방식입니다.

Lucene에서는 이를 위해 병합 색인(merge indexing) 기법을 사용합니다. 주요 특징은 다음과 같습니다:

  1. 새 문서가 추가되면, 해당 문서만을 위한 작은 색인을 따로 생성함
  2. 일정 개수의 작은 색인들이 쌓이면, 이들을 하나의 큰 색인으로 병합함
  3. 병합 과정은 백그라운드에서 이루어지므로 검색 성능에 영향을 주지 않음
  4. 이 과정을 반복하면서 전체 문서 집합의 색인을 증분식으로 갱신해 나감
  5. 검색 시에는 모든 색인을 동시에 검색하므로, 안정적인 검색 성능을 보장함

증분식 색인 기법을 통해 대용량 문서 집합에 대해서도 빠르게 색인을 갱신하고 일관된 검색 결과를 얻을 수 있습니다. 이는 Lucene이 실시간 검색 엔진으로 널리 사용되는 데 크게 기여한 핵심 기술입니다.

Doug Cutting이 개발한 루씬에서 사용된 증분식 색인 기법은 그의 1990년대 후반 연구에 기반합니다. 이에 대한 자세한 내용은 다음 논문에서 확인할 수 있습니다.

Cutting, D., & Pedersen, J. (1989). Optimizations for dynamic inverted index maintenance. In Proceedings of the 13th annual international ACM SIGIR conference on Research and development in information retrieval (pp. 405-411).

이 논문에서는 동적 색인 유지보수를 위한 최적화 기법에 대해 설명하고 있습니다. 특히 증분식 색인 구축 과정에서 발생하는 오버헤드를 최소화하기 위한 방법으로, 작은 색인들을 주기적으로 병합하는 기법을 제안하였습니다. 이는 Lucene의 증분식 색인 기법의 핵심 아이디어가 되었습니다.

논문 읽기

배경

많은 문서 검색 시스템에서는 사용자가 자유롭게 키워드를 입력하면 그에 맞는 문서를 찾아주는 자유 텍스트 검색(free-text search) 기능을 제공합니다. 이때 시스템 내부적으로는 역색인을 사용하여 검색을 수행하게 됩니다.

그런데 이런 시스템에서 다루는 문서들의 집합, 즉 코퍼스(corpus)는 끊임없이 변화합니다. 새로운 문서들이 추가되기도 하고, 기존 문서들이 삭제 또는 수정되기도 합니다. 이렇게 빠르게 변화하는 코퍼스에 대해 자유 텍스트 검색 서비스를 제공하려면, 역색인을 실시간으로 갱신할 수 있어야 합니다.

가령 어떤 문서가 새로 추가되었다면, 해당 문서에 출현한 단어들을 곧바로 역색인에 반영해야 그 문서도 검색될 수 있을 것입니다. 만약 색인 갱신이 지연된다면 사용자는 검색 결과에서 누락을 경험하게 될 것입니다.

따라서 빠르게 변화하는 코퍼스를 대상으로 자유 텍스트 검색을 제공하기 위해서는, 역색인을 동적으로 갱신하는 것이 기본적으로 요구됩니다. 단순히 주기적으로 색인을 다시 만드는 정적 방식으로는 한계가 있기 때문입니다.

이 논문은 바로 이러한 배경에서, B-tree를 기반으로 역색인을 효과적으로 동적 갱신하는 방안을 연구한 것입니다. 제안된 여러 최적화 기법들은 역색인 검색 시스템의 성능과 확장성을 높이는 데 기여할 수 있습니다.

역색인(Inverted Indices)

역색인은 기본적으로 단어(word)를 키로 하고, 그 단어가 출현한 문서들의 식별자(document ID)를 값으로 가지는 일종의 맵핑 테이블입니다. 논문에서는 이때 값에 해당하는 문서 식별자들의 집합을 포스팅(posting)이라고 지칭합니다.

예를 들어 "word1"이라는 단어가 문서1, 문서2, 문서3에 출현했다면, 이에 대한 역색인 엔트리는 다음과 같은 형태가 될 것입니다.

"word1" -> {1, 2, 3}

여기서 {1, 2, 3}이 "word1"의 포스팅이 됩니다.

이러한 역색인을 실제로 구현할 때는 다양한 성능 기준을 고려해야 합니다. 논문에서는 특히 다음 다섯 가지 기준을 강조하고 있습니다.

  1. 색인 구축 속도 (index build speed)
  2. 색인 검색 속도 (access speed)
  3. 색인의 크기 (index size)
  4. 동적 갱신의 용이성 (dynamics)
  5. 확장성 (scalability)

한편 실제 포스팅에는 문서 식별자 외에도 추가적인 정보가 포함될 수 있습니다. 예를들어 단어가 문서 내에서 출현한 위치(offset)라든가, 출현 빈도(frequency) 같은 것이 될 수 있습니다. 다만 이는 역색인을 어떤 검색 알고리즘에 활용하느냐에 따라 달라질 수 있습니다.

역색인에서 가장 빈번하게 수행되는 연산은 특정 단어가 주어졌을 때 그 단어가 포함된 문서들을 찾아내는 것입니다. 이를 위해서는 단어를 키로 하여 색인에 접근할 수 있어야 합니다. 따라서 역색인은 단어들에 대해 정렬되어 있거나, 단어를 키로 하는 해시 테이블 형태로 구성되는 것이 일반적입니다.

이러한 역색인을 실제로 구현할 때는 다양한 요인들을 고려해야 합니다. 먼저 새로운 문서가 추가되었을 때 이를 색인에 반영하는 데 걸리는 시간, 즉 블록 단위 갱신 속도가 중요합니다. 새 문서의 색인 생성이 너무 오래 걸리면 검색 시스템의 실시간성이 떨어질 수 있기 때문입니다.

또한 사용자가 특정 단어를 검색했을 때 그 결과를 가져오는 데 걸리는 시간, 즉 접근 속도도 매우 중요한 성능 지표입니다. 사용자는 검색 결과를 빨리 받아보기를 원하므로, 색인 구조는 빠른 검색을 뒷받침할 수 있어야 합니다.

아울러 색인을 저장하는 데 드는 공간 비용, 즉 색인 크기도 무시할 수 없는 요소입니다. 색인이 지나치게 커지면 저장 공간 뿐 아니라 입출력 비용도 커지게 되므로, 적절한 압축이나 최적화를 통해 색인 크기를 적정 수준으로 유지할 필요가 있습니다.

한편 문서 집합이 동적으로 변한다는 점도 역색인 구현에 많은 영향을 미칩니다. 새 문서의 추가나 기존 문서의 삭제, 수정이 빈번히 일어난다면 이를 색인에 반영하는 증분 갱신 작업이 용이해야 합니다. 그렇지 않으면 색인 유지에 너무 큰 비용이 들 수 있습니다.

특히 문서 삭제의 경우, 해당 문서에 대한 모든 색인 항목을 빠짐없이 제거해야 합니다. 그런데 문서 내용을 알 수 없다면 색인 전체를 뒤져가며 관련 항목을 찾아내야 할 수도 있습니다. 따라서 가능하다면 문서 식별자와 문서 내용을 함께 관리하는 것이 색인 관리에 유리합니다.

문서 집합의 크기가 커짐에 따라 이러한 요소들이 어떻게 변화하는지, 즉 확장성도 역시 중요하게 고려해야 할 부분입니다. 문서가 많아질수록 색인의 크기가 커지고 색인 연산의 비용도 증가할 수밖에 없는데, 이를 얼마나 효과적으로 감당할 수 있느냐가 확장성을 좌우하기 때문입니다.

역색인을 설계하고 구현할 때는 이처럼 다양한 요인들을 복합적으로 고려해야 합니다. 단순히 검색 성능만을 최적화할 것이 아니라, 갱신 용이성, 저장 비용, 확장성 등을 두루 살펴가며 절충안을 찾아내는 것이 중요합니다.

Naive B-tree의 단점

B-tree는 디스크와 같은 2차 저장소에 저장되는 균형 잡힌 트리 구조입니다. 각 노드는 디스크 페이지로 표현되며, 하나의 노드에는 여러 개의 엔트리(키-값 쌍)가 담길 수 있습니다. 이때 하나의 노드가 담을 수 있는 엔트리의 수를 B-tree의 분기 계수(branching factor)라고 합니다.

B-tree에서는 이 분기 계수를 b라고 할 때, 노드 간 균형을 맞추기 위한 알고리즘을 통해 삽입, 검색, 삭제 연산의 시간 복잡도가 O(log_b N)으로 보장됩니다. 여기서 N은 B-tree가 담고 있는 총 엔트리의 수입니다. 또한 B-tree의 높이, 즉 루트 노드에서 리프 노드까지의 거리를 'B-tree의 깊이(depth)'라고 하는데, 이는 log_b N과 같습니다.

B-tree에 저장된 엔트리는 키 순으로 정렬되어 있으므로, 순차 접근 시에는 전체 엔트리를 키 순으로 읽어올 수 있습니다. 이때의 시간 복잡도는 O(N)이며, 대략 N/b번의 디스크 접근이 필요합니다.

B-tree의 분기 계수 b는 보통 100 정도로 크게 잡습니다. 이렇게 하면 100만 개의 엔트리를 담은 B-tree의 깊이는 3 정도가 되므로, 최대 3번의 디스크 접근으로 원하는 엔트리에 도달할 수 있습니다.

만약 B-tree의 일부 노드를 메모리에 캐싱한다면 접근 속도를 더욱 높일 수 있습니다. 예를 들어 루트 노드를 항상 메모리에 들고 있으면 모든 연산에서의 디스크 접근 횟수를 1만큼 줄일 수 있는데, 이때 b개의 엔트리를 메모리에 추가로 저장하는 대신 디스크 접근을 1회 아낄 수 있게 되는 셈입니다.

이를 일반화하면, 메모리에 C개의 엔트리를 저장할 수 있다고 할 때 B-tree 연산에 필요한 디스크 접근 횟수는 대략 log_b N - log_b C가 됩니다 (N ≥ C 인 경우). 즉, 캐시 크기를 늘리면 디스크 접근 횟수를 줄일 수 있습니다.

나이브한 B-tree 기반 역색인 구현에서는 (단어, 위치) 쌍을 B-tree의 엔트리로 삼습니다. 단어 단위로 인접하게 배치하므로 특정 단어에 대한 포스팅 리스트를 얻으려면 B-tree를 차례로 읽으면서 관련 엔트리들을 뽑아내면 됩니다. 이때 b개 엔트리마다 한 번씩 디스크 접근이 일어나게 됩니다.

그러나 새로운 문서의 색인을 만들 때는, 그 안의 모든 단어에 대해 (단어, 위치) 엔트리를 B-tree에 삽입해야 합니다. n개의 단어를 새로 색인한다면 대략 n * (log_b N - log_b C)번의 디스크 읽기가 필요할 것입니다.

문서 내용을 알고 있다면 기존 문서에 대한 포스팅을 제거하는 것도 같은 비용으로 가능합니다. 그러나 문서 내용을 모른다면 B-tree 전체를 훑어가며 관련 엔트리를 찾아 지워야 하므로 비용이 크게 증가할 수 있습니다.

이상의 내용을 정리하면, B-tree는 역색인을 구현하는 데 있어 접근 속도, 갱신 지원, 확장성 측면에서 장점을 가지지만, 엔트리 중복으로 인한 공간 낭비와 문서 단위 색인 갱신의 비효율성이 단점으로 지적될 수 있습니다. 이러한 단점을 극복하기 위해 몇가지 최적화 방안을 도입할 수 있습니다.

Speed Optimization

앞서 살펴본 나이브한 B-tree 역색인에서는 새 문서의 색인을 추가할 때 문서 내 모든 단어에 대해 (단어, 위치) 엔트리를 역색인에 삽입해야 했습니다. 이때 각 삽입 연산마다 B-tree의 루트에서부터 리프까지 탐색해 내려가야 하므로 디스크 접근이 많이 발생하게 됩니다.

이를 개선하는 한 가지 방법은 B-tree의 상위 노드들을 메모리에 상주시켜 캐싱하는 것인데, 이렇게 하면 삽입 시 디스크 접근 횟수를 줄일 수 있습니다. 그런데 논문에서는 이것보다 더 나은 방법을 제시하고 있습니다.

바로 새 문서의 (단어, 위치) 정보들을 메모리 버퍼에 임시로 저장했다가, 버퍼가 가득 차면 이를 한꺼번에 정렬하여 B-tree에 병합하는 것입니다. 이 '병합을 통한 갱신(merge update)' 최적화를 적용하면 삽입 연산에 필요한 디스크 접근을 대폭 줄일 수 있습니다.

n개의 새로운 포스팅 정보를 메모리 버퍼에 받아두면, 실제로는 중복을 제외한 w개의 단어에 대한 색인 엔트리만 B-tree에 삽입하면 됩니다. 이때 각 단어의 엔트리는 인접한 위치에 저장될 가능성이 높으므로, 삽입을 위한 B-tree 탐색 비용이 줄어듭니다.

한편 단어 w에 대한 포스팅 리스트의 길이를 f(w)라 하고, 이것이 Zipf 분포를 따른다고 가정하면 f(w)와 w 사이의 관계식을 세울 수 있습니다. 이를 바탕으로 n, w, N(전체 색인의 크기) 사이의 관계를 분석해 볼 수 있는데, 결론적으로 병합 방식이 캐싱 방식보다 디스크 접근 횟수가 훨씬 적다는 것을 보일 수 있습니다.

수식의 자세한 전개 과정은 후술하겠습니다만, 논문에서 주어진 예시를 보면 병합 방식이 캐싱 방식 대비 약 1/5 수준의 디스크 접근만으로 갱신을 처리할 수 있음을 알 수 있습니다. 이는 버퍼링과 배치 처리가 역색인 갱신 성능 향상에 매우 효과적임을 보여주는 사례입니다.

[수식의 전개 과정]
먼저 캐싱 방식을 적용했을 때 n개의 새 포스팅 정보를 삽입하는데 필요한 디스크 읽기 횟수의 기댓값은 다음과 같이 주어집니다 (수식 2).

X = n(log_b N - log_b C)

여기서 b는 B-tree의 분기 계수, N은 전체 색인의 크기, C는 캐시의 크기입니다.

한편 병합 방식에서는 n개의 포스팅이 w개의 단어에 대한 포스팅 리스트로 변환된 후 B-tree에 삽입됩니다. 이때 필요한 디스크 읽기 횟수의 기댓값은 다음과 같습니다 (수식 3).

Y = w(log_b N - log_b w)

이제 두 방식의 비용을 비교하기 위해 C = n 이라 가정합시다. 즉, 캐시의 크기와 새로 삽입할 포스팅의 수가 같다고 볼 수 있습니다. 그러면 식 (2)와 (5)로부터 다음이 성립합니다.

X = z ln z (log_b N - log_b (z ln z)) (6) Y = z (log_b N - log_b z) (7)

여기서 z는 n개의 포스팅이 포함하는 고유 단어의 수입니다.

식 (7)을 약간 변형하면,

Y = X / ln z + z log_b(ln z) (8)

입니다.

만약 X > Y 라면 식 (8)에 의해,

X > X / ln z + z log_b(ln z)

이고, 이를 정리하면

z(ln z - 1) > log_b(ln z)

을 얻게 됩니다.

식 (6)을 이용해 위 부등식의 좌변을 X에 관한 식으로 바꾸고 이를 정리하면 다음과 같은 부등식을 얻게 됩니다.

(ln z - 1)(log_b N - log_b(z ln z)) > log_b(ln z)

이를 다시 정리하면,

ln^2 z log_b N > log_b z + log_b(ln z) ln z

입니다.

부등식의 양변에 지수함수를 취하면,

N > z (ln z)^(ln z / (ln z - 1)) (10)

이 됩니다. 즉, N과 z가 부등식 (10)을 만족할 때 X > Y 임을 알 수 있습니다.

그런데 z ln z = n = C 이고 식 (2)가 성립하려면 N ≥ C 이어야 하므로, N > z ln z 임을 알 수 있습니다. 충분히 큰 z에 대해 ln z / (ln z - 1)는 1에 가까우므로, 갱신 연산의 경우 대체로 X > Y 라 예상할 수 있습니다.

논문의 예시에서는 b = 100, N = 1,000,000, z ln z = 10,000 일 때 z ≈ 1,383 이고,

X ≈ 10,000, Y ≈ 1,977

로 계산됩니다. 이는 식 (11)에 의해서도 확인할 수 있는데,

X = ln z (Y - z log_b(ln z)) (11)

이므로 Y 값으로부터 계산된 X 값은 대략 10,000에 근접합니다.

따라서 병합 기반 갱신이 캐싱 기반 갱신보다 디스크 접근 비용 측면에서 훨씬 효율적임을 이론적으로 확인할 수 있습니다. 논문에서는 이처럼 Zipf 분포의 특성을 활용하여 현실적인 최적화 기법의 우수성을 입증하고 있습니다.

Space Optimization

가장 단순한 형태의 역색인에서는 색인의 각 엔트리가 (단어, 위치) 쌍으로 구성되므로, 같은 단어에 대한 엔트리들 사이에 단어 정보가 중복되는 문제가 있습니다. 이를 해결하기 위해 제안된 방법이 '그룹핑(grouping)'으로, 같은 단어에 대한 포스팅 정보를 (단어, 빈도, 위치 리스트) 형태로 묶는 것입니다.

이렇게 하면 엔트리 하나당 단어 정보를 한 번만 저장하므로 중복이 제거됩니다. 다만 위치 리스트의 길이를 별도로 저장해야 하는 작은 오버헤드가 발생하는데, 이는 동시에 해당 단어의 문서 내 출현 빈도를 나타내는 데에도 활용될 수 있습니다.

단순 색인 방식과 그룹핑 방식의 공간 효율성을 분석해 보면, N개의 포스팅에 대해 전자는 2N 개의 저장 공간을 사용하는 데 비해, 후자는 W(2 + ln W) 개의 공간을 사용합니다 (W는 고유 단어 수). 보통 W ln W ≪ N 이므로 그룹핑이 더 공간 효율적임을 알 수 있고, 이는 대략 50% 가량의 절감 효과가 있는 것으로 나타납니다.

한편 그룹핑을 적용한 B-tree 역색인에서는 하나의 엔트리에 들어갈 수 있는 위치 정보의 수가 제한되는 문제가 있습니다. 극단적으로 출현 빈도가 높은 단어의 경우 위치 리스트가 B-tree 노드 크기를 초과할 수 있습니다.

이를 해결하기 위해 제안된 것이 '힙 파일(heap file)'입니다. B-tree 노드에는 (단어, 빈도, 포인터) 형태의 엔트리를 저장하고, 포인터가 가리키는 힙 파일에 위치 정보를 별도 저장하는 것입니다. 힙 파일 내에서는 가변 길이 청크를 동적 할당하여 위치 리스트를 저장하게 됩니다.

힙 파일을 사용하면 하나의 단어에 대한 색인 엔트리를 읽어들이는 데 최대 log_b W + 1 회의 디스크 읽기가 필요합니다 (트리 탐색 + 힙 파일 접근). 그리고 새로운 문서에 대한 색인을 추가할 때에는 힙 파일의 청크 크기가 딱 맞지 않을 경우 재할당이 일어날 수 있는데, 이에 따른 추가 비용은 무시할 만한 수준입니다.

결과적으로 N개의 새 포스팅에 대한 역색인 갱신 시간은 n(log_b w - log_b c + 1) 에 비례하게 되며, 앞서 소개된 병합 최적화 기법을 함께 적용하면 이를 더욱 개선할 수 있습니다.

힙 파일을 사용한 역색인의 공간 복잡도는 단순 그룹핑 대비 힙 파일 포인터 정보가 추가되는 것을 제외하면 거의 동일합니다. 즉 공간 효율성의 큰 저하 없이 B-tree 노드 크기 제한 문제를 해결할 수 있습니다.

Pulsing

Pulsing은 B-tree 노드 공간을 더 효율적으로 활용하기 위한 방법입니다. 앞서 힙 파일을 사용하면 모든 단어의 위치 정보를 B-tree 노드 밖으로 빼냈지만, 사실 대부분의 단어는 출현 빈도가 그리 높지 않아 노드 안에 충분히 저장할 수 있습니다.

따라서 Pulsing에서는 각 단어별로 B-tree 노드 안에 저장할 수 있는 위치 정보의 수를 제한합니다. 이 제한값을 초과하는 단어들에 대해서만 힙 파일을 사용하게 됩니다. 이렇게 하면 B-tree를 통한 직접 접근이 가능한 단어가 많아지므로 전반적인 검색 속도 향상을 기대할 수 있습니다.

Pulsing을 적용한 B-tree의 엔트리는 (단어, 총빈도, 노드 내 위치 수, 위치 리스트, 힙 포인터) 형태가 됩니다. 노드 내 위치 수가 제한값 이하일 때는 힙 포인터가 생략되어 공간이 절약됩니다.

Pulsing의 효과는 'hit ratio'로 분석할 수 있는데, 이는 검색 시 B-tree 노드만으로 처리 가능한 비율을 뜻합니다. Hit ratio가 높을수록 힙 파일 접근 없이 검색이 완료될 확률이 높아집니다. 노드 내 위치 수 제한값인 t를 적절히 설정함으로써 원하는 수준의 hit ratio를 얻을 수 있습니다.

  • 각 단어마다 B-tree 노드 안에 저장할 수 있는 위치 정보의 수를 t개로 제한합니다. 이를 'pulsing threshold'라고 합니다.
  • 어떤 단어의 색인 엔트리를 B-tree에 추가할 때, 그 단어의 총 출현 횟수가 t 이하면 위치 정보를 노드 안에 직접 저장합니다.
  • 만약 이미 t개의 위치 정보가 노드 안에 있다면, 기존의 위치 정보를 모두 힙 파일로 옮기고 새 위치 정보도 힙 파일에 저장합니다. 이 과정을 'pulsing out'이라고 합니다.
  • 즉, B-tree 노드 안에는 언제나 출현 빈도가 높은 단어들의 최신 위치 정보 t개만 저장되고, 나머지는 힙 파일로 overflow됩니다.
  • 이때 B-tree 노드의 엔트리는 (단어, 총빈도, 노드 내 위치 수, 위치 리스트, 힙 포인터) 형태가 됩니다. 총빈도가 t 이하면 힙 포인터는 생략됩니다.
  • 검색할 때는 먼저 B-tree를 탐색해 해당 단어의 엔트리를 찾고, 노드 안의 위치 정보를 읽습니다. 만약 총빈도가 t보다 크다면 힙 파일에서 나머지 위치 정보를 읽어옵니다.
  • Pulsing의 효과는 'hit ratio'로 측정할 수 있습니다. 이는 B-tree 탐색만으로 검색이 완료될 확률입니다. Hit ratio는 threshold t에 의해 결정되는데, t가 클수록 hit ratio도 높아집니다.

Delta Encoding

Delta Encoding은 위치 정보 자체를 압축하는 방법입니다. 역색인 검색에서는 대개 위치 정보를 순차적으로 읽어가므로, 각 위치를 이전 위치와의 차이(delta)로 표현해도 무방합니다. 실제 위치 값 대신 delta를 저장하면 그 값이 상당히 작아지게 되어 압축 효과를 얻을 수 있습니다.

또한 delta 값을 저장할 때 크기에 따라 가변 바이트 수를 사용하면 공간 효율을 더욱 높일 수 있습니다. 이러한 인코딩을 적용하면 색인 크기를 크게 줄일 수 있습니다.

Delta Encoding은 힙 파일의 위치 리스트뿐 아니라 Pulsing의 노드 내 위치 리스트에도 활용 가능합니다. 다만 개별 위치에 대한 직접 접근은 어려워지므로 문서 삭제 연산 등에는 부적합하다는 단점이 있습니다.

'논문' 카테고리의 다른 글

[LFS] Log-Structured File system  (1) 2022.08.22

+ Recent posts