使用Zookeeper实现分布式锁

使用Zookeeper实现分布式锁

http://blog.csdn.net/massivestars/article/details/53771532

 

实现原理

Zookeeper的一个典型应用场景就是分布式锁,锁的实现是利用Zookeeper创建的临时时序节点(创建的时候CreateMode为 EPHEMERAL_SEQUENTIAL )和节点变动的监听器实现的。时序节点保证了节点的创建在分布式系统情况下还是有先后顺序的,监听器使得客户端能感受到节点的变动情况。


具体步骤

1、创建一个永久性节点,作锁的根目录

2、当要获取一个锁时,在锁目录下创建一个临时有序列的节点

3、检查锁目录的子节点是否有序列比它小,若有则监听比它小的上一个节点,当前锁处于等待状态

4、当等待时间超过Zookeeper session的连接时间(sessionTimeout)时,当前session过期,Zookeeper自动删除此session创建的临时节点,等待状态结束,获取锁失败

5、当监听器触发时,等待状态结束,获得锁



流程图




详细代码

ZookeeperClient工具类

为了方便取得Zookeeper实例,先实行一个获取Zookeeper工具类 ZookeeperClient:
  1. package org.massive.lock;  
  2.   
  3. import org.apache.zookeeper.WatchedEvent;  
  4. import org.apache.zookeeper.Watcher;  
  5. import org.apache.zookeeper.ZooKeeper;  
  6.   
  7. import java.io.IOException;  
  8. import java.util.concurrent.CountDownLatch;  
  9.   
  10. /** 
  11.  * Created by Massive on 2016/12/18. 
  12.  */  
  13. public class ZookeeperClient {  
  14.   
  15.     private static String connectionString = "localhost:2181";  
  16.     private static int sessionTimeout = 10000;  
  17.   
  18.   
  19.     public static ZooKeeper getInstance() throws IOException, InterruptedException {  
  20.         //--------------------------------------------------------------  
  21.         // 为避免连接还未完成就执行zookeeper的get/create/exists操作引起的(KeeperErrorCode = ConnectionLoss)  
  22.         // 这里等Zookeeper的连接完成才返回实例  
  23.         //--------------------------------------------------------------  
  24.         final CountDownLatch connectedSignal = new CountDownLatch(1);  
  25.         ZooKeeper zk = new ZooKeeper(connectionString, sessionTimeout, new Watcher() {  
  26.              @Override  
  27.              public void process(WatchedEvent event) {  
  28.                  if  (event.getState()  ==  Event.KeeperState.SyncConnected) {  
  29.                      connectedSignal.countDown();  
  30.                  }  
  31.              }  
  32.          });  
  33.         connectedSignal.await(sessionTimeout,TimeUnit.MILLISECONDS);  
  34.         return zk;  
  35.     }  
  36.   
  37.     public static String getConnectionString() {  
  38.         return connectionString;  
  39.     }  
  40.   
  41.     public static void setConnectionString(String connectionString) {  
  42.         ZookeeperClient.connectionString = connectionString;  
  43.     }  
  44.   
  45.     public static int getSessionTimeout() {  
  46.         return sessionTimeout;  
  47.     }  
  48.   
  49.     public static void setSessionTimeout(int sessionTimeout) {  
  50.         ZookeeperClient.sessionTimeout = sessionTimeout;  
  51.     }  
  52.   
  53. }  

DistributedLock

  1. package org.massive.lock;  
  2.   
  3. import org.apache.commons.lang3.RandomUtils;  
  4. import org.apache.zookeeper.*;  
  5. import org.apache.zookeeper.data.Stat;  
  6.   
  7. import java.io.IOException;  
  8. import java.util.List;  
  9. import java.util.SortedSet;  
  10. import java.util.TreeSet;  
  11. import java.util.concurrent.CountDownLatch;  
  12. import java.util.concurrent.TimeUnit;  
  13.   
  14. /** 
  15.  * Created by massive on 2016-12-15. 
  16.  */  
  17. public class DistributedLock {  
  18.   
  19.     private String lockId;  
  20.   
  21.     private static final String LOCK_ROOT = "/LOCKS";  
  22.   
  23.     //--------------------------------------------------------------  
  24.     // data为存储的节点数据内容  
  25.     // 由于锁机制用的是序列功能的特性,data的值不重要,只要利于网络传输即可  
  26.     //--------------------------------------------------------------  
  27.     private final static byte[] data = {0x120x34};  
  28.   
  29.     private final CountDownLatch latch = new CountDownLatch(1);  
  30.   
  31.     private ZooKeeper zk;  
  32.   
  33.     private int sessionTimeout;  
  34.   
  35.     public DistributedLock(ZooKeeper zk,int sessionTimeout) {  
  36.         this.zk = zk;  
  37.         this.sessionTimeout = sessionTimeout;  
  38.     }  
  39.   
  40.     public DistributedLock() throws IOException, KeeperException, InterruptedException {  
  41.         this.zk = ZookeeperClient.getInstance();  
  42.         this.sessionTimeout = ZookeeperClient.getSessionTimeout();  
  43.     }  
  44.   
  45.     class LockWatcher implements Watcher {  
  46.         @Override  
  47.         public void process(WatchedEvent event) {  
  48.             //--------------------------------------------------------------  
  49.             // 监控节点变化(本程序为序列的上一节点)  
  50.             // 若为节点删除,证明序列的上一节点已删除,此时释放阀门让当前的lock获得锁  
  51.             //--------------------------------------------------------------  
  52.             if (event.getType() == Event.EventType.NodeDeleted)  
  53.                 latch.countDown();  
  54.         }  
  55.     }  
  56.   
  57.     /** 
  58.      * @return 
  59.      * @throws KeeperException 
  60.      * @throws InterruptedException 
  61.      */  
  62.     public synchronized boolean lock() {  
  63.   
  64.         //--------------------------------------------------------------  
  65.         // 保证锁根节点存在,若不存在则创建它  
  66.         //--------------------------------------------------------------  
  67.         createLockRootIfNotExists();  
  68.   
  69.         try {  
  70.   
  71.             lockId = zk.create(LOCK_ROOT + "/", data,  
  72.                     ZooDefs.Ids.OPEN_ACL_UNSAFE,  
  73.                     CreateMode.EPHEMERAL_SEQUENTIAL);  
  74.   
  75.             System.out.println("thread " + Thread.currentThread().getName() +  
  76.                     " create the lock node: " + lockId + ", trying to get lock now");  
  77.   
  78.             //--------------------------------------------------------------  
  79.             // 获得锁根节点下的各锁子节点,并排序  
  80.             //--------------------------------------------------------------  
  81.             List<String> nodes = zk.getChildren(LOCK_ROOT, true);  
  82.             SortedSet<String> sortedNode = new TreeSet<String>();  
  83.   
  84.             for (String node : nodes) {  
  85.                 sortedNode.add(LOCK_ROOT + "/" + node);  
  86.             }  
  87.   
  88.             String first = sortedNode.first();  
  89.             SortedSet<String> lessThanMe = sortedNode.headSet(lockId);  
  90.             //--------------------------------------------------------------  
  91.             // 检查是否有比当前锁节点lockId更小的节点,若有则监控当前节点的前一节点  
  92.             //--------------------------------------------------------------  
  93.             if (lockId.equals(first)) {  
  94.                 System.out.println("thread " + Thread.currentThread().getName() +  
  95.                         " has get the lock, lockId is " + lockId);  
  96.                 return true;  
  97.             } else if (!lessThanMe.isEmpty()) {  
  98.                 String prevLockId = lessThanMe.last();  
  99.                 zk.exists(prevLockId, new LockWatcher());  
  100.                 //--------------------------------------------------------------  
  101.                 // 阀门等待sessionTimeout的时间  
  102.                 // 当等待sessionTimeout的时间过后,上一个lock的Zookeeper连接会过期,删除所有临时节点,触发监听器  
  103.                 //--------------------------------------------------------------  
  104.                 latch.await(sessionTimeout, TimeUnit.MILLISECONDS);  
  105.                 System.out.println("thread " + Thread.currentThread().getName() +  
  106.                         " has get the lock, lockId is " + lockId);  
  107.             }  
  108.   
  109.   
  110.         } catch (KeeperException e) {  
  111.             e.printStackTrace();  
  112.         } catch (InterruptedException e) {  
  113.             e.printStackTrace();  
  114.         }  
  115.   
  116.         return true;  
  117.     }  
  118.   
  119.   
  120.     public synchronized boolean unlock() {  
  121.         //--------------------------------------------------------------  
  122.         // 删除lockId节点以释放锁  
  123.         //--------------------------------------------------------------  
  124.         try {  
  125.             System.out.println("thread " + Thread.currentThread().getName() +  
  126.                     " unlock the lock: " + lockId + ", the node: " + lockId + " had been deleted");  
  127.             zk.delete(lockId, -1);  
  128.             return true;  
  129.         } catch (InterruptedException e) {  
  130.             e.printStackTrace();  
  131.         } catch (KeeperException e) {  
  132.             e.printStackTrace();  
  133.         } finally {  
  134.             try {  
  135.                 zk.close();  
  136.             } catch (InterruptedException e) {  
  137.                 e.printStackTrace();  
  138.             }  
  139.         }  
  140.         return false;  
  141.     }  
  142.   
  143.     /** 
  144.      * 保证锁根节点存在,若不存在则创建它 
  145.      */  
  146.     public void createLockRootIfNotExists() {  
  147.         try {  
  148.             Stat stat = zk.exists(LOCK_ROOT, false);  
  149.             if (stat == null) {  
  150.                 zk.create(LOCK_ROOT, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);  
  151.             }  
  152.         } catch (KeeperException e) {  
  153.             e.printStackTrace();  
  154.         } catch (InterruptedException e) {  
  155.             e.printStackTrace();  
  156.         }  
  157.     }  
  158.   
  159.     public static void main(String[] args) throws IOException, KeeperException, InterruptedException {  
  160.         final CountDownLatch latch = new CountDownLatch(10);  
  161.         for (int i = 0; i < 10; i++) {  
  162.             new Thread(new Runnable() {  
  163.                 public void run() {  
  164.                     DistributedLock lock = null;  
  165.                     try {  
  166.                         lock = new DistributedLock();  
  167.                         latch.countDown();  
  168.                         latch.await();  
  169.                         lock.lock();  
  170.                         Thread.sleep(RandomUtils.nextInt(200500));  
  171.                     }  catch (InterruptedException e) {  
  172.                         e.printStackTrace();  
  173.                     } catch (KeeperException e) {  
  174.                         e.printStackTrace();  
  175.                     } catch (IOException e) {  
  176.                         e.printStackTrace();  
  177.                     } finally {  
  178.                         if (lock != null) {  
  179.                             lock.unlock();  
  180.                         }  
  181.                     }  
  182.                 }  
  183.             }).start();  
  184.         }  
  185.     }  
  186.   
  187. }  


测试

运行main方法,本机器的某次输出结果
  1. thread Thread-8 create the lock node: /LOCKS/0000000090, trying to get lock now  
  2. thread Thread-9 create the lock node: /LOCKS/0000000092, trying to get lock now  
  3. thread Thread-3 create the lock node: /LOCKS/0000000093, trying to get lock now  
  4. thread Thread-2 create the lock node: /LOCKS/0000000091, trying to get lock now  
  5. thread Thread-5 create the lock node: /LOCKS/0000000094, trying to get lock now  
  6. thread Thread-7 create the lock node: /LOCKS/0000000095, trying to get lock now  
  7. thread Thread-1 create the lock node: /LOCKS/0000000098, trying to get lock now  
  8. thread Thread-6 create the lock node: /LOCKS/0000000096, trying to get lock now  
  9. thread Thread-4 create the lock node: /LOCKS/0000000097, trying to get lock now  
  10. thread Thread-0 create the lock node: /LOCKS/0000000099, trying to get lock now  
  11. thread Thread-8 has get the lock, lockId is /LOCKS/0000000090  
  12. thread Thread-8 unlock the lock: /LOCKS/0000000090, the node: /LOCKS/0000000090 had been deleted  
  13. thread Thread-2 has get the lock, lockId is /LOCKS/0000000091  
  14. thread Thread-2 unlock the lock: /LOCKS/0000000091, the node: /LOCKS/0000000091 had been deleted  
  15. thread Thread-9 has get the lock, lockId is /LOCKS/0000000092  
  16. thread Thread-9 unlock the lock: /LOCKS/0000000092, the node: /LOCKS/0000000092 had been deleted  
  17. thread Thread-3 has get the lock, lockId is /LOCKS/0000000093  
  18. thread Thread-3 unlock the lock: /LOCKS/0000000093, the node: /LOCKS/0000000093 had been deleted  
  19. thread Thread-5 has get the lock, lockId is /LOCKS/0000000094  
  20. thread Thread-5 unlock the lock: /LOCKS/0000000094, the node: /LOCKS/0000000094 had been deleted  
  21. thread Thread-7 has get the lock, lockId is /LOCKS/0000000095  
  22. thread Thread-7 unlock the lock: /LOCKS/0000000095, the node: /LOCKS/0000000095 had been deleted  
  23. thread Thread-6 has get the lock, lockId is /LOCKS/0000000096  
  24. thread Thread-6 unlock the lock: /LOCKS/0000000096, the node: /LOCKS/0000000096 had been deleted  
  25. thread Thread-4 has get the lock, lockId is /LOCKS/0000000097  
  26. thread Thread-4 unlock the lock: /LOCKS/0000000097, the node: /LOCKS/0000000097 had been deleted  
  27. thread Thread-1 has get the lock, lockId is /LOCKS/0000000098  
  28. thread Thread-1 unlock the lock: /LOCKS/0000000098, the node: /LOCKS/0000000098 had been deleted  
  29. thread Thread-0 has get the lock, lockId is /LOCKS/0000000099  
  30. thread Thread-0 unlock the lock: /LOCKS/0000000099, the node: /LOCKS/0000000099 had been deleted  

注:测试的结果证明了在同一JVM下,锁是正确的。由于本文锁的实现使用的是Zookeeper的监听机制,理论上在分布式系统下获取锁的顺序也是序列的顺序。但真正验证结果的正确性须在分布式系统下测试需同时运行几个JVM实例,在同一时刻去获取锁,再看输出结果。


 

 

参考文章:  

http://zookeeper.apache.org/doc/r3.4.9/recipes.html

http://www.ibm.com/developerworks/cn/opensource/os-cn-zookeeper/

 

 

本文为原创,转载请注明出处http://blog.csdn.net/massivestars/article/details/53771532

相关文章

相关标签/搜索