http://www.jiacheo.org/blog/122

zookeeper是hadoop下面的一個子項目, 用來協調跟hadoop相關的一些分布式的框架, 如hadoop, hive, pig等, 其實他們都是動物, 所以叫zookeeper(本人歪歪).

zookeeper其實是集群中每個節點都維護著一棵相同的樹, 樹的結構跟linux的目錄結構的概念差不多, 以/為跟節點, 下邊可以擴展任意的節點和葉子節點, 每個節點都可以寫入數據. 基于zookeeper的分布式鎖的實現, 其實是得益于zookeeper同步文件的強大性, 我們相信每時每刻我們訪問zookeeper的樹時, 相同節點返回的數據都是一致的. 這要靠zookeeper內部的一些算法來實現. 特別是leader的選舉算法, 這里就不說了, 感興趣的話可以去搜索一下看看.

我們知道了zookeeper集群的每個節點的數據都是一致的, 那么我們可以通過這些節點來作為鎖的標志.

首先給鎖設置一下API, 至少要包含, lock(鎖住), unlock(解鎖), isLocked(是否鎖住)三個方法

然后我們可以創建一個工廠(LockFactory), 用來專門生產鎖.

鎖的創建過程如下描述:

前提:每個鎖都需要一個路徑來指定(如:/jiacheo/lock)

1.根據指定的路徑, 查找zookeeper集群下的這個節點是否存在.(說明已經有鎖了)

2. 如果存在, 根據查詢者的一些特征數據(如ip地址/hostname), 當前的鎖是不是查詢者的

3. 如果不是查詢者的鎖, 則返回null, 說明創建鎖失敗

4. 如果是查詢者的鎖, 則把這個鎖返回給查詢者

5. 如果這個節點不存在, 說明當前沒有鎖, 那么創建一個臨時節點, 并將查詢者的特征信息寫入這個節點的數據中, 然后返回這個鎖.

根據以上5部, 一個分布式的鎖就可以創建了.

創建的鎖有三種狀態:

1. 創建失敗(null), 說明該鎖被其他查詢者使用了.’

2. 創建成功, 但當前沒有鎖住(unlocked), 可以使用

3. 創建成功, 但當前已經鎖住(locked)了, 不能繼續加鎖.

如圖, 如果我們getLock(“/jiacheo/lock1″,”192.168.0.100″), 想要獲取/jiacheo/lock1這個鎖的話, 我們先判斷這個節點是否存在, 存在的話獲取他的數據(data), 然后通過解析data, 我們可以知道這個節點是不是我們查詢者創建的(通過ip地址寫入節點數據中), 然后就可以返回一個鎖了.

具體的java實現(implementation)代碼如下:
1. Lock.java

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95

package org.jiacheo.zkdl.lock; ?? import java.net.InetAddress; import java.net.UnknownHostException; ?? import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.Stat; ?? /** ?* 類名：<b>Lock</b> <br/> ?* <p> ?* 類描述：? ?* </p> ?* 創建人：jiacheo <br/> ?* 創建時間：2011-1-27 上午01:30:25? <br/>?? ?* @version 2011-1-27?? ?* ?*/ public class Lock { ????private String path; ????private ZooKeeper zooKeeper; ????public Lock(String path){ ????????this.path = path; ????} ?????? ????/** ?????* <p> ?????* 方法描述: 上鎖 lock it ?????* </p> ?????* 創建人：jiacheo <br/> ?????* 創建時間：2011-1-27 上午01:30:50? <br/> ?????* @throws Exception ?????*/ ????public synchronized void lock() throws Exception{ ????????Stat stat = zooKeeper.exists(path, true); ????????String data = InetAddress.getLocalHost().getHostAddress()+":lock"; ????????zooKeeper.setData(path, data.getBytes(), stat.getVersion()); ????} ?????? ????/** ?????* <p> ?????* 方法描述：開鎖 unlock it ?????* </p> ?????* 創建人：jiacheo <br/> ?????* 創建時間：2011-1-27 上午01:31:20? <br/> ?????* @throws Exception ?????*/ ????public synchronized void unLock() throws Exception{ ????????Stat stat = zooKeeper.exists(path, true); ????????String data = InetAddress.getLocalHost().getHostAddress()+":unlock"; ????????zooKeeper.setData(path, data.getBytes(), stat.getVersion()); ????} ?????? ????/** ?????* <p> ?????* 方法描述：是否鎖住了, isLocked? ?????* </p> ?????* 創建人：jiacheo <br/> ?????* 創建時間：2011-1-27 上午01:31:43? <br/> ?????* @return ?????*/ ????public synchronized boolean isLock(){ ????????try { ????????????Stat stat = zooKeeper.exists(path, true); ????????????String data = InetAddress.getLocalHost().getHostAddress()+":lock"; ????????????String nodeData = new String(zooKeeper.getData(path, true, stat)); ????????????if(data.equals(nodeData)){ //????????????? lock = true; ????????????????return true; ????????????} ????????} catch (UnknownHostException e) { ????????????// ignore it ????????} catch (KeeperException e) { ????????????//TODO use log system and throw a new exception ????????} catch (InterruptedException e) { ????????????// TODO use log system and throw a new exception ????????} ????????return false; ????} ?? ????public String getPath() { ????????return path; ????} ?? ????public void setPath(String path) { ????????this.path = path; ????} ?? ????public void setZooKeeper(ZooKeeper zooKeeper) { ????????this.zooKeeper = zooKeeper; ????} ?????? ?????? }

2.LockFactory.java

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76

package org.jiacheo.zkdl.lock; ?? import java.io.IOException; import java.net.InetAddress; import java.util.Collections; ?? import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooDefs.Perms; import org.apache.zookeeper.data.ACL; import org.apache.zookeeper.data.Stat; ?? public class LockFactory { ?????? ????public static final ZooKeeper DEFAULT_ZOOKEEPER = getDefaultZookeeper(); ????//data格式:? ip:stat? 如: 10.232.35.70:lock 10.232.35.70:unlock ????public static synchronized Lock getLock(String path,String ip) throws Exception{ ????????if(DEFAULT_ZOOKEEPER != null){ ????????????Stat stat = null; ????????????try{ ????????????????stat = DEFAULT_ZOOKEEPER.exists(path, true); ????????????}catch (Exception e) { ????????????????// TODO: use log system and throw new exception ????????????} ????????????if(stat!=null){ ????????????????byte[] data = DEFAULT_ZOOKEEPER.getData(path, null, stat); ????????????????String dataStr = new String(data); ????????????????String[] ipv = dataStr.split(":"); ????????????????if(ip.equals(ipv[0])){ ????????????????????Lock lock = new Lock(path); ????????????????????lock.setZooKeeper(DEFAULT_ZOOKEEPER); ????????????????????return lock; ????????????????} ????????????????//is not your lock, return null ????????????????else{ ????????????????????return null; ????????????????} ????????????} ????????????//no lock created yet, you can get it ????????????else{ ????????????????createZnode(path); ????????????????Lock lock = new Lock(path); ????????????????lock.setZooKeeper(DEFAULT_ZOOKEEPER); ????????????????return lock; ????????????} ????????} ????????return null; ????} ?????? ????private static ZooKeeper getDefaultZookeeper() { ????????try { ????????????ZooKeeper zooKeeper = new ZooKeeper("10.232.35.72", 10*1000, new Watcher(){ ????????????????public void process(WatchedEvent event) { ????????????????????//節點的事件處理. you can do something when the node's data change //????????????????? System.out.println("event " + event.getType() + " has happened!"); ????????????????} ????????????}); ????????????return zooKeeper; ????????} catch (IOException e) { ????????????e.printStackTrace(); ????????} ????????return null; ????} ?? ????private static void createZnode(String path) throws Exception{ ?????????? ????????if(DEFAULT_ZOOKEEPER!=null){ ????????????InetAddress address = InetAddress.getLocalHost(); ????????????String data = address.getHostAddress()+":unlock"; ????????????DEFAULT_ZOOKEEPER.create(path, data.getBytes(),Collections.singletonList(new ACL(Perms.ALL,Ids.ANYONE_ID_UNSAFE)) , CreateMode.EPHEMERAL); ????????} ????} }

?

總結

以上是生活随笔為你收集整理的基于zookeeper实现的分布式锁的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。