【从入门到放弃-ZooKeeper】ZooKeeper实战-分布式锁

前言

上文【从入门到放弃-ZooKeeper】ZooKeeper实战-分布式队列中，我们一起写了下如何通过ZooKeeper的持久性顺序节点实现一个分布式队列。
本文我们来一起写一个ZooKeeper的实现的分布式锁。

设计

参考之前学习的【从入门到放弃-Java】并发编程-JUC-locks-ReentrantLock，实现java.util.concurrent.locks.Lock接口。
我们通过重写接口中的方法实现一个可重入锁。

• lock：请求锁，如果成功则直接返回，不成功则阻塞 直到获取锁。
• lockInterruptibly：请求锁，如果失败则一直阻塞等待 直到获取锁或线程中断
• tryLock：1、尝试获取锁，获取失败的话 直接返回false，不会再等待。2、尝试获取锁，获取成功返回true，否则一直请求，直到超时返回false
• unlock：释放锁

我们使用ZooKeeper的EPHEMERAL临时节点机制，如果能创建成功的话，则获取锁成功，释放锁或客户端断开连接后，临时节点自动删除，这样可以避免误删除或漏删除的情况。

获取锁失败后，这里我们使用轮询的方式来不断尝试创建。其实应该使用Watcher机制来实现，这样能避免大量的无用请求。在下一节更优雅的分布式锁实现机制中我们会用到。

DistributedLock

public class DistributedLock implements Lock {
    private static Logger logger = LoggerFactory.getLogger(DistributedQueue.class);

    //ZooKeeper客户端，进行ZooKeeper操作
    private ZooKeeper zooKeeper;

    //根节点名称
    private String dir;

    //加锁节点
    private String node;

    //ZooKeeper鉴权信息
    private List<ACL> acls;

    //要加锁节点
    private String fullPath;

    //加锁标识，为0时表示未获取到锁，每获取一次锁则加一，释放锁时减一。减到0时断开连接，删除临时节点。
    private volatile int state;

    /**
     * Constructor.
     *
     * @param zooKeeper the zoo keeper
     * @param dir       the dir
     * @param node      the node
     * @param acls      the acls
     */
    public DistributedLock(ZooKeeper zooKeeper, String dir, String node, List<ACL> acls) {
        this.zooKeeper = zooKeeper;
        this.dir = dir;
        this.node = node;
        this.acls = acls;
        this.fullPath = dir.concat("/").concat(node);
        init();
    }

    private void init() {
        try {
            Stat stat = zooKeeper.exists(dir, false);
            if (stat == null) {
                zooKeeper.create(dir, null, acls, CreateMode.PERSISTENT);
            }
        } catch (Exception e) {
            logger.error("[DistributedLock#init] error : " + e.toString(), e);
        }
    }
}

lock

public void lock() {
    //通过state实现重入机制，如果已经获取锁，则将state++即可。
    if (addLockCount()) {
        return;
    }
    //一直尝试获取锁，知道获取成功
    for (;;) {
        try {
            //创建临时节点
            zooKeeper.create(fullPath, null, acls, CreateMode.EPHEMERAL);
            //第一次获取锁，state++，这里不需要使用加锁机制保证原子性，因为同一时间，最多只有一个线程能create节点成功。
            state++;
            break;
        } catch (InterruptedException ie) {
            //如果捕获中断异常，则设置当前线程为中断状态
            logger.error("[DistributedLock#lock] error : " + ie.toString(), ie);
            Thread.currentThread().interrupt();
        } catch (KeeperException ke) {
            //如果捕获到的异常是 节点已存在 外的其他异常，则设置当前线程为中断状态
            logger.error("[DistributedLock#lock] error : " + ke.toString(), ke);
            if (!KeeperException.Code.NODEEXISTS.equals(ke.code())) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

lockInterruptibly

public void lockInterruptibly() throws InterruptedException {
    //通过state实现重入机制，如果已经获取锁，则将state++即可。
    if (addLockCount()) {
        return;
    }
    for (;;) {
        //如果当前线程为中断状态，则抛出中断异常
        if (Thread.interrupted()) {
            throw new InterruptedException();
        }
        try {
            zooKeeper.create(fullPath, null, acls, CreateMode.EPHEMERAL);
            state++;
            break;
        } catch (InterruptedException ie) {
            //如果捕获中断异常，则设置当前线程为中断状态
            logger.error("[DistributedLock#lockInterruptibly] error : " + ie.toString(), ie);
            Thread.currentThread().interrupt();
        } catch (KeeperException ke) {
            //如果捕获到的异常是 节点已存在 外的其他异常，则设置当前线程为中断状态
            logger.error("[DistributedLock#lockInterruptibly] error : " + ke.toString(), ke);
            if (!KeeperException.Code.NODEEXISTS.equals(ke.code())) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

tryLock

public boolean tryLock() {
    //通过state实现重入机制，如果已经获取锁，则将state++即可。
    if (addLockCount()) {
        return true;
    }
    //如果获取成功则返回true，失败则返回false
    try {
        zooKeeper.create(fullPath, null, acls, CreateMode.EPHEMERAL);
        state++;
        return true;
    } catch (Exception e) {
        logger.error("[DistributedLock#tryLock] error : " + e.toString(), e);
    }

    return false;
}


public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
    //通过state实现重入机制，如果已经获取锁，则将state++即可。
    if (addLockCount()) {
        return true;
    }

    //如果尝试获取超时，则返回false
    long nanosTimeout = unit.toNanos(time);
    if (nanosTimeout <= 0L) {
        return false;
    }

    final long deadline = System.nanoTime() + nanosTimeout;
    for (;;) {
        //如果当前线程为中断状态，则抛出中断异常
        if (Thread.interrupted()) {
            throw new InterruptedException();
        }

        //如果尝试获取超时，则返回false
        nanosTimeout = deadline - System.nanoTime();
        if (nanosTimeout <= 0L) {
            return false;
        }
        try {
            zooKeeper.create(fullPath, null, acls, CreateMode.EPHEMERAL);
            state++;
            return true;
        } catch (InterruptedException ie) {
            //如果捕获中断异常，则返回false
            logger.error("[DistributedLock#tryLock] error : " + ie.toString(), ie);
            return false;
        } catch (KeeperException ke) {
            //如果捕获到的异常是 节点已存在 外的其他异常，则返回false
            logger.error("[DistributedLock#tryLock] error : " + ke.toString(), ke);
            if (!KeeperException.Code.NODEEXISTS.equals(ke.code())) {
                return false;
            }
        }
    }
}

unlock

public void unlock() {
    //通过state实现重入机制，如果已经获取锁，释放锁时，需要将state--。
    delLockCount();
    
    //如果state为0时，说明不再持有锁，需要将连接关闭，自动删除临时节点
    if (state == 0 && zooKeeper != null) {
        try {
            zooKeeper.close();
        } catch (InterruptedException e) {
            logger.error("[DistributedLock#unlock] error : " + e.toString(), e);
        }
    }
}

addLockCount

private boolean addLockCount() {
    //如果state大于0，即已持有锁，将state数量加一
    if (state > 0) {
        synchronized (this) {
            if (state > 0) {
                state++;
                return true;
            }
        }
    }
    return false;
}

delLockCount

private boolean delLockCount() {
    //如果state大于0，即还持有锁，将state数量减一
    if (state > 0) {
        synchronized (this) {
            if (state > 0) {
                state--;
                return true;
            }
        }
    }
    return false;
}

总结

上面就是一个通过ZooKeeper实现的分布式可重入锁，利用了临时节点的特性。源代码可见：aloofJr
其中有几个可以优化的点。

• 轮询的方式换成Watcher机制
• 可重入锁实现方式的优化
• 所有线程竞争一个节点的创建，容易出现羊群效应，且是一种不公平的锁竞争模式

下节我们使用新的方式实现分布式锁来解决上面的几个问题，如果大家好的优化建议，欢迎一起讨论。

更多文章

见我的博客：https://nc2era.com

written by AloofJr，转载请注明出处