Zookeeper如何实现分布式锁

5月 3rd, 2025

ZooKeeper是一个分布式的、开源的分布式应用程序协调服务，它可以用来实现分布式锁。以下是使用ZooKeeper实现分布式锁的基本步骤：

1. 创建锁节点

临时顺序节点：在ZooKeeper中创建一个持久节点作为锁的根节点。
子节点：在根节点下创建多个临时顺序节点，每个节点代表一个锁请求。

2. 获取锁

获取节点列表：客户端获取根节点下的所有子节点，并按顺序排序。
检查顺序：客户端检查自己创建的节点是否是当前最小的节点。
- 如果是，则表示该客户端获得了锁。
- 如果不是，则监听比自己小的那个节点的删除事件。

3. 释放锁

删除节点：当客户端完成任务后，删除自己创建的临时顺序节点，从而释放锁。
通知其他客户端：删除节点会触发监听该节点的其他客户端的回调函数，通知它们有机会重新尝试获取锁。

4. 处理异常情况

会话超时：如果客户端会话超时，ZooKeeper会自动删除该客户端的临时节点，从而释放锁。
网络分区：在网络分区的情况下，ZooKeeper会确保只有一个节点能够成功获取锁，其他节点会等待直到分区恢复。

示例代码

以下是一个简单的Java示例，展示了如何使用ZooKeeper实现分布式锁：

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class DistributedLock {
    private static final String ZK_ADDRESS = "localhost:2181";
    private static final int SESSION_TIMEOUT = 3000;
    private static final String LOCK_ROOT = "/locks";
    private static final String LOCK_NODE = LOCK_ROOT + "/lock_";

    private ZooKeeper zk;
    private String lockPath;

    public DistributedLock() throws IOException, InterruptedException, KeeperException {
        zk = new ZooKeeper(ZK_ADDRESS, SESSION_TIMEOUT, event -> {
            // 处理连接事件
        });
        // 确保锁根节点存在
        Stat stat = zk.exists(LOCK_ROOT, false);
        if (stat == null) {
            zk.create(LOCK_ROOT, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    public void lock() throws KeeperException, InterruptedException {
        lockPath = zk.create(LOCK_NODE, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        while (true) {
            List<String> children = zk.getChildren(LOCK_ROOT, false);
            Collections.sort(children);
            if (lockPath.endsWith(children.get(0))) {
                // 获取到锁
                return;
            } else {
                // 监听前一个节点的删除事件
                String previousNode = getPreviousNode(children, lockPath);
                CountDownLatch latch = new CountDownLatch(1);
                zk.exists(LOCK_ROOT + "/" + previousNode, event -> {
                    if (event.getType() == Watcher.Event.EventType.NodeDeleted) {
latch.countDown();
                    }
                });
                latch.await();
            }
        }
    }

    public void unlock() throws KeeperException, InterruptedException {
        if (lockPath != null) {
            zk.delete(lockPath, -1);
            lockPath = null;
        }
    }

    private String getPreviousNode(List<String> children, String currentNode) {
        int index = children.indexOf(currentNode.substring(LOCK_ROOT.length() + 1));
        return index > 0 ? children.get(index - 1) : null;
    }

    public static void main(String[] args) {
        try {
            DistributedLock lock = new DistributedLock();
            lock.lock();
            // 执行业务逻辑
            lock.unlock();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}