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

import redis
import uuid
import time


class RedisSimpleLock:
    """Redis single instance distributed lock"""

    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())  # Ownership identifier

    def acquire(self, retry_count: int = 3, retry_delay_ms: int = 200) -> bool:
        """Attempt to acquire lock. Retry on failure."""
        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:
        """Release lock after ownership verification (using 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


# Usage example
client = redis.Redis(host="localhost", port=6379)
lock = RedisSimpleLock(client, "order:12345:lock", ttl_ms=5000)

if lock.acquire():
    try:
        # Perform critical section work
        print("Lock acquired, performing work...")
    finally:
        lock.release()
else:
    print("Failed to acquire lock")

-- safe_unlock.lua
-- Safe lock release after ownership verification
-- KEYS[1]: Lock key
-- ARGV[1]: Owner identification value
-- Returns: 1 (success), 0 (ownership mismatch)

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 distributed lock algorithm implementation"""

    CLOCK_DRIFT_FACTOR = 0.01  # Clock drift correction factor
    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]]:
        """Acquire lock. Returns (lock_value, validity_time) on success"""
        for _ in range(self.RETRY_COUNT):
            lock_value = str(uuid.uuid4())
            start_time = time.monotonic()
            acquired_count = 0

            # Step 1: Attempt lock acquisition on all nodes
            for client in self.nodes:
                if self._acquire_single(client, resource, lock_value):
                    acquired_count += 1

            # Step 2: Validity check
            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)

            # On failure, release on all nodes
            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:
        """Release lock on all nodes"""
        for client in self.nodes:
            self._release_single(client, resource, lock_value)


# Usage example
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"Lock acquired. Valid for: {validity_ms:.0f}ms")
        # Perform critical section work
    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-based distributed lock (using Kazoo library)"""

    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:
        """Acquire lock. Returns False if not acquired within timeout seconds"""
        try:
            return self.lock.acquire(timeout=timeout)
        except Exception as e:
            logging.error(f"Lock acquisition failed: {e}")
            return False

    def release(self) -> None:
        """Release lock"""
        try:
            self.lock.release()
        except Exception as e:
            logging.error(f"Lock release failed: {e}")

    def get_fencing_token(self) -> int:
        """Use zxid as Fencing Token"""
        data, stat = self.zk.get(self.lock_path)
        return stat.czxid  # Creation transaction ID (monotonically increasing)

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


# Usage example
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"Lock acquired, fencing token: {fencing_token}")
        # Write to storage with 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-based distributed lock"""

    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]]:
        """
        Acquire lock. Returns (revision, lease_id) on success.
        Revision can be used as Fencing Token.
        """
        self.lease = self.client.lease(self.ttl)
        self.lock_key = lock_key

        # Atomic lock acquisition via transaction
        # Create only when key doesn't exist (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:
            # Use Revision as Fencing Token
            revision = responses[0].header.revision
            self._start_keepalive()
            logging.info(
                f"Lock acquired. revision(fencing token): {revision}"
            )
            return (revision, self.lease.id)

        logging.warning("Lock acquisition failed: already held by another client")
        self.lease.revoke()
        return None

    def _start_keepalive(self) -> None:
        """Start Lease auto-renewal thread"""
        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 renewal failed: {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:
        """Release lock"""
        self._stop_keepalive.set()
        if self.lease:
            try:
                self.lease.revoke()
                logging.info("Lock released")
            except Exception as e:
                logging.error(f"Lock release failed: {e}")

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


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

if result:
    fencing_token, lease_id = result
    try:
        # Write to storage with fencing_token (revision)
        logging.info(f"Performing work, fencing token: {fencing_token}")
    finally:
        lock.release()

Time ------>

Client A: [Lock acquired] --- [GC pause starts] -------------------- [GC resumes, attempts write]
                                              TTL expires
Client B:                          [Lock acquired] --- [Write complete] --- [Write complete]

Result: Both A and B write -> Data corruption

class FencingAwareStorage:
    """Storage wrapper that validates Fencing Tokens"""

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

    def write(self, data: dict, fencing_token: int) -> bool:
        """Allow writes only when fencing_token is greater than previous value"""
        with self._lock:
            if fencing_token <= self.last_token:
                logging.warning(
                    f"Rejected: token {fencing_token} <= "
                    f"last {self.last_token}"
                )
                return False
            self.last_token = fencing_token
            # Perform actual write
            self._do_write(data)
            logging.info(
                f"Write successful: token {fencing_token}"
            )
            return True

    def _do_write(self, data: dict) -> None:
        # Actual storage write logic
        pass

from kazoo.client import KazooState


def connection_listener(state):
    """ZooKeeper connection state monitoring"""
    if state == KazooState.SUSPENDED:
        # Connection suspended: pause ongoing work
        logging.warning("ZK connection suspended - pausing operations")
        pause_current_operations()
    elif state == KazooState.LOST:
        # Session expired: lock loss confirmed, stop work immediately
        logging.error("ZK session expired - lock lost, aborting operations")
        abort_current_operations()
    elif state == KazooState.CONNECTED:
        # Reconnection successful: attempt lock reacquisition
        logging.info("ZK reconnected - attempting lock reacquisition")
        reacquire_lock()


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

Bad:     /locks/orders           (Single lock for all orders)
Medium:  /locks/orders/user-123  (Per-user lock)
Good:    /locks/orders/99999     (Per-order lock)