Redis 분산락, 직접 구현하면서 이해해보자! - 2편: Pub/Sub 방식

❇️ 들어가며

안녕하세요! 저번 글에서는 Spin Lock 방식의 분산락을 구현해 봤습니다. 이번 글에서는 Spin Lock 방식의 단점들을 보완한 Pub/Sub 방식을 구현해 보겠습니다. 이어지는 글이므로 1편: Spin Lock 방식을 먼저 읽어주시면 감사하겠습니다. 

 

글에서 설명할 코드는 Spring, Kotlin, Spring Data Redis로 작성되어 있습니다. 테스트 코드는 Testcontainers를 사용하기 때문에 Docker를 실행하고 테스트해 주시길 바랍니다. 전체 코드는 다음의 링크를 확인해 주세요.

spring-lab/distributed-lock-impl at main · alstn113/spring-lab

 

❇️ Spin Lock 방식의 단점

Spin Lock 방식의 단점들을 다시 한번 짚고 넘어가겠습니다.

• 짧은 주기의 재시도 요청으로 인해 서버와 Redis CPU 부하가 증가합니다.
• 락 해제 시점과 재시도 요청 타이밍의 불일치로 락 효율이 저하됩니다.
• 락 경쟁이 심해질수록 재시도 요청이 많아지므로 성능 저하가 더욱 심해집니다.

이처럼 Spin Lock은 락이 언제 풀릴지 모르기 때문에 계속 두드리는 방식입니다. 즉, 락이 해제되었는지 수동으로 반복 확인해야 합니다.

 

❇️ Pub/Sub 방식의 기본 원리

그렇다면 더 효율적인 방법은 무엇일까요? 락이 해제될 때 Redis가 직접 알려주는 구조를 만들면 됩니다. 대기 중인 스레드들이 특정 락을 Subscribe를 통해 구독하고, Redis가 락 해제 시점을 Publish를 통해 알리면 Spin 과정 없이 즉시 재시도가 가능합니다. 이런 구조가 바로 Pub/Sub 기반 분산락입니다. Spin Lock의 불필요한 반복 요청 문제를 해결하고, 락 해제 시점에 정확히 반응할 수 있어 더 효율적인 접근 방식입니다. 

 

❇️ Pub/Sub 방식의 분산락 구현

🔷 기본 틀 설계

먼저, 기본 틀을 간단하게 설계해 보겠습니다. 락 획득은 다음과 같이 진행됩니다.

1. 락 획득을 시도하고 성공 시 true를 반환합니다.
2. 락을 아직 획득하지 못했다면, 해당 키의 채널을 구독합니다.
3. 락이 해제될 때까지 알림을 기다립니다.
4. 락 해제 알림을 받으면 락 획득을 재시도합니다.
5. 락 획득에 실패하면 3번부터 다시 반복합니다. 타임아웃 시 false를 반환합니다.
6. 락 획득 성공/실패 여부와 상관없이 구독을 해제합니다.

락 해제는 Spin Lock 방식과 유사합니다. 단지 락 해제 시 락이 풀렸음을 알려줘야 합니다. 즉, Redis 명령어인 "Publish channel message"를 통해서 해당 채널에 알려야 합니다.

 

Pub/Sub을 사용하기 위해서 Spring Data Redis가 제공하는 RedisMessageListnerContainer를 등록하겠습니다.

package io.github.alstn113.app.distributedlock.pubsub

import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.data.redis.connection.RedisConnectionFactory
import org.springframework.data.redis.listener.RedisMessageListenerContainer

@Configuration
class RedisPubSubConfig {

    @Bean
    fun redisMessageListenerContainer(
        connectionFactory: RedisConnectionFactory,
    ): RedisMessageListenerContainer {
        val container = RedisMessageListenerContainer()
        container.setConnectionFactory(connectionFactory)

        return container
    }
}

🔷 락 해제 알림 및 구독 구현

MessageListener를 구현한 LockNotificationManager를 알아보겠습니다. 이는 락 해제 알림을 관리하는 컴포넌트입니다.

• 락을 획득하지 못한 스레드는 subscribe 메서드를 통해 Redis 채널을 구독하고 대기합니다.
• 락이 해제되면 Redis가 publish를 통해 알림을 전송합니다.
• LockNotificationManager는 RedisMessageListenerContainer를 통해 알림을 수신하고, 대기 중인 스레드를 깨워 락 획득 재시도를 가능하게 합니다.

하나의 서버 내에서 여러 스레드들이 개별적으로 채널을 구독할 수 있는 상황을 가정해 보겠습니다. 여러 스레드가 락 획득을 시도하고 락 획득에 실패한 스레드들은 해당 키에 대한 채널을 개별적으로 구독하게 됩니다. 그리고 락 해제 알림을 받으면 대기 중인 모든 스레드들이 깨어나 락 획득을 재시도하게 됩니다. 이 경우 다음과 같은 문제점이 있습니다.

• Redis에 불필요하게 많은 구독이 생깁니다.
• 락 획득은 한 스레드만 할 수 있으나 모든 스레드들이 깨어나 락 획득을 재시도해 부하를 줍니다.

이 문제를 해결하기 위해서 각 락 키에 대해 구독 상태(semaphore, counter)를 저장합니다. 동일 키에 여러 스레드가 대기 중이면 한 채널만 구독하고 세마포를 공유합니다. 즉, 불필요한 Redis 구독을 방지해 성능을 높이는 구조입니다. 또한 알림을 받으면 세마포를 릴리즈해서 여러 스레드 중 단 하나의 스레드만 깨어나 다음 락 획득 시도를 수행하게 했습니다.

 

먼저, LockNotificationManager의 전체 코드를 확인하고 세부적인 내용들에 대해서 알아보겠습니다.

package io.github.alstn113.app.distributedlock.pubsub

import org.slf4j.LoggerFactory
import org.springframework.data.redis.connection.Message
import org.springframework.data.redis.connection.MessageListener
import org.springframework.data.redis.listener.ChannelTopic
import org.springframework.data.redis.listener.RedisMessageListenerContainer
import org.springframework.data.redis.serializer.StringRedisSerializer
import org.springframework.stereotype.Component
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.Semaphore
import java.util.concurrent.atomic.AtomicInteger

@Component
class LockNotificationManager(
    private val container: RedisMessageListenerContainer,
) : MessageListener {

    private val log = LoggerFactory.getLogger(javaClass)
    private val serializer = StringRedisSerializer.UTF_8
    private val registry = ConcurrentHashMap<String, LockWaitEntry>()

    private data class LockWaitEntry(
        val semaphore: Semaphore = Semaphore(0),
        val counter: AtomicInteger = AtomicInteger(1),
    )

    fun subscribe(key: String): Semaphore {
        var created = false
        val entry = registry.compute(key) { _, existing ->
            if (existing == null) {
                log.info("새로운 LockWaitEntry 생성 및 구독: key={}", key)
                created = true
                LockWaitEntry()
            } else {
                existing.counter.incrementAndGet()
                log.info("기존 LockWaitEntry 재사용: key={}, refCount={}", key, existing.counter.get())
                existing
            }
        }!!

        if (created) {
            container.addMessageListener(this, ChannelTopic(key + NOTIFY_SUFFIX))
        }

        return entry.semaphore
    }

    fun unsubscribe(key: String) {
        registry.computeIfPresent(key) { _, entry ->
            val remain = entry.counter.decrementAndGet()
            log.info("LockWaitEntry 해제: key={}, 남은 refCount={}", key, remain)

            if (remain <= 0) {
                log.info("LockWaitEntry 제거 및 구독 해제: key={}", key)
                container.removeMessageListener(this, ChannelTopic(key + NOTIFY_SUFFIX))
                null
            } else {
                entry
            }
        }
    }

    override fun onMessage(message: Message, pattern: ByteArray?) {
        val channel = serializer.deserialize(message.channel) ?: return
        if (!channel.endsWith(NOTIFY_SUFFIX)) {
            return
        }

        val key = channel.removeSuffix(NOTIFY_SUFFIX)
        if (key.isBlank()) {
            return
        }

        val entry = registry[key] ?: return
        entry.semaphore.release()
        log.info("락 해제 알림 수신: key={}, 남은 대기자={}", key, entry?.counter ?: 0)
    }

    companion object {
        const val NOTIFY_SUFFIX = ":notify"
    }
}

 

subscribe 메서드는 락 해제 알림을 받기 위해 채널을 구독하는 메서드입니다. 락 키에 대한 LockWaitEntry가 존재하지 않는 경우 새로운 LockWaitEntry를 만들고, 해당 채널에 구독하게 됩니다. Semahore(0)은 허용 가능한 동시 접근 개수가 0개임을 의미합니다. 락 획득에 실패했고, 알림을 받기 전이므로 접근할 수 없음을 의미합니다. counter는 동일 키에 대해서 대기 중인 스레드 수입니다. 이미 락 키에 대한 LockWaitEntry가 존재하는 경우 기존 LockWaitEntry의 counter만 증가시킵니다. 이를 통해, 동일한 키에 대해서 불필요한 구독을 줄일 수 있습니다.

// LockNotification.kt 중 일부

private val registry = ConcurrentHashMap<String, LockWaitEntry>()

private data class LockWaitEntry(
    val semaphore: Semaphore = Semaphore(0),
    val counter: AtomicInteger = AtomicInteger(1),
)

fun subscribe(key: String): Semaphore {
    var created = false
    val entry = registry.compute(key) { _, existing ->
        if (existing == null) {
            log.info("새로운 LockWaitEntry 생성 및 구독: key={}", key)
            created = true
            LockWaitEntry()
        } else {
            existing.counter.incrementAndGet()
            log.info("기존 LockWaitEntry 재사용: key={}, refCount={}", key, existing.counter.get())
            existing
        }
    }!!

    if (created) {
        container.addMessageListener(this, ChannelTopic(key + NOTIFY_SUFFIX))
    }

    return entry.semaphore
}

 

unsubscribe는 락 획득 성공/실패와 상관없이 마지막에 구독을 해제하기 위해서 사용합니다. counter를 감소시키고, counter가 0인 경우는 동일 키에 대해서 대기 중인 스레드가 없음을 의미하므로 구독을 해제합니다.

fun unsubscribe(key: String) {
    registry.computeIfPresent(key) { _, entry ->
        val remain = entry.counter.decrementAndGet()
        log.info("LockWaitEntry 해제: key={}, 남은 refCount={}", key, remain)

        if (remain <= 0) {
            log.info("LockWaitEntry 제거 및 구독 해제: key={}", key)
            container.removeMessageListener(this, ChannelTopic(key + NOTIFY_SUFFIX))
            null
        } else {
            entry
        }
    }
}

 

onMessage는 채널을 통해 받은 알림을 처리하기 위해서 구현한 것입니다. 락 해제 알림을 받은 경우 해당 키에 대한 LockWaitEntry의 Semaphore를 release 합니다. release는 허용 가능한 동시 접근 개수(permits)를 1 증가시킵니다. 즉, 여러 스레드 중 하나가 접근할 수 있게 함을 의미합니다.

override fun onMessage(message: Message, pattern: ByteArray?) {
    val channel = serializer.deserialize(message.channel) ?: return
    if (!channel.endsWith(NOTIFY_SUFFIX)) {
        return
    }

    val key = channel.removeSuffix(NOTIFY_SUFFIX)
    if (key.isBlank()) {
        return
    }

    val entry = registry[key] ?: return
    entry.semaphore.release()
    log.info("락 해제 알림 수신: key={}, 남은 대기자={}", key, entry?.counter ?: 0)
}

🔷 락 획득 시도와 락 해제 구현

DistributedLock 인터페이스를 구현한 PubSubLock을 만들어보겠습니다. PubSubLock을 만들기 위한 내부 메서드들은 Spin Lock과 크게 다르지 않습니다. 중요한 부분은 락 획득 실패 시 구독하고 알림을 받기 위해 대기하는 부분입니다. PubSubLock의 전체 코드를 보고, 중요한 부분들에 대해서 설명하겠습니다.

package io.github.alstn113.app.distributedlock.pubsub

import io.github.alstn113.app.distributedlock.DistributedLock
import org.springframework.beans.factory.annotation.Qualifier
import org.springframework.data.redis.core.StringRedisTemplate
import org.springframework.data.redis.core.script.RedisScript
import org.springframework.stereotype.Component
import java.time.Duration
import java.util.*
import java.util.concurrent.Semaphore
import java.util.concurrent.TimeUnit


@Component
class PubSubLock(
    private val redisTemplate: StringRedisTemplate,
    private val manager: LockNotificationManager,
    @Qualifier("tryAcquireScript") private val tryAcquireScript: RedisScript<Long>,
    @Qualifier("unlockPublishScript") private val unlockPublishScript: RedisScript<Long>,
) : DistributedLock {

    override fun tryLock(key: String, waitTime: Duration, leaseTime: Duration): Boolean {
        val deadlineMillis = System.currentTimeMillis() + waitTime.toMillis()

        val ttl = tryAcquire(key, leaseTime)
        if (ttl == -1L) {
            return true
        }

        if (isDeadlineExceeded(deadlineMillis)) {
            return false
        }

        val semaphore = manager.subscribe(key)

        try {
            return waitForUnlock(
                key = key,
                leaseTime = leaseTime,
                deadlineMillis = deadlineMillis,
                semaphore = semaphore
            )
        } finally {
            manager.unsubscribe(key)
        }
    }

    private fun waitForUnlock(
        key: String,
        leaseTime: Duration,
        deadlineMillis: Long,
        semaphore: Semaphore,
    ): Boolean {
        while (true) {
            if (isDeadlineExceeded(deadlineMillis)) {
                return false
            }

            // 대기하기 전에 알림이 올 수 있으므로 즉시 재시도
            val ttl = tryAcquire(key, leaseTime)
            if (ttl == -1L) {
                return true
            }

            if (isDeadlineExceeded(deadlineMillis)) {
                return false
            }

            // 락이 해제되기를 대기, 알림 시 여러 스레드 중 하나만 깨어날 수 있음
            val remainTimeMillis = deadlineMillis - System.currentTimeMillis()
            val acquired = semaphore.tryAcquire(remainTimeMillis, TimeUnit.MILLISECONDS)
            if (!acquired) { // 타임아웃 발생
                return false
            }
        }
    }

    override fun unlock(key: String) {
        val notifyChannel = "${key}${LockNotificationManager.NOTIFY_SUFFIX}"

        redisTemplate.execute(
            unlockPublishScript,
            listOf(key),
            getLockOwnerId(),
            notifyChannel,
        )
    }

    /**
     * @return 락을 획득하거나 재진입한 경우 -1 반환, 실패 시 남은 TTL(밀리초) 반환
     */
    private fun tryAcquire(key: String, leaseTime: Duration): Long {
        val leaseTimeMillis = leaseTime.toMillis()

        return redisTemplate.execute(
            tryAcquireScript,
            listOf(key),
            leaseTimeMillis.toString(),
            getLockOwnerId()
        )
    }

    private fun isDeadlineExceeded(deadlineMillis: Long): Boolean {
        return System.currentTimeMillis() > deadlineMillis
    }

    private fun getLockOwnerId(): String {
        val threadId = Thread.currentThread().threadId()
        return "$INSTANCE_ID:$threadId"
    }

    companion object {
        private val INSTANCE_ID = UUID.randomUUID().toString()
    }
}

 

Spin Lock과 동일한 방식으로 tryAcquire을 해서 락 획득을 시도합니다. 락 획득에 성공한 경우 true를 반환합니다. 락 획득이 실패한 경우에는 manager.subscribe(key)를 통해서 알림 채널을 구독하게 됩니다. 앞서 설명했듯이, 여러 스레드가 구독하는 경우 실제로 하나만 구독하게 됩니다. waitForUnlock 메서드에서는 알림을 받고 락 획득을 시도합니다. 마지막으로 락 획득 성공/실패 여부와 상관없이 unsubscribe를 하게 됩니다. 여기서도 알림을 대기 중인 스레드가 0인 경우에만 실제 구독을 해제하게 됩니다.

val semaphore = manager.subscribe(key)

try {
    return waitForUnlock(
        key = key,
        leaseTime = leaseTime,
        deadlineMillis = deadlineMillis,
        semaphore = semaphore
    )
} finally {
    manager.unsubscribe(key)
}

 

waitForUnlock의 while 반복문의 시작 부분에서 즉시 tryAcquire 메서드를 호출해 락 획득을 재시도합니다. 이렇게 하는 이유는 락 획득 실패 직후, 알림 구독을 설정하기 전에 락이 이미 해제되는 경우가 있을 수 있기 때문입니다. 그런 상황에서 재시도를 하지 않으면 락 해제 알림을 받지 못한 채 무한히 대기하게 됩니다. 이 문제를 방지하기 위해, 구독 직후 한 번 더 락 획득을 시도하여 이미 해제된 락을 놓치지 않도록 합니다. 

 

semaphore.tryAcquire 메서드는 Semaphore의 permit(허가권)을 얻기 위해 정해진 시간 동안 대기하는 동작을 의미합니다. Redis로부터 락 해제 알림이 도착하면 semaphore.release 메서드가 호출되어 permit이 1 증가하고, 그 결과 여러 대기 스레드 중 하나의 스레드가 tryAcquire 메서드를 통해 깨어나 락 획득을 재시도하게 됩니다.

private fun waitForUnlock(
    key: String,
    leaseTime: Duration,
    deadlineMillis: Long,
    semaphore: Semaphore,
): Boolean {
    while (true) {
        if (isDeadlineExceeded(deadlineMillis)) {
            return false
        }

        // 대기하기 전에 알림이 올 수 있으므로 즉시 재시도
        val ttl = tryAcquire(key, leaseTime)
        if (ttl == -1L) {
            return true
        }

        if (isDeadlineExceeded(deadlineMillis)) {
            return false
        }

        // 락이 해제되기를 대기, 알림 시 여러 스레드 중 하나만 깨어날 수 있음
        val remainTimeMillis = deadlineMillis - System.currentTimeMillis()
        val acquired = semaphore.tryAcquire(remainTimeMillis, TimeUnit.MILLISECONDS)
        if (!acquired) { // 타임아웃 발생
            return false
        }
    }
}

 

❇️ 실제 분산락 라이브러리(Redisson)은 어떻게 구현될까?

지금까지 Pub/Sub 방식의 분산락을 직접 구현해봤습니다. 이 방식은 기본적인 동작을 충실히 수행하지만, 락 요청 순서를 보장하지 않기 때문에 특정 스레드가 계속 락을 얻지 못하는 기아 문제가 발생할 수 있습니다. 또한 leaseTime을 초과하여 로직이 실행되는 경우, 락이 이미 해제된 상태에서 작업이 진행될 수 있어 데이터 무결성이 깨질 위험성이 있습니다.

 

Redisson에서는 다양한 상황에 대응할 수 있는 분산락 구현체를 제공합니다.

• RedissonLock: 기본적인 분산 락, 이번 글에서 구현한 Pub/Sub 기반 방식과 유사합니다.
• RedissonSpinLock: 지속적인 재시도를 수행하는 Spin Lock 방식. 이전 글에서 구현한 방식과 유사합니다.
• RedissonFairLock: 락 요청 순서를 보장하여 기아 현상을 방지하는 공정한 분산락입니다.

이 외에도 여러 특화된 구현체가 존재합니다. 특히 주목할 만한 것은 RedissonRedLock과 RedissonFencedLock입니다.

🔷 RedLock과 그에 대한 한계

RedLock 알고리즘은 Redis 창시자가 제안한 분산락 프로토콜로, 여러 노드에 동시에 락을 획득하여 단일 Redis 장애에도 안전한 락을 제공하려는 목적이 있었습니다. 하지만 실제 환경에서는 다음과 같은 문제들이 발생할 수 있었습니다. 

• 락 획득 후 만료된 락을 막을 방법의 부재
• 서버 간 시계 불일치(Clock Drift)

이로 인해 RedLock은 일관성을 보장하기 어렵다고 판단하고 RedissonRedLock이 Deprecated 되었습니다.

🔷 FencedLock과 그에 대한 한계

FencedLock은 RedLock의 복잡한 합의 구조를 대체하기보다는, 락 이후 단계의 안정성을 보완하기 위해 설계되었습니다. 이 락은 Fencing Token이라는 단조 증가하는 숫자를 함께 반환하며, 리소스에 접근할 때 이 토큰을 함께 전달합니다. 리소스 쪽에서는 이전 토큰보다 작은 요청을 거부함으로써, leaseTime 만료 이후 도착한 늦은 요청을 방지할 수 있습니다. 즉, FencedLock은 누가 더 늦게 락을 잡았는지를 명확히 구분하여, leaseTime 초과나 네트워크 지연으로 발생할 수 있는 데이터 충돌을 예방합니다.

다만, FencedLock의 토큰 증가 로직은 Redis 상에서 동작하기 때문에, 토큰 단조 증가가 절대적으로 보장되지는 않습니다. 따라서 Redis를 락 관리용으로, DB를 Fencing Token 관리용으로 사용하는 것이 바람직합니다.

한편, 중요한 분산 트랜잭션 제어나 강한 일관성 보장이 필요한 시스템에서는, Redis 기반 분산락보다는 Zookeeper, Etcd 등 합의 기반 시스템을 사용하는 것이 바람직합니다. 결국 성능과 일관성 보장 사이에서 트레이드오프를 고려해 시스템을 설계해야 합니다.

 

❇️ 마치며

Spin Lock 방식과 Pub/Sub 방식을 직접 구현해보면서 분산락의 핵심 원리를 이해할 수 있었습니다. 그동안 RedLock과 그 한계에 대해서는 알고 있었지만, Deprecated되었다는 점과 FencedLock이 존재한다는 사실은 이번에 새롭게 알게 되었습니다. 이번 경험을 통해 분산 환경에서 락의 신뢰성을 어떻게 확보할 수 있을지 고민해볼 수 있는 좋은 기회가 되었습니다. 혹시 내용 중 잘못된 부분이 있다면 댓글로 알려주시면 감사하겠습니다.

 

❇️ 참고

• jhzlo님의 블로그 - RedLock의 한계
• diggingcode님의 블로그 - Redisson 구현체 모음  
• Martin Kleppmann님의 블로그 - Redlock 한계에 대한 글
• Redisson 공식문서 - FencedLock