SpringBoot2.3整合Zookeeper3.6实现分布式锁

小半 • 2023年1月16日下午3:05 • 技术随笔 • 阅读 212

在分布式项目中经常会使用到分布式锁，常见的分布式锁实现方案有Redis和Zookeeper，本文主要介绍如何使用Zookeeper实现分布式锁。

1. Zookeeper实现分布式锁原理

2. 代码实现

public class DistributedLock {

    @Value("${zookeeper.connect.address}")
    private String connectStr;
    @Value("${zookeeper.connect.time-out}")
    private int timeOut;
    private ZooKeeper zkClient;
    private String rootNode = "locks";
    private String subNode = "seq-";
    private String waitPath;
    private CountDownLatch connectLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);
    private String currentNode;

    public DistributedLock() throws KeeperException, InterruptedException, IOException {
        zkClient = new ZooKeeper(connectStr, timeOut, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    connectLatch.countDown();
                }
                if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
                    waitLatch.countDown();
                }
            }
        });
        connectLatch.await();
        //获取根节点状态
        Stat exists = zkClient.exists("/" + rootNode, false);
        //如果根节点不存在，则创建永久根节点
        if (exists == null) {
            zkClient.create("/" + rootNode, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    public void zkLock() {
        try {
            currentNode = zkClient.create("/" + rootNode + "/" + subNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            List<String> childrenNodes = zkClient.getChildren("/" + rootNode, false);
            //列表只有一个子节点，说明client获得锁
            if (childrenNodes.size() == 1) {
                return;
            } else {
                Collections.sort(childrenNodes);
                String thisNode = currentNode.substring(("/" + rootNode + "/").length());
                int index = childrenNodes.indexOf(thisNode);
                if (index == -1) {
                    //数据异常
                } else if (index == 0) {
                    //当前client获得锁
                    return;
                } else {
                    this.waitPath = "/" + rootNode + "/" + childrenNodes.get(index - 1);
                    zkClient.getData(waitPath, true, new Stat());
                    waitLatch.await();
                    return;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void zkUnlock() {
        try {
            zkClient.delete(this.currentNode, -1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}