ZooKeeper 实现分布式锁

2,065 阅读6分钟
原文链接: yiqiwuliao.com

ZooKeeper 实现分布式锁

利用 Watcher 机制和 ZooKeeper EPHEMERAL_SEQUENTIAL 节点的特点,实现分布式锁。

实现原理:
EPHEMERAL_SEQUENTIAL 该类节点具有顺序递增特点,不会持久化到磁盘,在线程执行完毕后,会自动删除。

Watcher 机制使得在节点被删除时,能够获得通知,并且能接收到被 push 过来的消息。

分布式锁:在多进程环境下,公共的资源可能会被不同的进程占用,为了防止多个线程同时更改数据,需要为该资源加锁,当一个线程占用后,其他线程必须得等待。

具体实现思路为:
dis_lock_program

实现代码:

public class ZooDistributeLock implements Watcher {

    private static final Logger LOG = LoggerFactory.getLogger(ZooDistributeLock.class);

    private static final String LOCK_PATH = "/lock";

    // 模拟开启的线程数
    private static final int THREAD_NUM = 5;

    // 用于等待所有线程都连接成功后再执行任务
    private static CountDownLatch startFlag = new CountDownLatch(1);

    // 用于确保所有线程执行完毕
    private static CountDownLatch threadFlag = new CountDownLatch(THREAD_NUM);

    private ZooKeeper zk = null;

    private String currentPath;

    private String lockPath;

    public static void main(String[] args) {
        for (int i = 0; i < THREAD_NUM; i++) {
            new Thread() {
                @Override
                public void run() {
                    ZooDistributeLock zooDistributeLock = new ZooDistributeLock();
                    try {
                        zooDistributeLock.connection();
                        zooDistributeLock.createNode();
                        zooDistributeLock.getLock();
                    } catch (IOException | InterruptedException | KeeperException e) {
                        e.printStackTrace();
                    }
                }
            }.start();
        }
        try {
            threadFlag.await();
            LOG.info("所有线程执行完毕...");
        } catch (InterruptedException e) {
            LOG.error(e.getMessage(), e);
        }
    }

    @Override
    public void process(WatchedEvent event) {

        Event.KeeperState state = event.getState();
        Event.EventType type = event.getType();

        if (Event.KeeperState.SyncConnected == state) {
            if (Event.EventType.None == type) {
                // 标识连接成功
                LOG.info("成功连接上ZK服务器");
                startFlag.countDown();
            }

            if (Event.EventType.NodeDeleted == type && event.getPath().equals(this.lockPath)) {
                LOG.info("节点" + this.lockPath + "的锁已经被释放");
                try {
                    // 上一个节点释放了,当前节点去获取锁
                    getLock();
                } catch (KeeperException | InterruptedException e) {
                    LOG.error(e.getMessage(), e);
                }
            }
        }

    }

    /**
     * 连接到 ZK
     *
     * @throws IOException
     */
    private void connection() throws IOException, InterruptedException {

        zk = new ZooKeeper("127.0.0.1:2181", 5000, this);

        // 等待连接成功后再执行下一步操作
        startFlag.await();
    }

    // 创建节点,并初始化当前路径
    private void createNode() throws KeeperException, InterruptedException, UnsupportedEncodingException {
        this.currentPath = this.zk.create(LOCK_PATH, "".getBytes("UTF-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode
                .EPHEMERAL_SEQUENTIAL);
    }

    private void getLock() throws KeeperException, InterruptedException {
        if (minNode()) {
            doSomething();
            // 释放锁
            releaseLock();
        }
    }

    /**
     * 当前是否为最小节点
     *
     * @return
     */
    private boolean minNode() {

        // 当前序号
        try {
            initLockPath();
            // 判断前一个节点存在不存在,如果存在,则表示当前节点不是最小节点
            // zk.getData(this.lockPath, this, new Stat());
            zk.getData(this.lockPath, true, new Stat());
            LOG.info(this.currentPath + " 不是最小值,没有获取锁,等待 " + this.lockPath + " 释放锁");
            return false;
        } catch (KeeperException e) {
            LOG.info(this.currentPath + " 是最小值,获得锁");
            return true;
        } catch (InterruptedException e) {
            LOG.error(e.getMessage(), e);
        }
        return true;
    }

    private void doSomething() {
        LOG.info("处理业务逻辑...");
    }

    /**
     * 释放锁并关闭连接
     *
     * @throws KeeperException
     * @throws InterruptedException
     */
    private void releaseLock() throws KeeperException, InterruptedException {
        Thread.sleep(2000);
        if (this.zk != null) {
            LOG.info(this.currentPath + " 业务处理完毕,释放锁...");
            zk.delete(this.currentPath, -1);
            this.zk.close();
            LOG.info(Thread.currentThread().getName() + "关闭 zookeeper 连接");
        }
        threadFlag.countDown();
    }

    /**
     * 初始化 lockpath
     */
    private void initLockPath() {

        int currentSeq = Integer.parseInt(this.currentPath.substring(LOCK_PATH.length()));

        // 上一个序号
        int preSeq = currentSeq - 1;

        String preSeqStr = String.valueOf(preSeq);
        while (preSeqStr.length() < 10) {
            preSeqStr = "0" + preSeqStr;
        }
        this.lockPath = LOCK_PATH + preSeqStr;
    }

    @Override
    public String toString() {
        return "ZooDistributeLock{" +
                "zk=" + zk +
                ", currentPath='" + currentPath + '\'' +
                ", lockPath='" + lockPath + '\'' +
                '}';
    }
}

控制台打印结果:

2017-07-10 14:53:13  [ Thread-0-EventThread:60 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:73) 成功连接上ZK服务器
2017-07-10 14:53:13  [ Thread-3-EventThread:60 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:73) 成功连接上ZK服务器
2017-07-10 14:53:13  [ Thread-4-EventThread:60 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:73) 成功连接上ZK服务器
2017-07-10 14:53:13  [ Thread-2-EventThread:60 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:73) 成功连接上ZK服务器
2017-07-10 14:53:13  [ Thread-1-EventThread:60 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:73) 成功连接上ZK服务器
2017-07-10 14:53:13  [ Thread-1:78 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:130) /lock0000000330 不是最小值,没有获取锁,等待 /lock0000000329 释放锁
2017-07-10 14:53:13  [ Thread-3:78 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:130) /lock0000000329 不是最小值,没有获取锁,等待 /lock0000000328 释放锁
2017-07-10 14:53:13  [ Thread-2:78 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:130) /lock0000000328 不是最小值,没有获取锁,等待 /lock0000000327 释放锁
2017-07-10 14:53:13  [ Thread-4:78 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:130) /lock0000000331 不是最小值,没有获取锁,等待 /lock0000000330 释放锁
2017-07-10 14:53:13  [ Thread-0:80 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:133) /lock0000000327 是最小值,获得锁
2017-07-10 14:53:13  [ Thread-0:80 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.doSomething(ZooDistributeLock.java:142) 处理业务逻辑...
2017-07-10 14:53:15  [ Thread-0:2082 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:154) /lock0000000327 业务处理完毕,释放锁...
2017-07-10 14:53:15  [ Thread-0:2087 ] - [ INFO ] org.apache.zookeeper.ZooKeeper.close(ZooKeeper.java:684) Session: 0x15d1c5a431e018b closed
2017-07-10 14:53:15  [ Thread-2-EventThread:2088 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:78) 节点/lock0000000327的锁已经被释放
2017-07-10 14:53:15  [ Thread-0:2088 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:157) Thread-0关闭 zookeeper 连接
2017-07-10 14:53:15  [ Thread-2-EventThread:2089 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:133) /lock0000000328 是最小值,获得锁
2017-07-10 14:53:15  [ Thread-2-EventThread:2089 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.doSomething(ZooDistributeLock.java:142) 处理业务逻辑...
2017-07-10 14:53:15  [ Thread-0-EventThread:2090 ] - [ INFO ] org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:519) EventThread shut down for session: 0x15d1c5a431e018b
2017-07-10 14:53:17  [ Thread-2-EventThread:4094 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:154) /lock0000000328 业务处理完毕,释放锁...
2017-07-10 14:53:17  [ Thread-3-EventThread:4097 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:78) 节点/lock0000000328的锁已经被释放
2017-07-10 14:53:17  [ Thread-2-EventThread:4098 ] - [ INFO ] org.apache.zookeeper.ZooKeeper.close(ZooKeeper.java:684) Session: 0x15d1c5a431e018c closed
2017-07-10 14:53:17  [ Thread-3-EventThread:4098 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:133) /lock0000000329 是最小值,获得锁
2017-07-10 14:53:17  [ Thread-3-EventThread:4098 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.doSomething(ZooDistributeLock.java:142) 处理业务逻辑...
2017-07-10 14:53:17  [ Thread-2-EventThread:4098 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:157) Thread-2-EventThread关闭 zookeeper 连接
2017-07-10 14:53:17  [ Thread-2-EventThread:4098 ] - [ INFO ] org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:519) EventThread shut down for session: 0x15d1c5a431e018c
2017-07-10 14:53:19  [ Thread-3-EventThread:6103 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:154) /lock0000000329 业务处理完毕,释放锁...
2017-07-10 14:53:19  [ Thread-1-EventThread:6106 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:78) 节点/lock0000000329的锁已经被释放
2017-07-10 14:53:19  [ Thread-3-EventThread:6107 ] - [ INFO ] org.apache.zookeeper.ZooKeeper.close(ZooKeeper.java:684) Session: 0x15d1c5a431e0189 closed
2017-07-10 14:53:19  [ Thread-1-EventThread:6107 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:133) /lock0000000330 是最小值,获得锁
2017-07-10 14:53:19  [ Thread-3-EventThread:6107 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:157) Thread-3-EventThread关闭 zookeeper 连接
2017-07-10 14:53:19  [ Thread-1-EventThread:6108 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.doSomething(ZooDistributeLock.java:142) 处理业务逻辑...
2017-07-10 14:53:19  [ Thread-3-EventThread:6108 ] - [ INFO ] org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:519) EventThread shut down for session: 0x15d1c5a431e0189
2017-07-10 14:53:21  [ Thread-1-EventThread:8111 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:154) /lock0000000330 业务处理完毕,释放锁...
2017-07-10 14:53:21  [ Thread-4-EventThread:8115 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:78) 节点/lock0000000330的锁已经被释放
2017-07-10 14:53:21  [ Thread-1-EventThread:8116 ] - [ INFO ] org.apache.zookeeper.ZooKeeper.close(ZooKeeper.java:684) Session: 0x15d1c5a431e018a closed
2017-07-10 14:53:21  [ Thread-1-EventThread:8116 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:157) Thread-1-EventThread关闭 zookeeper 连接
2017-07-10 14:53:21  [ Thread-4-EventThread:8116 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:133) /lock0000000331 是最小值,获得锁
2017-07-10 14:53:21  [ Thread-4-EventThread:8116 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.doSomething(ZooDistributeLock.java:142) 处理业务逻辑...
2017-07-10 14:53:21  [ Thread-1-EventThread:8116 ] - [ INFO ] org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:519) EventThread shut down for session: 0x15d1c5a431e018a
2017-07-10 14:53:23  [ Thread-4-EventThread:10119 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:154) /lock0000000331 业务处理完毕,释放锁...
2017-07-10 14:53:23  [ Thread-4-EventThread:10123 ] - [ INFO ] org.apache.zookeeper.ZooKeeper.close(ZooKeeper.java:684) Session: 0x15d1c5a431e0188 closed
2017-07-10 14:53:23  [ Thread-4-EventThread:10124 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:157) Thread-4-EventThread关闭 zookeeper 连接
2017-07-10 14:53:23  [ Thread-4-EventThread:10124 ] - [ INFO ] org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:519) EventThread shut down for session: 0x15d1c5a431e0188
2017-07-10 14:53:23  [ main:10124 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.main(ZooDistributeLock.java:58) 所有线程执行完毕...