分布式鎖2 Java非常用技術(shù)方案探討之ZooKeeper

 1 public class MyDistributedLockByZK {
 2     /** 線程池 **/
 3     private static ExecutorService executorService = null;
 4     private static final int THREAD_NUM = 5;
 5     private static int threadNo = 0;
 6     private static CountDownLatch threadCompleteLatch = new CountDownLatch(THREAD_NUM);
 7     
 8     /** ZK的相關(guān)配置常量 **/
 9     private static final String CONNECTION_STRING = "192.168.224.170:2181";
10     private static final int SESSION_TIMEOUT = 10000;
11     // 此變量在LockWatcher中也有一個(gè)同名的靜態(tài)變量，正式使用的時(shí)候，提取到常量類中共同維護(hù)即可。
12     private static final String LOCK_ROOT_PATH = "/myDisLocks";
13     
14     public static void main(String[] args) {
15         // 定義線程池
16         executorService = Executors.newFixedThreadPool(THREAD_NUM, new ThreadFactory() {
17             @Override
18             public Thread newThread(Runnable r) {
19                 String name = String.format("第[%s]個(gè)測試線程", ++threadNo);
20                 Thread ret = new Thread(Thread.currentThread().getThreadGroup(), r, name, 0);
21                 ret.setDaemon(false);
22                 return ret;
23             }
24         });
25         
26         // 啟動(dòng)線程
27         if (executorService != null) {
28             startProcess();
29         }
30     }
31     
32     /**
33      * @author zhangyi03
34      * @date 2017-5-23 下午5:57:27
35      * @description 模擬并發(fā)執(zhí)行任務(wù)
36      */
37      public static void startProcess() {    
38         Runnable disposeBusinessRunnable= new Thread(new Runnable() {
39             public void run() {
40                 String threadName = Thread.currentThread().getName();
41                 
42                 LockWatcher lock = new LockWatcher(threadCompleteLatch);
43                 try {
44                     /** 步驟1: 當(dāng)前線程創(chuàng)建ZK連接  **/
45                     lock.createConnection(CONNECTION_STRING, SESSION_TIMEOUT);
46                     
47                     /** 步驟2: 創(chuàng)建鎖的根節(jié)點(diǎn)  **/
48                     // 注意，此處創(chuàng)建根節(jié)點(diǎn)的方式其實(shí)完全可以在初始化的時(shí)候由主線程單獨(dú)進(jìn)行根節(jié)點(diǎn)的創(chuàng)建，沒有必要在業(yè)務(wù)線程中創(chuàng)建。
49                     // 這里這樣寫只是一種思路而已，不必局限于此
50                     synchronized (MyDistributedLockByZK.class){
51                         lock.createPersistentPath(LOCK_ROOT_PATH, "該節(jié)點(diǎn)由" + threadName + "創(chuàng)建", true);
52                     }
53                     
54                     /** 步驟3: 開啟鎖競爭并執(zhí)行任務(wù) **/
55                     lock.getLock();
56                 } catch (Exception e) {
57                     e.printStackTrace();
58                 } 
59             }  
60         });
61         
62         for (int i = 0; i < THREAD_NUM; i++) {
63             executorService.execute(disposeBusinessRunnable);
64         }
65         executorService.shutdown();
66         
67         try {
68             threadCompleteLatch.await();
69             System.out.println("所有線程運(yùn)行結(jié)束!");
70         } catch (InterruptedException e) {
71             e.printStackTrace();
72         }
73      }
74 }

  1 public class LockWatcher implements Watcher {
  2     /** 成員變量 **/
  3     private ZooKeeper zk = null;
  4     // 當(dāng)前業(yè)務(wù)線程競爭鎖的時(shí)候創(chuàng)建的節(jié)點(diǎn)路徑
  5     private String selfPath = null;
  6     // 當(dāng)前業(yè)務(wù)線程競爭鎖的時(shí)候創(chuàng)建節(jié)點(diǎn)的前置節(jié)點(diǎn)路徑
  7     private String waitPath = null;
  8     // 確保連接zk成功；只有當(dāng)收到Watcher的監(jiān)聽事件之后，才執(zhí)行后續(xù)的操作，否則請(qǐng)求阻塞在createConnection()創(chuàng)建ZK連接的方法中
  9     private CountDownLatch connectSuccessLatch = new CountDownLatch(1);
 10     // 標(biāo)識(shí)線程是否執(zhí)行完任務(wù)
 11     private CountDownLatch threadCompleteLatch = null;
 12     
 13     /** ZK的相關(guān)配置常量 **/
 14     private static final String LOCK_ROOT_PATH = "/myDisLocks";
 15     private static final String LOCK_SUB_PATH = LOCK_ROOT_PATH + "/thread";
 16     
 17     public LockWatcher(CountDownLatch latch) {
 18         this.threadCompleteLatch = latch;
 19     }
 20     
 21     @Override
 22     public void process(WatchedEvent event) {
 23         if (event == null) {
 24             return;
 25         }
 26         
 27         // 通知狀態(tài)
 28         Event.KeeperState keeperState = event.getState();
 29         // 事件類型
 30         Event.EventType eventType = event.getType();
 31         
 32         // 根據(jù)通知狀態(tài)分別處理
 33         if (Event.KeeperState.SyncConnected == keeperState) {
 34             if ( Event.EventType.None == eventType ) {
 35                 System.out.println(Thread.currentThread().getName() + "成功連接上ZK服務(wù)器");
 36                 // 此處代碼的主要作用是用來輔助判斷當(dāng)前線程確實(shí)已經(jīng)連接上ZK
 37                 connectSuccessLatch.countDown();
 38             }else if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
 39                 System.out.println(Thread.currentThread().getName() + "收到情報(bào)，排我前面的家伙已掛，我準(zhǔn)備再次確認(rèn)我是不是最小的節(jié)點(diǎn)?。?);
 40                 try {
 41                     if(checkMinPath()){
 42                         getLockSuccess();
 43                     }
 44                 } catch (Exception e) {
 45                     e.printStackTrace();
 46                 } 
 47             }
 48         } else if ( Event.KeeperState.Disconnected == keeperState ) {
 49             System.out.println(Thread.currentThread().getName() + "與ZK服務(wù)器斷開連接");
 50         } else if ( Event.KeeperState.AuthFailed == keeperState ) {
 51             System.out.println(Thread.currentThread().getName() + "權(quán)限檢查失敗");
 52         } else if ( Event.KeeperState.Expired == keeperState ) {
 53             System.out.println(Thread.currentThread().getName() + "會(huì)話失效");
 54         }
 55     }
 56     
 57      /**
 58      * @author zhangyi03
 59      * @date 2017-5-23 下午6:07:03
 60      * @description 創(chuàng)建ZK連接
 61      * @param connectString ZK服務(wù)器地址列表
 62      * @param sessionTimeout Session超時(shí)時(shí)間
 63      * @throws IOException
 64      * @throws InterruptedException
 65      */
 66     public void createConnection(String connectString, int sessionTimeout) throws IOException, InterruptedException {
 67         zk = new ZooKeeper(connectString, sessionTimeout, this);
 68         // connectSuccessLatch.await(1, TimeUnit.SECONDS) 正式實(shí)現(xiàn)的時(shí)候可以考慮此處是否采用超時(shí)阻塞
 69         connectSuccessLatch.await();
 70     }
 71     
 72     /**
 73      * @author zhangyi03
 74      * @date 2017-5-23 下午6:15:48
 75      * @description 創(chuàng)建ZK節(jié)點(diǎn)
 76      * @param path 節(jié)點(diǎn)path
 77      * @param data 初始數(shù)據(jù)內(nèi)容
 78      * @param needWatch
 79      * @return
 80      * @throws KeeperException
 81      * @throws InterruptedException
 82      */
 83     public boolean createPersistentPath(String path, String data, boolean needWatch) throws KeeperException, InterruptedException {
 84         if(zk.exists(path, needWatch) == null){
 85             String result = zk.create( path,data.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
 86             System.out.println(Thread.currentThread().getName() + "創(chuàng)建節(jié)點(diǎn)成功, path: " + result + ", content: " + data);
 87         }
 88         return true;
 89     }
 90     
 91     /**
 92      * @author zhangyi03
 93      * @date 2017-5-23 下午6:24:46
 94      * @description 獲取分布式鎖
 95      * @throws KeeperException
 96      * @throws InterruptedException
 97      */
 98      public void getLock() throws Exception {
 99         selfPath = zk.create(LOCK_SUB_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
100         System.out.println(Thread.currentThread().getName() + "創(chuàng)建鎖路徑:" + selfPath);
101         if(checkMinPath()){
102             getLockSuccess();
103         }
104      }
105      
106      /**
107      * @author zhangyi03
108      * @date 2017-5-23 下午7:02:41
109      * @description 獲取鎖成功
110      * @throws KeeperException
111      * @throws InterruptedException
112      */
113     private void getLockSuccess() throws KeeperException, InterruptedException {
114          if(zk.exists(selfPath, false) == null){
115              System.err.println(Thread.currentThread().getName() + "本節(jié)點(diǎn)已不在了...");
116              return;
117          }
118          System.out.println(Thread.currentThread().getName() + "獲取鎖成功，開始處理業(yè)務(wù)數(shù)據(jù)！");
119          Thread.sleep(2000);
120          System.out.println(Thread.currentThread().getName() + "處理業(yè)務(wù)數(shù)據(jù)完成，刪除本節(jié)點(diǎn)：" + selfPath);
121          zk.delete(selfPath, -1);
122          releaseConnection();
123          threadCompleteLatch.countDown();
124      }
125 
126      /**
127      * @author zhangyi03
128      * @date 2017-5-23 下午7:06:46
129      * @description 關(guān)閉ZK連接
130      */
131     private void releaseConnection() {
132         if (zk != null) {
133             try {
134                 zk.close();
135             } catch (InterruptedException e) {
136                 e.printStackTrace();
137             }
138         }
139         System.out.println(Thread.currentThread().getName() + "釋放ZK連接");
140      }
141 
142      /**
143      * @author zhangyi03
144      * @date 2017-5-23 下午6:57:14
145      * @description 檢查自己是不是最小的節(jié)點(diǎn)
146      * @param selfPath
147      * @return
148      * @throws KeeperException
149      * @throws InterruptedException
150      */
151     private boolean checkMinPath() throws Exception {
152           List<String> subNodes = zk.getChildren(LOCK_ROOT_PATH, false);
153           // 根據(jù)元素按字典序升序排序
154           Collections.sort(subNodes);
155           System.err.println(Thread.currentThread().getName() + "創(chuàng)建的臨時(shí)節(jié)點(diǎn)名稱:" + selfPath.substring(LOCK_ROOT_PATH.length()+1));
156           int index = subNodes.indexOf(selfPath.substring(LOCK_ROOT_PATH.length()+1));
157           System.err.println(Thread.currentThread().getName() + "創(chuàng)建的臨時(shí)節(jié)點(diǎn)的index:" + index);
158           switch (index){
159               case -1: {
160                   System.err.println(Thread.currentThread().getName() + "創(chuàng)建的節(jié)點(diǎn)已不在了..." + selfPath);
161                   return false;
162               }
163               case 0:{
164                   System.out.println(Thread.currentThread().getName() +  "子節(jié)點(diǎn)中，我果然是老大" + selfPath);
165                   return true;
166               }
167               default:{
168                   // 獲取比當(dāng)前節(jié)點(diǎn)小的前置節(jié)點(diǎn),此處只關(guān)注前置節(jié)點(diǎn)是否還在存在，避免驚群現(xiàn)象產(chǎn)生
169                   waitPath = LOCK_ROOT_PATH +"/"+ subNodes.get(index - 1);
170                   System.out.println(Thread.currentThread().getName() + "獲取子節(jié)點(diǎn)中，排在我前面的節(jié)點(diǎn)是:" + waitPath);
171                   try {
172                       zk.getData(waitPath, true, new Stat());
173                       return false;
174                   } catch (Exception e) {
175                       if (zk.exists(waitPath, false) == null) {
176                           System.out.println(Thread.currentThread().getName() + "子節(jié)點(diǎn)中，排在我前面的" + waitPath + "已失蹤，該我了");
177                           return checkMinPath();
178                       } else {
179                           throw e;
180                       }
181                   }
182               }
183                   
184           }
185      }
186 }