分布式的几件小事（十）分布式锁是啥

/*** ZooKeeperSession* @author Administrator**/publicclassZooKeeperSession {
privatestaticCountDownLatchconnectedSemaphore=newCountDownLatch(1);
privateZooKeeperzookeeper;
privateCountDownLatchlatch;
publicZooKeeperSession() {
try {
this.zookeeper=newZooKeeper(
"192.168.31.187:2181,192.168.31.19:2181,192.168.31.227:2181", 
50000, 
newZooKeeperWatcher());            
try {
connectedSemaphore.await();
            } catch(InterruptedExceptione) {
e.printStackTrace();
            }
System.out.println("ZooKeeper session established......");
        } catch (Exceptione) {
e.printStackTrace();
        }
    }
/*** 获取分布式锁* @param productId*/publicBooleanacquireDistributedLock(LongproductId) {
Stringpath="/product-lock-"+productId;
try {
zookeeper.create(path, "".getBytes(), 
Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
returntrue;
        } catch (Exceptione) {
while(true) {
try {
Statstat=zk.exists(path, true); // 相当于是给node注册一个监听器，去看看这个监听器是否存在if(stat!=null) {
this.latch=newCountDownLatch(1);
this.latch.await(waitTime, TimeUnit.MILLISECONDS);
this.latch=null;
                        }
zookeeper.create(path, "".getBytes(), 
Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
returntrue;
                     } catch(Exceptione) {
continue;
                     }
            }
// 很不优雅，我呢就是给大家来演示这么一个思路,比较通用的.      }
returntrue;
    }
/*** 释放掉一个分布式锁* @param productId*/publicvoidreleaseDistributedLock(LongproductId) {
Stringpath="/product-lock-"+productId;
try {
zookeeper.delete(path, -1); 
System.out.println("release the lock for product[id="+productId+"]......");  
        } catch (Exceptione) {
e.printStackTrace();
        }
    }
/*** 建立zk session的watcher* @author Administrator**/privateclassZooKeeperWatcherimplementsWatcher {
publicvoidprocess(WatchedEventevent) {
System.out.println("Receive watched event: "+event.getState());
if(KeeperState.SyncConnected==event.getState()) {
connectedSemaphore.countDown();
            } 
if(this.latch!=null) {  
this.latch.countDown();  
             }
        }
    }
/*** 封装单例的静态内部类* @author Administrator**/privatestaticclassSingleton {
privatestaticZooKeeperSessioninstance;
static {
instance=newZooKeeperSession();
        }
publicstaticZooKeeperSessiongetInstance() {
returninstance;
        }
    }
/*** 获取单例* @return*/publicstaticZooKeeperSessiongetInstance() {
returnSingleton.getInstance();
    }
/*** 初始化单例的便捷方法*/publicstaticvoidinit() {
getInstance();
    }
}

publicclassZooKeeperDistributedLockimplementsWatcher{
privateZooKeeperzk;
privateStringlocksRoot="/locks";
privateStringproductId;
privateStringwaitNode;
privateStringlockNode;
privateCountDownLatchlatch;
privateCountDownLatchconnectedLatch=newCountDownLatch(1);
privateintsessionTimeout=30000; 
publicZooKeeperDistributedLock(StringproductId){
this.productId=productId;
try {
Stringaddress="192.168.31.187:2181,192.168.31.19:2181,192.168.31.227:2181";
zk=newZooKeeper(address, sessionTimeout, this);
connectedLatch.await();
        } catch (IOExceptione) {
thrownewLockException(e);
        } catch (KeeperExceptione) {
thrownewLockException(e);
        } catch (InterruptedExceptione) {
thrownewLockException(e);
        }
    }
publicvoidprocess(WatchedEventevent) {
if(event.getState()==KeeperState.SyncConnected){
connectedLatch.countDown();
return;
        }
if(this.latch!=null) {  
this.latch.countDown(); 
        }
    }
publicvoidacquireDistributedLock() {   
try {
if(this.tryLock()){
return;
            }
else{
waitForLock(waitNode, sessionTimeout);
            }
        } catch (KeeperExceptione) {
thrownewLockException(e);
        } catch (InterruptedExceptione) {
thrownewLockException(e);
        } 
    }
publicbooleantryLock() {
try {
// 传入进去的locksRoot + “/” + productId// 假设productId代表了一个商品id，比如说1// locksRoot = locks// /locks/10000000000，/locks/10000000001，/locks/10000000002lockNode=zk.create(locksRoot+"/"+productId, newbyte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
// 看看刚创建的节点是不是最小的节点// locks：10000000000，10000000001，10000000002List<String>locks=zk.getChildren(locksRoot, false);
Collections.sort(locks);
if(lockNode.equals(locksRoot+"/"+locks.get(0))){
//如果是最小的节点,则表示取得锁returntrue;
            }
//如果不是最小的节点，找到比自己小1的节点intpreviousLockIndex=-1;
for(inti=0; i<locks.size(); i++) {
if(lockNode.equals(locksRoot+“/”+locks.get(i))) {
previousLockIndex=i-1;
break;
                 }
            }
this.waitNode=locks.get(previousLockIndex);
        } catch (KeeperExceptione) {
thrownewLockException(e);
        } catch (InterruptedExceptione) {
thrownewLockException(e);
        }
returnfalse;
    }
privatebooleanwaitForLock(StringwaitNode, longwaitTime) throwsInterruptedException, KeeperException {
Statstat=zk.exists(locksRoot+"/"+waitNode, true);
if(stat!=null){
this.latch=newCountDownLatch(1);
this.latch.await(waitTime, TimeUnit.MILLISECONDS);                 this.latch=null;
        }
returntrue;
   }
publicvoidunlock() {
try {
// 删除/locks/10000000000节点// 删除/locks/10000000001节点System.out.println("unlock "+lockNode);
zk.delete(lockNode,-1);
lockNode=null;
zk.close();
        } catch (InterruptedExceptione) {
e.printStackTrace();
        } catch (KeeperExceptione) {
e.printStackTrace();
        }
    }
publicclassLockExceptionextendsRuntimeException {
privatestaticfinallongserialVersionUID=1L;
publicLockException(Stringe){
super(e);
        }
publicLockException(Exceptione){
super(e);
        }
    }
}