分散ロック（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スクリプト使用）"""
        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     (個別注文単位ロック)