一、前言

在同一个jvm进程中时，可以使用JUC提供的一些锁来解决多个线程竞争同一个共享资源时候的线程安全问题，但是当多个不同机器上的不同jvm进程共同竞争同一个共享资源时候，juc包的锁就无能无力了，这时候就需要分布式锁了。常见的有使用zk的最小版本，redis的set函数，数据库锁来实现，本节我们谈谈使用zookeeper的序列节点机制来实现一个分布式锁。

二、使用zookeeper实现分布式锁

首先我们先来看看使用zk实现分布式锁的原理，在zk中是使用文件目录的格式存放节点内容，其中节点类型分为：
- 持久节点（PERSISTENT ）：节点创建后，一直存在，直到主动删除了该节点。
- 临时节点（EPHEMERAL）：生命周期和客户端会话绑定，一旦客户端会话失效，这个节点就会自动删除。
- 序列节点（SEQUENTIAL ）：多个线程创建同一个顺序节点时候，每个线程会得到一个带有编号的节点，节点编号是递增不重复的，如下图：

如上图，三个线程分别创建路径为/root/node的节点，可知在zk服务器端会在root下存在三个node节点，并且器编号唯一递增。

具体在节点创建过程中，可以混合使用，比如临时顺序节点（EPHEMERAL_SEQUENTIAL），这里我们就使用临时顺序节点来实现分布式锁。

分布式锁实现：
- 创建临时顺序节点,比如/root/node，假设返回结果为nodeId。
- 获取/root下所有孩子节点，用自己创建的nodeId的序号与所有子节点比较，看看自己是不是编号最小的。如果是最小的则就相当于获取到了锁，如果自己不是最小的，则从所有子节点里面获取比自己次小的一个节点，然后设置监听该节点的事件，然后挂起当前线程。
- 当最小编号的线程获取锁，处理完业务后删除自己对应的nodeId，删除后会激活比自己大一号的节点的线程从阻塞变为运行态，被激活的线程应该就是当前node序列号最小的了，然后就会获取到锁。

下面我们看看代码实现：

public class ZookeeperDistributedLock {
    public final static Joiner j = Joiner.on("|").useForNull("");

    //zk客户端
    private ZooKeeper zk;
    //zk是一个目录结构，root为最外层目录
    private String root = "/locks";
    //锁的名称
    private String lockName;
    //当前线程创建的序列node
    private ThreadLocal<String> nodeId = new ThreadLocal<>();
    //用来同步等待zkclient链接到了服务端
    private CountDownLatch connectedSignal = new CountDownLatch(1);
    private final static int sessionTimeout = 3000;
    private final static byte[] data= new byte[0];

    public ZookeeperDistributedLock(String config, String lockName) {
        this.lockName = lockName;

        try {
            zk = new ZooKeeper(config, sessionTimeout, new Watcher() {

                @Override
                public void process(WatchedEvent event) {
                    // 建立连接
                    if (event.getState() == KeeperState.SyncConnected) {
                        connectedSignal.countDown();
                    }
                }

            });

            connectedSignal.await();
            Stat stat = zk.exists(root, false);
            if (null == stat) {
                // 创建根节点
                zk.create(root, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    class LockWatcher implements Watcher {
        private CountDownLatch latch = null;

        public LockWatcher(CountDownLatch latch) {
            this.latch = latch;
        }

        @Override
        public void process(WatchedEvent event) {

            if (event.getType() == Event.EventType.NodeDeleted)
                latch.countDown();
        }
    }

    public void lock() {
        try {

            // 创建临时子节点
            String myNode = zk.create(root + "/" + lockName , data, ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL);

            System.out.println(j.join(Thread.currentThread().getName() + myNode, "created"));

            // 取出所有子节点
            List<String> subNodes = zk.getChildren(root, false);
            TreeSet<String> sortedNodes = new TreeSet<>();
            for(String node :subNodes) {
                sortedNodes.add(root +"/" +node);
            }

            String smallNode = sortedNodes.first();
            String preNode = sortedNodes.lower(myNode);

            if (myNode.equals( smallNode)) {
                // 如果是最小的节点,则表示取得锁
                System.out.println(j.join(Thread.currentThread().getName(), myNode, "get lock"));
                this.nodeId.set(myNode);
                return;
            }

            CountDownLatch latch = new CountDownLatch(1);
            Stat stat = zk.exists(preNode, new LockWatcher(latch));// 同时注册监听。
            // 判断比自己小一个数的节点是否存在,如果不存在则无需等待锁,同时注册监听
            if (stat != null) {
                System.out.println(j.join(Thread.currentThread().getName(), myNode,
                        " waiting for " + root + "/" + preNode + " released lock"));

                latch.await();// 等待，这里应该一直等待其他线程释放锁
                nodeId.set(myNode);
                latch = null;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

    public void unlock() {
        try {
            System.out.println(j.join(Thread.currentThread().getName(), nodeId.get(), "unlock "));
            if (null != nodeId) {
                zk.delete(nodeId.get(), -1);
            }
            nodeId.remove();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

}

ZookeeperDistributedLock的构造函数创建zkclient，并且注册了监听事件，然后调用connectedSignal.await()挂起当前线程。当zkclient链接到服务器后，会给监听器发送SyncConnected事件，监听器判断当前链接已经建立了，则调用 connectedSignal.countDown();激活当前线程，然后创建root节点。

获取锁的方法lock，内部首先创建/root/lockName的顺序临时节点，然后获取/root下所有的孩子节点，并对子节点进行排序，然后判断自己是不是最小的编号，如果是直接返回true标示获取锁成功。否者看比自己小一个号的节点是否存在，存在则注册该节点的事件，然后挂起当前线程，等待比自己小一个数的节点释放锁后发送节点删除事件，事件里面激活当前线程。

释放锁的方法unlock比较简单，就是简单的删除获取锁时候创建的节点。

三、总结

本文使用zk的临时顺序节点以及节点事件通知机制实现了一个分布式锁，大家想想是否还有优化的空间，我知道的还有一个改进点，另外一个思考，为何要使用临时节点那？期待下回分解，

最后

想了解JDK NIO和更多Netty基础的可以单击我

想了解更多关于粘包半包问题单击我
更多关于分布式系统中服务降级策略的知识可以单击 单击我
想系统学dubbo的单击我
想学并发的童鞋可以 单击我