Java 基於 ZooKeeper 實作分散式鎖 (可重入、公平、支援超時)

ZooKeeper 分散式鎖完整實作：基於臨時有序節點

利用 ZooKeeper 的臨時節點（Ephemeral）自動釋放鎖特性、有序節點（Sequential）保證公平性、Watcher 機制實現阻塞等待，實作 `java.util.concurrent.locks.Lock` 介面，可直接替換本地鎖用於分散式環境。

核心原理

1. 嘗試獲取鎖：在 `/_lock_` 下建立臨時有序節點。
2. 取得所有同前綴子節點，按序號排序。
3. 若自己是最小序號 → 獲得鎖，返回。
4. 否則 → 監聽「比自己小 1 的節點」，阻塞等待（`CountDownLatch`）。
5. 釋放鎖：刪除自己的節點 → 觸發下一個等待者的 Watcher → 喚醒。

完整實作 (精簡版)

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.*;

public class ZkDistributedLock implements Lock, Watcher {

    private final ZooKeeper zk;
    private final String root = "/locks";
    private final String lockName;
    private final String splitStr = "_lock_";
    private String myZnode;
    private String waitNode;
    private final CountDownLatch latch = new CountDownLatch(1);
    private final List exceptions = new CopyOnWriteArrayList<>();

    public ZkDistributedLock(String connectString, String lockName) throws IOException, InterruptedException, KeeperException {
        this.lockName = lockName;
        if (lockName.contains(splitStr)) throw new IllegalArgumentException("lockName 不可包含 '" + splitStr + "'");
        this.zk = new ZooKeeper(connectString, 30000, this);
        ensureRootExists();
    }

    private void ensureRootExists() throws KeeperException, InterruptedException {
        if (zk.exists(root, false) == null) {
            zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    @Override public void process(WatchedEvent event) {
        if (event.getType() == Event.EventType.NodeDeleted) {
            latch.countDown(); // 被監聽節點刪除 → 喚醒
        }
    }

    @Override
    public void lock() {
        if (!exceptions.isEmpty()) throw new LockException(exceptions.get(0));
        try {
            if (tryLock()) return;
            waitForLock(waitNode, 0); // 0 = 無限等待
        } catch (KeeperException | InterruptedException e) {
            throw new LockException(e);
        }
    }

    @Override
    public boolean tryLock() {
        try {
            myZnode = zk.create(root + "/" + lockName + splitStr, new byte[0],
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            List children = zk.getChildren(root, false);
            List lockNodes = new ArrayList<>();
            for (String child : children) {
                if (child.startsWith(lockName + splitStr)) lockNodes.add(child);
            }
            Collections.sort(lockNodes);
            if (myZnode.endsWith(lockNodes.get(0))) return true; // 我是最小

            // 找到前一個節點
            int idx = Collections.binarySearch(lockNodes, myZnode.substring(myZnode.lastIndexOf('/') + 1));
            waitNode = lockNodes.get(idx - 1);
        } catch (KeeperException | InterruptedException e) {
            throw new LockException(e);
        }
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) {
        try {
            if (tryLock()) return true;
            return waitForLock(waitNode, unit.toMillis(time));
        } catch (Exception e) {
            throw new LockException(e);
        }
    }

    private boolean waitForLock(String lower, long timeoutMs) throws InterruptedException, KeeperException {
        if (lower == null) return true;
        Stat stat = zk.exists(root + "/" + lower, true); // true = 註冊 Watcher
        if (stat != null) {
            if (timeoutMs > 0) latch.await(timeoutMs, TimeUnit.MILLISECONDS);
            else latch.await();
        }
        return true;
    }

    @Override
    public void unlock() {
        try {
            if (myZnode != null) {
                zk.delete(myZnode, -1);
                myZnode = null;
            }
        } catch (InterruptedException | KeeperException e) {
            throw new LockException(e);
        } finally {
            try { zk.close(); } catch (InterruptedException ignored) {}
        }
    }

    @Override public void lockInterruptibly() throws InterruptedException { lock(); }
    @Override public Condition newCondition() { return null; }

    public static class LockException extends RuntimeException {
        public LockException(String msg) { super(msg); }
        public LockException(Throwable cause) { super(cause); }
    }
}

使用範例

// 應用程式碼
ZkDistributedLock lock = new ZkDistributedLock("127.0.0.1:2181", "order-service");
try {
    lock.lock();
    try {
        // 執行分散式臨界區邏輯
        processOrder(orderId);
    } finally {
        lock.unlock();
    }
} catch (Exception e) {
    log.error("獲取分散式鎖失敗", e);
}

生產環境強化建議

• 連線狀態管理：實作 `ConnectionStateListener`，`SESSION_EXPIRED` 時拋出異常、清理狀態。
• 鎖超時自動釋放：節點本身是 Ephemeral，Client 掛掉 Session 過期自動刪除，天然防死鎖。
• 羊群效應緩解：僅監聽前一個節點（如本實作），避免所有 Client 同時監聽同一節點。
• 使用成熟庫：Apache Curator 提供 `InterProcessMutex`、`InterProcessSemaphoreMutex` 等生產級實作，建議直接使用。

效能基準 (參考)

單機 ZooKeeper 3.5+，鎖獲取/釋放延遲約 1~5ms (本地網路)，吞吐量受限於 ZooKeeper 寫入 TPS (約 10k ops/sec)。高併發場景建議分片多組鎖或改用 Redis Redlock。