Zookeeper实现分布式锁

在分布式的环境中有时需要控制共享资源的访问，协同每一个客户端的操作步骤。分布式锁的实现有很多种方案，下面先来介绍一下基于Zookeeper实现的分布式锁方案。

流程图如下:



注意点：

1.locks节点必须为持久节点，

2.子节点的类型为EPHEMERAL_SEQUENTIAL

核心代码:

1.定义锁的一些方法:

public interface DistributedLock {

/*
* 获取锁，如果没有得到就等待
*/
public void acquire() throws Exception;

/*
* 获取锁，直到超时
*/
public boolean acquire(long time, TimeUnit unit) throws Exception;

/*
* 释放锁
*/
public void release() throws Exception;

}2.定义锁的实现类:
public class SimpleDistributedLockMutex extends BaseDistributedLock implements
DistributedLock {
//锁名称前缀
private static final String LOCK_NAME = "lock_";
private final String basePath;
private String ourLockPath;
private boolean internalLock(long time, TimeUnit unit) throws Exception
{
ourLockPath = attemptLock(time, unit);
return ourLockPath != null;
}
public SimpleDistributedLockMutex(ZkClientExt client, String basePath){

super(client,basePath,LOCK_NAME);
this.basePath = basePath;
}
public void acquire() throws Exception {
if ( !internalLock(-1, null) )
{
throw new IOException("连接丢失!在路径:'"+basePath+"'下不能获取锁!");
}
}

public boolean acquire(long time, TimeUnit unit) throws Exception {

return internalLock(time, unit);
}

public void release() throws Exception {
releaseLock(ourLockPath);
}
}
具体实现细节:
public class BaseDistributedLock {
private final ZkClientExt client;
private final String path;
private final String basePath;
private final String lockName;
private static final Integer MAX_RETRY_COUNT = 10;
public BaseDistributedLock(ZkClientExt client, String path, String lockName){
this.client = client;
this.basePath = path;
this.path = path.concat("/").concat(lockName);
this.lockName = lockName;
}
private void deleteOurPath(String ourPath) throws Exception{
client.delete(ourPath);
}
private String createLockNode(ZkClient client, String path) throws Exception{
return client.createEphemeralSequential(path, null);
}
private boolean waitToLock(long startMillis, Long millisToWait, String ourPath) throws Exception{
boolean haveTheLock = false;
boolean doDelete = false;
try
{
while ( !haveTheLock )
{
List<String> children = getSortedChildren();
String sequenceNodeName = ourPath.substring(basePath.length()+1);
int ourIndex = children.indexOf(sequenceNodeName);
if ( ourIndex<0 ){
throw new ZkNoNodeException("节点没有找到: " + sequenceNodeName);
}
boolean isGetTheLock = ourIndex == 0;
String pathToWatch = isGetTheLock ? null : children.get(ourIndex - 1);
if ( isGetTheLock ){
haveTheLock = true;
}else{
String previousSequencePath = basePath .concat( "/" ) .concat( pathToWatch );
final CountDownLatch latch = new CountDownLatch(1);
final IZkDataListener previousListener = new IZkDataListener() {

public void handleDataDeleted(String dataPath) throws Exception {
latch.countDown();
}
public void handleDataChange(String dataPath, Object data) throws Exception {

}
};
try
{
//如果节点不存在会出现异常
client.subscribeDataChanges(previousSequencePath, previousListener);
if ( millisToWait != null )
{
millisToWait -= (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
if ( millisToWait <= 0 )
{
doDelete = true; // timed out - delete our node
break;
}
latch.await(millisToWait, TimeUnit.MICROSECONDS);
}
else
{
latch.await();
}
}
catch ( ZkNoNodeException e )
{
//ignore
}finally{
client.unsubscribeDataChanges(previousSequencePath, previousListener);
}

}
}
}
catch ( Exception e )
{
//发生异常需要删除节点
doDelete = true;
throw e;
}
finally
{
//如果需要删除节点
if ( doDelete )
{
deleteOurPath(ourPath);
}
}
return haveTheLock;
}
private String getLockNodeNumber(String str, String lockName)
{
int index = str.lastIndexOf(lockName);
if ( index >= 0 )
{
index += lockName.length();
return index <= str.length() ? str.substring(index) : "";
}
return str;
}
List<String> getSortedChildren() throws Exception
{
try{

List<String> children = client.getChildren(basePath);
Collections.sort
(
children,
new Comparator<String>()
{
public int compare(String lhs, String rhs)
{
return getLockNodeNumber(lhs, lockName).compareTo(getLockNodeNumber(rhs, lockName));
}
}
);
return children;

}catch(ZkNoNodeException e){

client.createPersistent(basePath, true);
return getSortedChildren();

}
}
protected void releaseLock(String lockPath) throws Exception{
deleteOurPath(lockPath);

}
protected String attemptLock(long time, TimeUnit unit) throws Exception{
final long startMillis = System.currentTimeMillis();
final Long millisToWait = (unit != null) ? unit.toMillis(time) : null;
String ourPath = null;
boolean hasTheLock = false;
boolean isDone = false;
int retryCount = 0;
//网络闪断需要重试一试
while ( !isDone )
{
isDone = true;
try
{
ourPath = createLockNode(client, path);
hasTheLock = waitToLock(startMillis, millisToWait, ourPath);
}
catch ( ZkNoNodeException e )
{
if ( retryCount++ < MAX_RETRY_COUNT )
{
isDone = false;
}
else
{
throw e;
}
}
}
if ( hasTheLock )
{
return ourPath;
}
return null;
}
}