Java 中如何使用 ZooKeeper 实现分布式锁？

推荐答案

// www.javascriptcn.com code example
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.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 LOCK_ROOT_PATH = "/locks";
    private static final String LOCK_NODE_NAME = "lock_";
    private ZooKeeper zooKeeper;
    private String lockPath;

    public DistributedLock(String zkAddress) throws IOException, InterruptedException, KeeperException {
        CountDownLatch connectedSignal = new CountDownLatch(1);
        this.zooKeeper = new ZooKeeper(zkAddress, 30000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    connectedSignal.countDown();
                }
            }
        });
        connectedSignal.await();

        // Ensure the root path exists
        Stat stat = zooKeeper.exists(LOCK_ROOT_PATH, false);
        if (stat == null) {
            zooKeeper.create(LOCK_ROOT_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    public void acquireLock() throws KeeperException, InterruptedException {
        // Create an ephemeral sequential node
        lockPath = zooKeeper.create(LOCK_ROOT_PATH + "/" + LOCK_NODE_NAME, new byte[0],
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        // Try to acquire the lock
        while (true) {
            List<String> children = zooKeeper.getChildren(LOCK_ROOT_PATH, false);
            Collections.sort(children);
            String smallestChild = children.get(0);

            if (lockPath.endsWith(smallestChild)) {
                // Acquired the lock
                return;
            } else {
                // Watch the previous node
                String previousNode = children.get(Collections.binarySearch(children, lockPath.substring(LOCK_ROOT_PATH.length() + 1)) - 1);
                final CountDownLatch latch = new CountDownLatch(1);
                Stat stat = zooKeeper.exists(LOCK_ROOT_PATH + "/" + previousNode, new Watcher() {
                    @Override
                    public void process(WatchedEvent event) {
                        if (event.getType() == Event.EventType.NodeDeleted) {
                            latch.countDown();
                        }
                    }
                });

                if (stat != null) {
                    latch.await();
                }
            }
        }
    }

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

    public void close() throws InterruptedException {
        zooKeeper.close();
    }
}

本题详细解读

1. ZooKeeper 分布式锁的基本原理

ZooKeeper 是一个分布式协调服务，可以用来实现分布式锁。其基本原理是通过在 ZooKeeper 中创建一个临时顺序节点，每个节点代表一个锁请求。锁的获取顺序由节点的创建顺序决定，最小的节点获得锁。当锁释放时，删除对应的节点，下一个最小的节点将获得锁。

2. 实现步骤

1. 连接 ZooKeeper：首先需要连接到 ZooKeeper 服务器，并确保锁的根路径存在。

2. 创建临时顺序节点：每个客户端在尝试获取锁时，会在锁的根路径下创建一个临时顺序节点。

3. 获取锁：客户端检查自己创建的节点是否是最小的节点。如果是，则获取锁；如果不是，则监听前一个节点的删除事件。

4. 释放锁：客户端释放锁时，删除自己创建的节点。

5. 关闭连接：在不再需要锁时，关闭与 ZooKeeper 的连接。

3. 代码解析

• ZooKeeper 连接：在构造函数中，通过 ZooKeeper 类连接到 ZooKeeper 服务器，并确保锁的根路径存在。

• 创建临时顺序节点：在 acquireLock 方法中，使用 create 方法创建一个临时顺序节点。

• 获取锁：通过 getChildren 方法获取所有子节点，并排序。如果当前节点是最小的节点，则获取锁；否则，监听前一个节点的删除事件。

• 释放锁：在 releaseLock 方法中，删除当前节点以释放锁。

• 关闭连接：在 close 方法中，关闭与 ZooKeeper 的连接。

4. 注意事项

• 临时节点：使用临时节点可以确保客户端断开连接时，锁会自动释放。

• 顺序节点：顺序节点确保了锁的公平性，先到先得。

• 监听机制：通过监听前一个节点的删除事件，避免了轮询带来的性能开销。

通过以上步骤和代码实现，可以在 Java 中使用 ZooKeeper 实现一个简单的分布式锁。