분산 락(Distributed Lock) 패턴 비교: Redis Redlock vs ZooKeeper vs etcd — 정합성과 가용성의 트레이드오프

import redis
import uuid
import time


class RedisSimpleLock:
    """Redis 단일 인스턴스 기반 분산 락"""

    def __init__(self, client: redis.Redis, resource: str, ttl_ms: int = 10000):
        self.client = client
        self.resource = resource
        self.ttl_ms = ttl_ms
        self.lock_value = str(uuid.uuid4())  # 소유권 식별자

    def acquire(self, retry_count: int = 3, retry_delay_ms: int = 200) -> bool:
        """락 획득 시도. 실패 시 재시도."""
        for attempt in range(retry_count):
            result = self.client.set(
                self.resource,
                self.lock_value,
                nx=True,
                px=self.ttl_ms
            )
            if result:
                return True
            time.sleep(retry_delay_ms / 1000.0)
        return False

    def release(self) -> bool:
        """소유권 검증 후 락 해제 (Lua script 사용)"""
        lua_script = """
        if redis.call("GET", KEYS[1]) == ARGV[1] then
            return redis.call("DEL", KEYS[1])
        else
            return 0
        end
        """
        result = self.client.eval(lua_script, 1, self.resource, self.lock_value)
        return result == 1


# 사용 예시
client = redis.Redis(host="localhost", port=6379)
lock = RedisSimpleLock(client, "order:12345:lock", ttl_ms=5000)

if lock.acquire():
    try:
        # 임계 영역 작업 수행
        print("락 획득 성공, 작업 수행 중...")
    finally:
        lock.release()
else:
    print("락 획득 실패")

-- safe_unlock.lua
-- 소유권 검증 후 안전하게 락 해제
-- KEYS[1]: 락 키
-- ARGV[1]: 소유자 식별 값
-- 반환: 1 (성공), 0 (소유권 불일치)

if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end

import redis
import uuid
import time
from typing import List, Optional, Tuple


class Redlock:
    """Redlock 분산 락 알고리즘 구현"""

    CLOCK_DRIFT_FACTOR = 0.01  # 클럭 드리프트 보정 계수
    RETRY_DELAY_MS = 200
    RETRY_COUNT = 3

    def __init__(self, nodes: List[dict], ttl_ms: int = 10000):
        self.nodes = [
            redis.Redis(host=n["host"], port=n["port"], socket_timeout=0.1)
            for n in nodes
        ]
        self.quorum = len(self.nodes) // 2 + 1
        self.ttl_ms = ttl_ms

    def _acquire_single(self, client: redis.Redis, resource: str,
                        value: str) -> bool:
        try:
            return bool(client.set(resource, value, nx=True, px=self.ttl_ms))
        except redis.RedisError:
            return False

    def _release_single(self, client: redis.Redis, resource: str,
                        value: str) -> None:
        lua = """
        if redis.call("GET", KEYS[1]) == ARGV[1] then
            return redis.call("DEL", KEYS[1])
        else
            return 0
        end
        """
        try:
            client.eval(lua, 1, resource, value)
        except redis.RedisError:
            pass

    def acquire(self, resource: str) -> Optional[Tuple[str, float]]:
        """락 획득. 성공 시 (lock_value, validity_time) 반환"""
        for _ in range(self.RETRY_COUNT):
            lock_value = str(uuid.uuid4())
            start_time = time.monotonic()
            acquired_count = 0

            # Step 1: 모든 노드에 락 획득 시도
            for client in self.nodes:
                if self._acquire_single(client, resource, lock_value):
                    acquired_count += 1

            # Step 2: 유효성 검증
            elapsed_ms = (time.monotonic() - start_time) * 1000
            drift = self.ttl_ms * self.CLOCK_DRIFT_FACTOR + 2
            validity_time = self.ttl_ms - elapsed_ms - drift

            if acquired_count >= self.quorum and validity_time > 0:
                return (lock_value, validity_time)

            # 실패 시 모든 노드에서 해제
            for client in self.nodes:
                self._release_single(client, resource, lock_value)

            time.sleep(self.RETRY_DELAY_MS / 1000.0)

        return None

    def release(self, resource: str, lock_value: str) -> None:
        """모든 노드에서 락 해제"""
        for client in self.nodes:
            self._release_single(client, resource, lock_value)


# 사용 예시
nodes = [
    {"host": "redis1.example.com", "port": 6379},
    {"host": "redis2.example.com", "port": 6379},
    {"host": "redis3.example.com", "port": 6379},
    {"host": "redis4.example.com", "port": 6379},
    {"host": "redis5.example.com", "port": 6379},
]

redlock = Redlock(nodes, ttl_ms=10000)
result = redlock.acquire("payment:order:99999")

if result:
    lock_value, validity_ms = result
    try:
        print(f"락 획득 성공. 유효 시간: {validity_ms:.0f}ms")
        # 임계 영역 작업 수행
    finally:
        redlock.release("payment:order:99999", lock_value)

from kazoo.client import KazooClient
from kazoo.recipe.lock import Lock
import logging

logging.basicConfig(level=logging.INFO)


class ZooKeeperDistributedLock:
    """ZooKeeper 기반 분산 락 (Kazoo 라이브러리 활용)"""

    def __init__(self, hosts: str, lock_path: str):
        self.zk = KazooClient(hosts=hosts)
        self.zk.start()
        self.lock = Lock(self.zk, lock_path)
        self.lock_path = lock_path

    def acquire(self, timeout: float = 30.0) -> bool:
        """락 획득. timeout 초 내에 획득하지 못하면 False 반환"""
        try:
            return self.lock.acquire(timeout=timeout)
        except Exception as e:
            logging.error(f"락 획득 실패: {e}")
            return False

    def release(self) -> None:
        """락 해제"""
        try:
            self.lock.release()
        except Exception as e:
            logging.error(f"락 해제 실패: {e}")

    def get_fencing_token(self) -> int:
        """zxid를 Fencing Token으로 활용"""
        data, stat = self.zk.get(self.lock_path)
        return stat.czxid  # 생성 시 트랜잭션 ID (단조 증가)

    def close(self) -> None:
        self.zk.stop()
        self.zk.close()


# 사용 예시
zk_lock = ZooKeeperDistributedLock(
    hosts="zk1:2181,zk2:2181,zk3:2181",
    lock_path="/locks/payment/order-99999"
)

if zk_lock.acquire(timeout=10.0):
    try:
        fencing_token = zk_lock.get_fencing_token()
        logging.info(f"락 획득, fencing token: {fencing_token}")
        # fencing_token과 함께 스토리지에 쓰기
        # storage.write(data, fencing_token=fencing_token)
    finally:
        zk_lock.release()

zk_lock.close()

import etcd3
import threading
import logging
from typing import Optional, Tuple

logging.basicConfig(level=logging.INFO)


class EtcdDistributedLock:
    """etcd Lease 기반 분산 락"""

    def __init__(self, host: str = "localhost", port: int = 2379,
                 ttl: int = 10):
        self.client = etcd3.client(host=host, port=port)
        self.ttl = ttl
        self.lease: Optional[etcd3.Lease] = None
        self._keepalive_thread: Optional[threading.Thread] = None
        self._stop_keepalive = threading.Event()

    def acquire(self, lock_key: str,
                timeout: float = 30.0) -> Optional[Tuple[int, int]]:
        """
        락 획득. 성공 시 (revision, lease_id) 반환.
        revision을 Fencing Token으로 사용 가능.
        """
        self.lease = self.client.lease(self.ttl)
        self.lock_key = lock_key

        # 트랜잭션으로 원자적 락 획득
        # 키가 존재하지 않을 때만 생성 (Compare-And-Swap)
        success, responses = self.client.transaction(
            compare=[
                self.client.transactions.create(lock_key) == 0
            ],
            success=[
                self.client.transactions.put(
                    lock_key, "locked", lease=self.lease
                )
            ],
            failure=[]
        )

        if success:
            # Revision을 Fencing Token으로 사용
            revision = responses[0].header.revision
            self._start_keepalive()
            logging.info(
                f"락 획득 성공. revision(fencing token): {revision}"
            )
            return (revision, self.lease.id)

        logging.warning("락 획득 실패: 이미 다른 클라이언트가 보유 중")
        self.lease.revoke()
        return None

    def _start_keepalive(self) -> None:
        """Lease 자동 갱신 스레드 시작"""
        self._stop_keepalive.clear()

        def keepalive_loop():
            while not self._stop_keepalive.is_set():
                try:
                    self.lease.refresh()
                except Exception as e:
                    logging.error(f"Lease 갱신 실패: {e}")
                    break
                self._stop_keepalive.wait(timeout=self.ttl / 3.0)

        self._keepalive_thread = threading.Thread(
            target=keepalive_loop, daemon=True
        )
        self._keepalive_thread.start()

    def release(self) -> None:
        """락 해제"""
        self._stop_keepalive.set()
        if self.lease:
            try:
                self.lease.revoke()
                logging.info("락 해제 완료")
            except Exception as e:
                logging.error(f"락 해제 실패: {e}")

    def close(self) -> None:
        self.release()
        self.client.close()


# 사용 예시
lock = EtcdDistributedLock(host="etcd1.example.com", ttl=15)
result = lock.acquire("locks/payment/order-99999")

if result:
    fencing_token, lease_id = result
    try:
        # fencing_token(revision)과 함께 스토리지에 쓰기
        logging.info(f"작업 수행 중, fencing token: {fencing_token}")
    finally:
        lock.release()

시간 ------>

클라이언트 A: [락 획득] --- [GC pause 시작] -------------------- [GC 복귀, 쓰기 시도]
                                              TTL 만료
클라이언트 B:                          [락 획득] --- [쓰기 완료] --- [쓰기 완료]

결과: A와 B 모두 쓰기 수행 -> 데이터 손상

class FencingAwareStorage:
    """Fencing Token을 검증하는 스토리지 래퍼"""

    def __init__(self):
        self.last_token = 0
        self._lock = threading.Lock()

    def write(self, data: dict, fencing_token: int) -> bool:
        """fencing_token이 이전 값보다 클 때만 쓰기 허용"""
        with self._lock:
            if fencing_token <= self.last_token:
                logging.warning(
                    f"거부: token {fencing_token} <= "
                    f"last {self.last_token}"
                )
                return False
            self.last_token = fencing_token
            # 실제 쓰기 수행
            self._do_write(data)
            logging.info(
                f"쓰기 성공: token {fencing_token}"
            )
            return True

    def _do_write(self, data: dict) -> None:
        # 실제 스토리지 쓰기 로직
        pass

from kazoo.client import KazooState


def connection_listener(state):
    """ZooKeeper 연결 상태 모니터링"""
    if state == KazooState.SUSPENDED:
        # 연결 일시 중단: 진행 중인 작업 일시 정지
        logging.warning("ZK 연결 중단 - 작업 일시 정지")
        pause_current_operations()
    elif state == KazooState.LOST:
        # 세션 만료: 락 소실 확정, 작업 즉시 중단
        logging.error("ZK 세션 만료 - 락 소실, 작업 중단")
        abort_current_operations()
    elif state == KazooState.CONNECTED:
        # 재연결 성공: 락 재획득 시도
        logging.info("ZK 재연결 - 락 재획득 시도")
        reacquire_lock()


zk = KazooClient(hosts="zk1:2181,zk2:2181,zk3:2181")
zk.add_listener(connection_listener)
zk.start()

나쁜 예:  /locks/orders           (모든 주문에 대한 단일 락)
보통:     /locks/orders/user-123  (사용자 단위 락)
좋은 예:  /locks/orders/99999     (개별 주문 단위 락)