分布式锁实践（二）-ZooKeeper实现总结

写在最前面

前几周写了篇 利用Redis实现分布式锁 ，今天简单总结下ZooKeeper实现分布式锁的过程。其实生产上我只用过Redis或者数据库的方式，之前还真没了解过ZooKeeper怎么实现分布式锁。这周简单写了个小Demo，更坚定了我继续使用Redis的信心了。

ZooKeeper分布式锁的实现原理

在分布式解决方案中，Zookeeper是一个分布式协调工具。当多个JVM客户端，同时在ZooKeeper上创建相同的一个临时节点，因为临时节点路径是保证唯一，只要谁能够创建节点成功，谁就能够获取到锁。没有创建成功节点，就会进行等待，当释放锁的时候，采用事件通知给客户端重新获取锁资源。如果请求超时直接返回给客户端超时，重新请求即可。

代码实现

为了更好的展现效果，我这里设置每个线程请求需要1s，请求超时时间为30s。

 首先我们先写一个测试类，模拟多线程多客户端请求的情况：

public class ZkLockTest implements Runnable {
    private ZkLock zkLock = new ZkDistributedLock();

    public void run() {
        try {
            if (zkLock.getLock((long)30000,null)) {
                System.out.println("线程:" + Thread.currentThread().getName() + ",抢购成功:" + System.currentTimeMillis());
            } else {
                System.out.println("线程:" + Thread.currentThread().getName() + ",抢购超时失败请重试:" + System.currentTimeMillis());
            }
            Thread.sleep(1000);
        } catch (Exception e) {

        } finally {
            zkLock.unLock();
        }
    }

    public static void main(String[] args) {
        System.out.println("zk分布式锁开始。。");
        for (int i = 0; i < 100; i++) {
            new Thread(new ZkLockTest()).start();
        }
    }

}

模拟100个线程，去同时争夺锁。当然上述写法 100个线程不会同时启动，如果需要的话可以用信号量的形式控制。

其次，写一个锁的接口

public interface ZkLock {

    // 获取锁
    Boolean getLock(Long acquireTimeout,Long endTime);

    // 释放锁
    void unLock();
}

这里我定义了两个接口，分别对应获取锁和释放锁。

在获取锁中有两个参数，含义分别为锁超时时间和最终计算的超时时间，具体看下文代码就懂了。

public class ZkDistributedLock implements ZkLock {
    // 集群连接地址
    private String CONNECTION = "127.0.0.1:2181";
    // zk客户端连接
    private ZkClient zkClient = new ZkClient(CONNECTION);
    // path路径
    private String lockPath = "/lock";
    private CountDownLatch countDownLatch;
    //请求设置的超时时间：acquireTimeout 毫秒。最终超时时间endTime
    public Boolean getLock(Long acquireTimeout,Long endTime) {
        Boolean lock = false;
        if (endTime == null) {
            //等待超时时间
            endTime = System.currentTimeMillis() + acquireTimeout;
        }
        if (tryLock()) {
            System.out.println("####获取锁成功######");
            lock = true;
        } else {
            if (waitLock(endTime)) {
                if (getLock(null,endTime)) {
                    lock = true;
                }
            }
        }
       return lock;
    }

    public void unLock() {
        if (zkClient != null) {
            System.out.println("#######释放锁#########");
            zkClient.close();
        }
    }
    
    private boolean tryLock() {
        try {
            zkClient.createEphemeral(lockPath);
            return true;
        } catch (Exception e) {
            
            return false;
        }

    }

    private Boolean waitLock(Long endTime) {
       // System.out.println("进入等待");
        // 使用zk临时事件监听
        IZkDataListener iZkDataListener = null;
        try {
            // 使用zk临时事件监听
            iZkDataListener = new IZkDataListener() {
                
                public void handleDataDeleted(String path) throws Exception {
                    if (countDownLatch != null) {
                        countDownLatch.countDown();
                    }
                }
                public void handleDataChange(String arg0, Object arg1) throws Exception {
                    
                }
            };
            // 注册事件通知
            zkClient.subscribeDataChanges(lockPath, iZkDataListener);
            if (System.currentTimeMillis() < endTime) {
                if (zkClient.exists(lockPath)) {
                    countDownLatch = new CountDownLatch(1);
                    try {
                        countDownLatch.await();
                        return true;
                    } catch (Exception e) {
                        
                    }
                } else {
                    return true;
                }
            } else {
                System.out.println("超时返回");
            }
        } catch (Exception e) {
            
        } finally {
            // 监听完毕后，移除事件通知
            zkClient.unsubscribeDataChanges(lockPath, iZkDataListener);
        }
        return false;
    }

}

这个类是我实现zk锁的核心类，和上文原理图中类似。首先用户请求的时候需要获取锁，第一个争夺到锁的用户执行相关逻辑后释放锁，在这个过程中如果程序出错断开连接，因为临时节点的缘故，节点也会自动删除释放锁的。

另外就是其他争夺锁失败的用户，我这里设置了一定的等待时间，当在时间内原锁释放，还是可以重新去获取锁的。这里要说下锁释放的监听，在原生的zookeeper中，使用watcher需要每次先注册，而且使用一次就需要注册一次。而在zkClient中，没有注册watcher的必要，而是引入了listener的概念，即只要client在某一个节点中注册了listener，只要服务端发生变化，就会通知当前注册listener的客户端。我这里使用的是IZkDataListener，这个类是zkClient提供的一个接口，它可以在当前节点数据内容或版本发生变化或者当前节点被删除时触发。

触发后我们就可以重新去争夺锁，当再次争夺失败进入等待时会再次检测当前请求是否超时。

下面我们来看下上述代码的实现效果：

zk分布式锁开始。。
####获取锁成功######
线程:Thread-3,抢购成功:1544183770509
#######释放锁#########
####获取锁成功######
线程:Thread-81,抢购成功:1544183771555
#######释放锁#########

.........

超时返回
线程:Thread-11,抢购超时失败请重试:1544183800677
超时返回
线程:Thread-1,抢购超时失败请重试:1544183800681
#######释放锁#########
#######释放锁#########
####获取锁成功######
线程:Thread-49,抢购成功:1544183801710
超时返回
线程:Thread-25,抢购超时失败请重试:1544183801729
超时返回
#######释放锁#########
#######释放锁#########

 释放锁说的可能并不准确，应该说是关闭连接，有些线程实际上是没有得到锁的。

简单尝试了下zk实现分布式锁的方式，当然上述代码如果应用到生产中肯定问题还是不少的，因为兴趣点不在这，就不仔细研究了。简单来说，相比其他方式实现步骤更为复杂，感觉更容易出问题。

总结

经过三种方式的应用和简单实践，总结实现分布式锁三种方式的优缺点如下

1、数据库实现：

优点，实现简单只是for update的显示加锁。缺点，性能问题较大，而且本身系统在设计时是需要尽量减轻数据库的压力的。

2、Redis实现：

优点：一般互联网项目都会集成，本身是nosql数据库，缓存实现简单，高并发应付自如，同时新版的Jedis完美解决了以往程序出错，未设置超时时间死锁的问题。

缺点：网络问题可能会引起锁删除失败，超时时间有一定的延迟。

3、ZooKeeper实现：

优点：Zookeeper临时节点先天可控的有效期设置，避免了程序引发的死锁问题

缺点：实现过于繁杂，相比其他两种写法更容易出问题，另外还需要单独维护zk。

结论：

我个人更为推荐Redis的实现方式，实现简单，性能也比较好，同时引入集群可以提高可用性。Jedis多参的设置方式也较好的保证了有效期的控制和死锁的问题。