ZooKeeper實現分布式鎖

全部代碼下載：CSDN鏈接：CSDN鏈接;寫文章不易，歡迎大家采我的文章，以及給出有用的評論，當然大家也可以關注一下我的github；多謝；

1.分布式鎖的由來：

在程序開發過程中不得不考慮的就是並發問題。在java中對於同一個jvm而言，jdk已經提供了lock和同步等。但是在分布式情況下，往往存在多個進程對一些資源產生競爭關系，而這些進程往往在不同的機器上，這個時候jdk中提供的已經不能滿足。分布式鎖顧明思議就是可以滿足分布式情況下的並發鎖。 下面我們講解怎么利用zk實現分布式鎖。

2.實現思路：

2.1 zk簡單介紹：

ZooKeeper是Apache軟件基金會的一個軟件項目，他為大型分布式計算提供開源的分布式配置服務、同步服務和命名注冊。在 ZooKeeper 中，節點類型可以分為持久節點（PERSISTENT ）、臨時節點（EPHEMERAL），以及時序節點（SEQUENTIAL ），具體在節點創建過程中，一般是組合使用，可以生成 4 種節點類型：持久節點（PERSISTENT），持久順序節點（PERSISTENT_SEQUENTIAL），臨時節點（EPHEMERAL），臨時順序節點（EPHEMERAL_SEQUENTIAL）；具體節點含義，谷歌之。

2.2 利用zk實現：

當很多進程需要訪問共享資源時，我們可以通過zk來實現分布式鎖。主要步驟是： 
1.建立一個節點，假如名為：lock 。節點類型為持久節點（PERSISTENT） 
2.每當進程需要訪問共享資源時，會調用分布式鎖的lock()或tryLock()方法獲得鎖，這個時候會在第一步創建的lock節點下建立相應的順序子節點，節點類型為臨時順序節點（EPHEMERAL_SEQUENTIAL），通過組成特定的名字name+lock+順序號。 
3.在建立子節點后，對lock下面的所有以name開頭的子節點進行排序，判斷剛剛建立的子節點順序號是否是最小的節點，假如是最小節點，則獲得該鎖對資源進行訪問。 
4.假如不是該節點，就獲得該節點的上一順序節點，並給該節點是否存在注冊監聽事件。同時在這里阻塞。等待監聽事件的發生，獲得鎖控制權。 
5.當調用完共享資源后，調用unlock（）方法，關閉zk，進而可以引發監聽事件，釋放該鎖。 
實現的分布式鎖是嚴格的按照順序訪問的並發鎖。

3.代碼實現：

下面將講解使用java實現分布式鎖： 
1.建立類DistributedLock，實現java.util.concurrent.locks.Lock;和org.apache.zookeeper.Watcher接口 
2.實現lock下面的方法：主要包括lock，tryLock，unlock等 
3.實現watcher接口下的process方法。 
4.在構造器中對zk進行初始化。 
5.詳細見代碼注釋

package cn.wpeace.zktest; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.Watcher.Event.KeeperState; import org.apache.zookeeper.data.Stat; /** * @author peace * */ public class DistributedLock implements Lock, Watcher{ private ZooKeeper zk; private String root = "/locks";//根 private String lockName;//競爭資源的標志 private String waitNode;//等待前一個鎖 private String myZnode;//當前鎖 private CountDownLatch latch;//計數器 private CountDownLatch connectedSignal=new CountDownLatch(1); private int sessionTimeout = 30000; /** * 創建分布式鎖,使用前請確認config配置的zookeeper服務可用 * @param config 192.168.1.127:2181 * @param lockName 競爭資源標志,lockName中不能包含單詞_lock_ */ public DistributedLock(String config, String lockName){ this.lockName = lockName; // 創建一個與服務器的連接 try { zk = new ZooKeeper(config, sessionTimeout, this); connectedSignal.await(); Stat stat = zk.exists(root, false);//此去不執行 Watcher if(stat == null){ // 創建根節點 zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); } } catch (IOException e) { throw new LockException(e); } catch (KeeperException e) { throw new LockException(e); } catch (InterruptedException e) { throw new LockException(e); } } /** * zookeeper節點的監視器 */ public void process(WatchedEvent event) { //建立連接用 if(event.getState()==KeeperState.SyncConnected){ connectedSignal.countDown(); return; } //其他線程放棄鎖的標志 if(this.latch != null) { this.latch.countDown(); } } public void lock() { try { if(this.tryLock()){ System.out.println("Thread " + Thread.currentThread().getId() + " " +myZnode + " get lock true"); return; } else{ waitForLock(waitNode, sessionTimeout);//等待鎖 } } catch (KeeperException e) { throw new LockException(e); } catch (InterruptedException e) { throw new LockException(e); } } public boolean tryLock() { try { String splitStr = "_lock_"; if(lockName.contains(splitStr)) throw new LockException("lockName can not contains \\u000B"); //創建臨時子節點 myZnode = zk.create(root + "/" + lockName + splitStr, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL); System.out.println(myZnode + " is created "); //取出所有子節點 List<String> subNodes = zk.getChildren(root, false); //取出所有lockName的鎖 List<String> lockObjNodes = new ArrayList<String>(); for (String node : subNodes) { String _node = node.split(splitStr)[0]; if(_node.equals(lockName)){ lockObjNodes.add(node); } } Collections.sort(lockObjNodes); if(myZnode.equals(root+"/"+lockObjNodes.get(0))){ //如果是最小的節點,則表示取得鎖 System.out.println(myZnode + "==" + lockObjNodes.get(0)); return true; } //如果不是最小的節點，找到比自己小1的節點 String subMyZnode = myZnode.substring(myZnode.lastIndexOf("/") + 1); waitNode = lockObjNodes.get(Collections.binarySearch(lockObjNodes, subMyZnode) - 1);//找到前一個子節點 } catch (KeeperException e) { throw new LockException(e); } catch (InterruptedException e) { throw new LockException(e); } return false; } public boolean tryLock(long time, TimeUnit unit) { try { if(this.tryLock()){ return true; } return waitForLock(waitNode,time); } catch (Exception e) { e.printStackTrace(); } return false; } private boolean waitForLock(String lower, long waitTime) throws InterruptedException, KeeperException { Stat stat = zk.exists(root + "/" + lower,true);//同時注冊監聽。 //判斷比自己小一個數的節點是否存在,如果不存在則無需等待鎖,同時注冊監聽 if(stat != null){ System.out.println("Thread " + Thread.currentThread().getId() + " waiting for " + root + "/" + lower); this.latch = new CountDownLatch(1); this.latch.await(waitTime, TimeUnit.MILLISECONDS);//等待，這里應該一直等待其他線程釋放鎖 this.latch = null; } return true; } public void unlock() { try { System.out.println("unlock " + myZnode); zk.delete(myZnode,-1); myZnode = null; zk.close(); } catch (InterruptedException e) { e.printStackTrace(); } catch (KeeperException e) { e.printStackTrace(); } } public void lockInterruptibly() throws InterruptedException { this.lock(); } public Condition newCondition() { return null; } public class LockException extends RuntimeException { private static final long serialVersionUID = 1L; public LockException(String e){ super(e); } public LockException(Exception e){ super(e); } } } 

4.測試：

1.下載我的工程代碼：點擊下載 
2.或者直接使用該類：

 DistributedLock lock = new DistributedLock("192.168.1.127:2181","lock"); lock.lock(); //共享資源 if(lock != null) lock.unlock(); 

3.執行步驟： 
1.啟動zk:zkServer.sh start 
2.啟動測試代碼 
3.執行結果： 
 

參考鏈接：https://blog.csdn.net/peace1213/article/details/52571445