利用ZooKeeper简单实现分布式锁
下面讲解下怎么利用ZooKeeper简单实现分布式锁，对于zk的安装和简单使用请参考我的文章：Hadoop集群之ZooKeeper和Hbase环境搭建
1.分布式锁的由来：
在程序开发过程中不得不考虑的就是并发问题。在java中对于同一个jvm而言，jdk已经提供了lock和同步等。但是在分布式情况下，往往存在多个进程对一些资源产生竞争关系，而这些进程往往在不同的机器上，这个时候jdk中提供的已经不能满足。分布式锁顾明思议就是可以满足分布式情况下的并发锁。下面我们讲解怎么利用zk实现分布式锁。
2.实现思路：
2.1zk简单介绍：
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.详细见代码注释
packagecn.wpeace.zktest;
importjava.io.IOException;
importjava.util.ArrayList;
importjava.util.Collections;
importjava.util.List;
importjava.util.concurrent.CountDownLatch;
importjava.util.concurrent.TimeUnit;
importjava.util.concurrent.locks.Condition;
importjava.util.concurrent.locks.Lock;
importorg.apache.zookeeper.CreateMode;
importorg.apache.zookeeper.KeeperException;
importorg.apache.zookeeper.WatchedEvent;
importorg.apache.zookeeper.Watcher;
importorg.apache.zookeeper.ZooDefs;
importorg.apache.zookeeper.ZooKeeper;
importorg.apache.zookeeper.Watcher.Event.KeeperState;
importorg.apache.zookeeper.data.Stat;
/
@authorpeace
/
publicclassDistributedLockimplementsLock,Watcher{
privateZooKeeperzk;
privateStringroot="/locks";//根
privateStringlockName;//竞争资源的标志
privateStringwaitNode;//等待前一个锁
privateStringmyZnode;//当前锁
privateCountDownLatchlatch;//计数器
privateCountDownLatchconnectedSignal=newCountDownLatch(1);
privateintsessionTimeout=30000;
/
创建分布式锁,使用前请确认config配置的zookeeper服务可用
@paramconfig192.168.1.127:2181
@paramlockName竞争资源标志,lockName中不能包含单词_lock_
/
publicDistributedLock(Stringconfig,StringlockName){
this.lockName=lockName;
//创建一个与服务器的连接
try{
zk=newZooKeeper(config,sessionTimeout,this);
connectedSignal.await();
Statstat=zk.exists(root,false);//此去不执行Watcher
if(stat==null){
//创建根节点
zk.create(root,newbyte[0],ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
}
}catch(IOExceptione){
thrownewLockException(e);
}catch(KeeperExceptione){
thrownewLockException(e);
}catch(InterruptedExceptione){
thrownewLockException(e);
}
}
/
zookeeper节点的监视器
/
publicvoidprocess(WatchedEventevent){
//建立连接用
if(event.getState()==KeeperState.SyncConnected){
connectedSignal.countDown();
return;
}
//其他线程放弃锁的标志
if(this.latch!=null){
this.latch.countDown();
}
}
publicvoidlock(){
try{
if(this.tryLock()){
System.out.println("Thread"+Thread.currentThread().getId()+""+myZnode+"getlocktrue");
return;
}
else{
waitForLock(waitNode,sessionTimeout);//等待锁
}
}catch(KeeperExceptione){
thrownewLockException(e);
}catch(InterruptedExceptione){
thrownewLockException(e);
}
}
publicbooleantryLock(){
try{
StringsplitStr="_lock_";
if(lockName.contains(splitStr))
thrownewLockException("lockNamecannotcontains\\u000B");
//创建临时子节点
myZnode=zk.create(root+"/"+lockName+splitStr,newbyte[0],ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(myZnode+"iscreated");
//取出所有子节点
ListsubNodes=zk.getChildren(root,false);
//取出所有lockName的锁
ListlockObjNodes=newArrayList();
for(Stringnode: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));
returntrue;
}
//如果不是最小的节点，找到比自己小1的节点
StringsubMyZnode=myZnode.substring(myZnode.lastIndexOf("/")+1);
waitNode=lockObjNodes.get(Collections.binarySearch(lockObjNodes,subMyZnode)-1);//找到前一个子节点
}catch(KeeperExceptione){
thrownewLockException(e);
}catch(InterruptedExceptione){
thrownewLockException(e);
}
returnfalse;
}
publicbooleantryLock(longtime,TimeUnitunit){
try{
if(this.tryLock()){
returntrue;
}
returnwaitForLock(waitNode,time);
}catch(Exceptione){
e.printStackTrace();
}
returnfalse;
}
privatebooleanwaitForLock(Stringlower,longwaitTime)throwsInterruptedException,KeeperException{
Statstat=zk.exists(root+"/"+lower,true);//同时注册监听。
//判断比自己小一个数的节点是否存在,如果不存在则无需等待锁,同时注册监听
if(stat!=null){
System.out.println("Thread"+Thread.currentThread().getId()+"waitingfor"+root+"/"+lower);
this.latch=newCountDownLatch(1);
this.latch.await(waitTime,TimeUnit.MILLISECONDS);//等待，这里应该一直等待其他线程释放锁
this.latch=null;
}
returntrue;
}
publicvoidunlock(){
try{
System.out.println("unlock"+myZnode);
zk.delete(mwww.sm136.comyZnode,-1);
myZnode=null;
zk.close();
}catch(InterruptedExceptione){
e.printStackTrace();
}catch(KeeperExceptione){
e.printStackTrace();
}
}
publicvoidlockInterruptibly()throwsInterruptedException{
this.lock();
}
publicConditionnewCondition(){
returnnull;
}
publicclassLockExceptionextendsRuntimeException{
privatestaticfinallongserialVersionUID=1L;
publicLockException(Stringe){
super(e);
}
publicLockException(Exceptione){
super(e);
}
}
}
4.测试：
1.下载我的工程代码：点击下载
2.或者直接使用该类：
DistributedLocklock=newDistributedLock("192.168.1.127:2181","lock");
lock.lock();
//共享资源
if(lock!=null)
lock.unlock();
3.执行步骤：
1.启动zk:zkServer.shstart
2.启动测试代码
3.执行结果：
|
|
