目录
疯狂创客圈 Java 分布式聊天室【 亿级流量】实战系列之 -26【 】
写在前面
大家好，我是作者尼恩。目前和几个小伙伴一起，组织了一个高并发的实战社群【疯狂创客圈】。正在开始高并发、亿级流程的 IM 聊天程序 学习和实战
前面，已经完成一个高性能的 Java 聊天程序的四件大事：
接下来，需要进入到分布式开发的环节了。 分布式的中间件，疯狂创客圈的小伙伴们，一致的选择了zookeeper，不仅仅是由于其在大数据领域，太有名了。更重要的是，很多的著名框架，都使用了zk。
本篇介绍 ZK Curator 的分布式锁实现。
1.1. 分布式锁 简介
在我们进行单机应用开发，涉及并发同步的时候，我们往往采用synchronized或者Lock的方式来解决多线程间的代码同步问题。但当我们的应用是分布式集群工作的情况下，那么就需要一种更加高级的锁机制，来处理种跨机器的进程之间的数据同步问题。
这就是分布式锁。
1.1.1. 图解：公平锁和可重入锁 模型
分布式锁的概念和原理，比较抽象难懂。如果用一个简单的故事来类比，估计就简单多了。
很久以前，在一个村子有一口井，水质非常的好，村民们都抢着取井里的水。井就那么一口，村里的人很多，村民为争抢取水打架斗殴，甚至头破血流。
问题总是要解决，于是村长绞尽脑汁，最终想出了一个凭号取水的方案。井边安排一个看井人，维护取水的秩序。
说起来，秩序很简单，取水之前，先取号。号排在前面的，就可以先取水。先到的排在前面，那些后到的，没有排在最前面的人，一个一个挨着，在井边排成一队。取水示意图如下 ：
这种排队取水模型，就是一种锁的模型。排在最前面的号，拥有取水权，就是一种典型的独占锁。另外，先到先得，号排在前面的人先取到水，取水之后就轮到下一个号取水，至少，看起来挺公平的，说明它是一种公平锁。
在公平独占锁的基础上，再进一步，看看可重入锁的模型。
假定，取水时以家庭为单位，哪个家庭任何人拿到号，就可以排号取水，而且如果一个家庭有一个人拿到号，其它家人这时候过来打水不用再取号。新的排号取水示意图如下 ：
如上图的1号，老公有号，他的老婆来了，直接排第一个，妻凭夫贵。再看上图的2号，父亲正在打水，他的儿子和女儿也到井边了，直接排第二个，这个叫做子凭父贵。 等等，如果是同一个家庭，可以直接复用排号，不用重新取号从后面排起。
以上这个故事模型，就是可以重入锁的模型。只要满足条件，同一个排号，可以用来多次取水。在锁的模型中，相当于一把锁，可以被多次锁定，这就叫做可重入锁。
1.1.2. 图解： zookeeper分布式锁的原理
理解了锁的原理后，就会发现，Zookeeper 天生就是一副分布式锁的胚子。
首先，Zookeeper的每一个节点，都是一个天然的顺序发号器。
在每一个节点下面创建子节点时，只要选择的创建类型是有序（EPHEMERAL_SEQUENTIAL 临时有序或者PERSISTENT_SEQUENTIAL 永久有序）类型，那么，新的子节点后面，会加上一个次序编号。这个次序编号，是上一个生成的次序编号加一
比如，创建一个用于发号的节点“/test/lock”，然后以他为父亲节点，可以在这个父节点下面创建相同前缀的子节点，假定相同的前缀为“/test/lock/seq-”，在创建子节点时，同时指明是有序类型。如果是第一个创建的子节点，那么生成的子节点为/test/lock/seq-0000000000，下一个节点则为/test/lock/seq-0000000001，依次类推，等等。
其次，Zookeeper节点的递增性，可以规定节点编号最小的那个获得锁。
一个zookeeper分布式锁，首先需要创建一个父节点，尽量是持久节点（PERSISTENT类型），然后每个要获得锁的线程都会在这个节点下创建个临时顺序节点，由于序号的递增性，可以规定排号最小的那个获得锁。所以，每个线程在尝试占用锁之前，首先判断自己是排号是不是当前最小，如果是，则获取锁。
第三，Zookeeper的节点监听机制，可以保障占有锁的方式有序而且高效。
每个线程抢占锁之前，先抢号创建自己的ZNode。同样，释放锁的时候，就需要删除抢号的Znode。抢号成功后，如果不是排号最小的节点，就处于等待通知的状态。等谁的通知呢？不需要其他人，只需要等前一个Znode 的通知就可以了。当前一个Znode 删除的时候，就是轮到了自己占有锁的时候。第一个通知第二个、第二个通知第三个，击鼓传花似的依次向后。
Zookeeper的节点监听机制，可以说能够非常完美的，实现这种击鼓传花似的信息传递。具体的方法是，每一个等通知的Znode节点，只需要监听linsten或者 watch 监视排号在自己前面那个，而且紧挨在自己前面的那个节点。 只要上一个节点被删除了，就进行再一次判断，看看自己是不是序号最小的那个节点，如果是，则获得锁。
为什么说Zookeeper的节点监听机制，可以说是非常完美呢？
一条龙式的首尾相接，后面监视前面，就不怕中间截断吗？比如，在分布式环境下，由于网络的原因，或者服务器挂了或则其他的原因，如果前面的那个节点没能被程序删除成功，后面的节点不就永远等待么？
其实，Zookeeper的内部机制，能保证后面的节点能够正常的监听到删除和获得锁。在创建取号节点的时候，尽量创建临时znode 节点而不是永久znode 节点，一旦这个 znode 的客户端与Zookeeper集群服务器失去联系，这个临时 znode 也将自动删除。排在它后面的那个节点，也能收到删除事件，从而获得锁。
说Zookeeper的节点监听机制，是非常完美的。还有一个原因。
Zookeeper这种首尾相接，后面监听前面的方式，可以避免羊群效应。所谓羊群效应就是每个节点挂掉，所有节点都去监听，然后做出反映，这样会给服务器带来巨大压力，所以有了临时顺序节点，当一个节点挂掉，只有它后面的那一个节点才做出反映。
1.1.3. 分布式锁的基本流程
接下来就是基于zookeeper，实现一下分布式锁。
首先定义了一个锁的接口，很简单，一个加锁方法，一个解锁方法。
/** * create by 尼恩 @ 疯狂创客圈 **/public interface Lock { boolean lock() throws Exception; boolean unlock();} 使用zookeeper实现分布式锁的算法流程，大致如下：
（1）如果锁空间的根节点不存在，首先创建Znode根节点。这里假设为“/test/lock”。这个根节点，代表了一把分布式锁。
（2）客户端如果需要占用锁，则在“/test/lock”下创建临时的且有序的子节点。
这里，尽量使一个有意义的子节点前缀，比如“/test/lock/seq-”。则第一个客户端对应的子节点为“/test/lock/seq-000000000”，第二个为 “/test/lock/seq-000000001”，以此类推。
如果前缀为“/test/lock/”，则第一个客户端对应的子节点为“/test/lock/000000000”，第二个为 “/test/lock/000000001” ，以此类推，也非常直观。
（3）客户端如果需要占用锁，还需要判断，判断自己创建的子节点是否为当前子节点列表中序号最小的子节点。如果是则认为获得锁，否则监听前一个Znode子节点变更消息，获得子节点变更通知后重复此步骤直至获得锁；
（4）获取锁后，开始处理业务流程。完成业务流程后，删除对应的子节点，完成释放锁的工作。以便后面的节点获得分布式锁。
1.1.4. 加锁的实现
lock方法的具体算法是，首先尝试着去加锁，如果加锁失败就去等待，然后再重复。
代码如下：
@Override public boolean lock() { try { boolean locked = false; locked = tryLock(); if (locked) { return true; } while (!locked) { await(); if (checkLocked()) { locked=true; } } return true; } catch (Exception e) { e.printStackTrace(); unlock(); } return false; } 尝试加锁的tryLock方法是关键。做了两件重要的事情：
（1）创建临时顺序节点，并且保存自己的节点路径
（2）判断是否是第一个，如果是第一个，则加锁成功。如果不是，就找到前一个Znode节点，并且保存其路径到prior_path。
tryLock方法代码节选如下：
private boolean tryLock() throws Exception { //创建临时Znode List waiters = getWaiters(); locked_path = ZKclient.instance .createEphemeralSeqNode(LOCK_PREFIX); if (null == locked_path) { throw new Exception("zk error"); } locked_short_path = getShorPath(locked_path); //获取等待的子节点列表，判断自己是否第一个 if (checkLocked()) { return true; } // 判断自己排第几个 int index = Collections.binarySearch(waiters, locked_short_path); if (index < 0) { // 网络抖动，获取到的子节点列表里可能已经没有自己了 throw new Exception("节点没有找到: " + locked_short_path); } //如果自己没有获得锁，则要监听前一个节点 prior_path = ZK_PATH + "/" + waiters.get(index - 1); return false; } 创建临时顺序节点后，其完整路径存放在 locked_path 成员中。另外还截取了一个后缀路径，放在 locked_short_path 成员中。 这个后缀路径，是一个短路径，只有完整路径的最后一层。在和取到的远程子节点列表中的其他路径进行比较时，需要用到短路径。因为子节点列表的路径，都是短路径，只有最后一层。
然后，调用checkLocked方法，判断是否是锁定成功。如果是则返回。如果自己没有获得锁，则要监听前一个节点。找出前一个节点的路径，保存在 prior_path 成员中，供后面的await 等待方法，去监听使用。
在进入await等待方法的介绍前，先说下checkLocked 锁定判断方法。
在checkLocked方法中，判断是否可以持有锁。判断规则很简单：当前创建的节点，是否在上一步获取到的子节点列表的第一个位置：
如果是，说明可以持有锁，返回true，表示加锁成功；
如果不是，说明有其他线程早已先持有了锁，返回false。
checkLocked方法的代码如下：
private boolean checkLocked() { //获取等待的子节点列表 List waiters = getWaiters(); //节点按照编号，升序排列 Collections.sort(waiters); // 如果是第一个，代表自己已经获得了锁 if (locked_short_path.equals(waiters.get(0))) { log.info("成功的获取分布式锁,节点为{}", locked_short_path); return true; } return false; } checkLocked方法比较简单，就是获取到所有子节点列表，并且从小到大根据节点名称进行排序，主要依靠后10位数字，因为前缀都是一样的。
排序的结果，如果自己的locked_short_path位置在第一个，代表自己已经获得了锁。
现在正式进入等待方法await的介绍。
等待方法await，表示在争夺锁失败以后的等待逻辑。那么此处该线程应该做什么呢？
private void await() throws Exception { if (null == prior_path) { throw new Exception("prior_path error"); } final CountDownLatch latch = new CountDownLatch(1); //订阅比自己次小顺序节点的删除事件 Watcher w = new Watcher() { @Override public void process(WatchedEvent watchedEvent) { System.out.println("监听到的变化 watchedEvent = " + watchedEvent); log.info("[WatchedEvent]节点删除"); latch.countDown(); } }; client.getData().usingWatcher(w).forPath(prior_path); latch.await(WAIT_TIME, TimeUnit.SECONDS); } 首先添加一个watcher监听，而监听的地址正是上面一步返回的prior_path 成员。这里，仅仅会监听自己前一个节点的变动，而不是父节点下所有节点的变动。然后，调用latch.await，进入等待状态，等到latch.countDown()被唤醒。
一旦prior_path节点发生了变动，那么就将线程从等待状态唤醒，重新一轮的锁的争夺。
至此，关于加锁的算法基本完成。但是，上面还没有实现锁的可重入。
什么是可重入呢？
只需要保障同一个线程进入加锁的代码，可以重复加锁成功即可。
修改前面的lock方法，在前面加上可重入的判断逻辑。代码如下： public boolean lock() { synchronized (this) { if (lockCount.get() == 0) { thread = Thread.currentThread(); lockCount.incrementAndGet(); } else { if (!thread.equals(Thread.currentThread())) { return false; } lockCount.incrementAndGet(); return true; } } //... } 为了变成可重入，在代码中增加了一个加锁的计数器lockCount ，计算重复加锁的次数。如果是同一个线程加锁，只需要增加次数，直接返回，表示加锁成功。
1.1.5. 释放锁的实现
释放锁主要有两个工作：
（1）减少重入锁的计数，如果不是0，直接返回，表示成功的释放了一次；
（2）如果计数器为0，移除Watchers监听器，并且删除创建的Znode临时节点；
代码如下：
@Override public boolean unlock() { if (!thread.equals(Thread.currentThread())) { return false; } int newLockCount = lockCount.decrementAndGet(); if (newLockCount < 0) { throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + locked_path); } if (newLockCount != 0) { return true; } try { if (ZKclient.instance.isNodeExist(locked_path)) { client.delete().forPath(locked_path); } } catch (Exception e) { e.printStackTrace(); return false; } return true; } 这里，为了尽量保证线程安全，可重入计数器的类型，不是int类型，而是Java并发包中的原子类型——AtomicInteger。
1.1.1. 分布式锁的应用场景
前面的实现，主要的价值是展示一下分布式锁的基础开发和原理。实际的开发中，如果需要使用到分布式锁，并不需要自己造轮子，可以直接使用curator客户端中的各种官方实现的分布式锁，比如其中的InterProcessMutex 可重入锁。
InterProcessMutex 可重入锁的使用实例如下：
@Testpublic void testzkMutex() throws InterruptedException { CuratorFramework client=ZKclient.instance.getClient(); final InterProcessMutex zkMutex = new InterProcessMutex(client,"/mutex"); ; for (int i = 0; i < 10; i++) { FutureTaskScheduler.add(() -> { try { zkMutex.acquire(); for (int j = 0; j < 10; j++) { count++; } try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } log.info("count = " + count); zkMutex.release(); } catch (Exception e) { e.printStackTrace(); } }); } Thread.sleep(Integer.MAX_VALUE);} 最后，总结一下Zookeeper分布式锁。
Zookeeper分布式锁，能有效的解决分布式问题，不可重入问题，实现起来较为简单。
但是，Zookeeper实现的分布式锁其实存在一个缺点，那就是性能并不太高。因为每次在创建锁和释放锁的过程中，都要动态创建、销毁瞬时节点来实现锁功能。ZK中创建和删除节点只能通过Leader服务器来执行，然后Leader服务器还需要将数据同不到所有的Follower机器上。
所以，在高性能，高并发的场景下，不建议使用Zk的分布式锁。
目前分布式锁，比较成熟、主流的方案是基于redis及基于zookeeper的二种方案。这两种锁，应用场景不同。而 zookeeper只是其中的一种。Zk的分布式锁的应用场景，主要高可靠，而不是太高并发的场景下。
在并发量很高，性能要求很高的场景下，推荐使用基于redis的分布式锁。
写在最后
下一篇： zookeeper + netty 实现高并发IM 聊天
疯狂创客圈 亿级流量 高并发IM 实战 系列
Java (Netty) 聊天程序【 亿级流量】实战 开源项目实战
- Netty 源码、原理、JAVA NIO 原理
- Java 面试题 一网打尽
疯狂创客圈