分布式锁基于zookeeper实现

public class LockUtil {
private static final Logger logger = LoggerFactory.getLogger(LockUtil.class);
private static final byte[]  data      = { 0x12, 0x34 };
private static Watcher watcher =  new Watcher() {

public void process(WatchedEvent event) {
LoggerUtils.logInfo(logger,"process : " + event.getType());
}
};
private static ZooKeeper           zookeeper=null;
private final String         root;                                     //根节点路径
private String               id;
private LockNode             idName;
private String               ownerId;
private String               lastChildId;
private Throwable            other     = null;
private KeeperException      exception = null;
private InterruptedException interrupt = null;
private static final Long DEFAULT_TIMEOUT_PERIOD=1000L;
private ReentrantLock       reentrantLock = new ReentrantLock();

static{
try
{
zookeeper =new ZooKeeper("ip：port", DEFAULT_TIMEOUT_PERIOD.intValue(),watcher);
}
catch (IOException e)
{
logger.error("获取zookeeper错误");
}
}

public LockUtil(String root) {
this.root = root;
ensureExists(root);

}

/**
* 尝试获取锁操作，阻塞式可被中断
* @throws NestableRuntimeException
*/
public void lock() throws InterruptedException, KeeperException, NestableRuntimeException {
// 可能初始化的时候就失败了
if (exception != null) {
throw exception;
}

if (interrupt != null) {
throw interrupt;
}

if (other != null) {
throw new NestableRuntimeException(other);
}

if (isOwner()) {//锁重入
return;
}
reentrantLock.lock();
BooleanMutex mutex = new BooleanMutex();
acquireLock(mutex);
// 避免zookeeper重启后导致watcher丢失，会出现死锁使用了超时进行重试
try {
mutex.get(DEFAULT_TIMEOUT_PERIOD, TimeUnit.MICROSECONDS);// 阻塞等待值为true
// mutex.get();
} catch (TimeoutException e) {
if (!mutex.state()) {
lock();
}
}

if (exception != null) {
throw exception;
}

if (interrupt != null) {
throw interrupt;
}

if (other != null) {
throw new NestableRuntimeException(other);
}
}

/**
* 尝试获取锁对象, 不会阻塞
*
* @throws InterruptedException
* @throws KeeperException
* @throws NestableRuntimeException
*/
public boolean tryLock() throws KeeperException, NestableRuntimeException {
// 可能初始化的时候就失败了
if (exception != null) {
throw exception;
}

if (isOwner()) {//锁重入
return true;
}
reentrantLock.lock();
acquireLock(null);

if (exception != null) {
throw exception;
}

if (interrupt != null) {
Thread.currentThread().interrupt();
}

if (other != null) {
throw new NestableRuntimeException(other);
}

return isOwner();
}

/**
* 释放锁对象
*/
public void unlock() throws KeeperException {
if (id != null) {
try {
zookeeper.delete(root + "/" + id, -1);
reentrantLock.unlock();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (KeeperException.NoNodeException e) {
// do nothing
} finally {
id = null;
}
} else {
//do nothing
}
}

private void ensureExists(final String path) {
try {
Stat stat = zookeeper.exists(path, false);
if (stat != null) {
return;
}

zookeeper.create(path, data,Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
} catch (KeeperException e) {
exception = e;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
interrupt = e;
}
}

/**
* 返回锁对象对应的path
*/
public String getRoot() {
return root;
}

/**
* 判断当前是不是锁的owner
*/
public boolean isOwner() {
return id != null && ownerId != null && id.equals(ownerId);
}

/**
* 返回当前的节点id
*/
public String getId() {
return this.id;
}

// ===================== helper method =============================

/**
* 执行lock操作，允许传递watch变量控制是否需要阻塞lock操作
*/
private Boolean acquireLock(final BooleanMutex mutex) {
try {
do {
if (id == null) {//构建当前lock的唯一标识
long sessionId = zookeeper.getSessionId();
String prefix = "x-" + sessionId + "-";
//如果第一次，则创建一个节点
String path = zookeeper.create(root + "/" + prefix, data,
Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
int index = path.lastIndexOf("/");
id = StringUtils.substring(path, index + 1);
idName = new LockNode(id);
}

if (id != null) {
List<String> names = zookeeper.getChildren(root, false);
if (names.isEmpty()) {
id = null;//异常情况，重新创建一个
} else {
//对节点进行排序
SortedSet<LockNode> sortedNames = new TreeSet<LockNode>();
for (String name : names) {
sortedNames.add(new LockNode(name));
}

if (sortedNames.contains(idName) == false) {
id = null;//清空为null，重新创建一个
continue;
}

//将第一个节点做为ownerId
ownerId = sortedNames.first().getName();
if (mutex != null && isOwner()) {
mutex.set(true);//直接更新状态，返回
return true;
} else if (mutex == null) {
return isOwner();
}

SortedSet<LockNode> lessThanMe = sortedNames.headSet(idName);
if (!lessThanMe.isEmpty()) {
//关注一下排队在自己之前的最近的一个节点
LockNode lastChildName = lessThanMe.last();
lastChildId = lastChildName.getName();
//异步watcher处理
Stat stat=zookeeper.exists(root + "/" + lastChildId, new AsyncWatcher() {

public void asyncProcess(WatchedEvent event) {
acquireLock(mutex);
}

});

if (stat == null) {
acquireLock(mutex);// 如果节点不存在，需要自己重新触发一下，watcher不会被挂上去
}
} else {
if (isOwner()) {
mutex.set(true);
} else {
id = null;// 可能自己的节点已超时挂了，所以id和ownerId不相同
}
}
}
}
} while (id == null);
} catch (KeeperException e) {
exception = e;
if (mutex != null) {
mutex.set(true);
}
} catch (InterruptedException e) {
interrupt = e;
if (mutex != null) {
mutex.set(true);
}
} catch (Throwable e) {
other = e;
if (mutex != null) {
mutex.set(true);
}
}

if (isOwner() && mutex != null) {
mutex.set(true);
}
return Boolean.FALSE;
}

public class LockNode implements Comparable<LockNode>
{
private final String name;

private String       prefix;

private int          sequence = -1;

public LockNode(String name)
{
this.name = name;
this.prefix = name;
int idx = name.lastIndexOf('-');
if (idx >= 0)
{
this.prefix = name.substring(0, idx);
try
{
this.sequence = Integer.parseInt(name.substring(idx + 1));
}
catch (Exception e)
{
// ignore
}
}
}

public int compareTo(LockNode that)
{
int s1 = this.sequence;
int s2 = that.sequence;
if (s1 == -1 && s2 == -1)
{
return this.name.compareTo(that.name);
}
if (s1 == -1)
{
return -1;
}
else if (s2 == -1)
{
return 1;
}
else
{
return s1 - s2;
}
}

public String getName()
{
return name;
}

public int getSequence()
{
return sequence;
}

public String getPrefix()
{
return prefix;
}

public String toString()
{
return name.toString();
}

// ==================== hashcode & equals方法=======================
@Override
public int hashCode()
{
final int prime = 31;
int result = 1;
result = prime * result + ((name == null) ? 0 : name.hashCode());
return result;
}

@Override
public boolean equals(Object obj)
{
if (this == obj)
{
return true;
}
if (obj == null)
{
return false;
}
if (getClass() != obj.getClass())
{
return false;
}
LockNode other = (LockNode) obj;
if (name == null)
{
if (other.name != null)
{
return false;
}
}
else if (!name.equals(other.name))
{
return false;
}
return true;
}
}

public class BooleanMutex {

private Sync sync;

public BooleanMutex() {
sync = new Sync();
set(false);
}

public BooleanMutex(Boolean mutex) {
sync = new Sync();
set(mutex);
}

/**
* 阻塞等待Boolean为true
*
* @throws InterruptedException
*/
public void get() throws InterruptedException {
sync.innerGet();
}

/**
* 阻塞等待Boolean为true,允许设置超时时间
*
* @param timeout
* @param unit
* @throws InterruptedException
* @throws TimeoutException
*/
public void get(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
sync.innerGet(unit.toNanos(timeout));
}

/**
* 重新设置对应的Boolean mutex
*
* @param mutex
*/
public void set(Boolean mutex) {
if (mutex) {
sync.innerSetTrue();
} else {
sync.innerSetFalse();
}
}

public boolean state() {
return sync.innerState();
}

/**
* Synchronization control for BooleanMutex. Uses AQS sync state to
* represent run status
*/
private final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -7828117401763700385L;

/** State value representing that TRUE */
private static final int  TRUE             = 1;
/** State value representing that FALSE */
private static final int  FALSE            = 2;

private boolean isTrue(int state) {
return (state & TRUE) != 0;
}

/**
* 实现AQS的接口，获取共享锁的判断
*/
protected int tryAcquireShared(int state) {
// 如果为true，直接允许获取锁对象
// 如果为false，进入阻塞队列，等待被唤醒
return isTrue(getState()) ? 1 : -1;
}

/**
* 实现AQS的接口，释放共享锁的判断
*/
protected boolean tryReleaseShared(int ignore) {
//始终返回true，代表可以release
return true;
}

boolean innerState() {
return isTrue(getState());
}

void innerGet() throws InterruptedException {
acquireSharedInterruptibly(0);
}

void innerGet(long nanosTimeout) throws InterruptedException, TimeoutException {
if (!tryAcquireSharedNanos(0, nanosTimeout))
throw new TimeoutException();
}

void innerSetTrue() {
for (;;) {
int s = getState();
if (s == TRUE) {
return; //直接退出
}
if (compareAndSetState(s, TRUE)) {// cas更新状态，避免并发更新true操作
releaseShared(0);//释放一下锁对象，唤醒一下阻塞的Thread
}
}
}

void innerSetFalse() {
for (;;) {
int s = getState();
if (s == FALSE) {
return; //直接退出
}
if (compareAndSetState(s, FALSE)) {//cas更新状态，避免并发更新false操作
setState(FALSE);
}
}
}

}
}

public abstract class AsyncWatcher implements Watcher
{
private static final int       DEFAULT_POOL_SIZE    = 30;

private static final int       DEFAULT_ACCEPT_COUNT = 60;

private static ExecutorService executor             = new ThreadPoolExecutor(
1,
DEFAULT_POOL_SIZE,
0L,
TimeUnit.MILLISECONDS,
new ArrayBlockingQueue(
DEFAULT_ACCEPT_COUNT),
new NamedThreadFactory(
"Arbitrate-Async-Watcher"),
new ThreadPoolExecutor.CallerRunsPolicy());

public void process(final WatchedEvent event)
{
executor.execute(new Runnable()
{// 提交异步处理
@Override
public void run()
{
asyncProcess(event);
}
});
}

public abstract void asyncProcess(WatchedEvent event);
}

public class NestableRuntimeException extends Exception
{
public NestableRuntimeException(Throwable nest){
super(nest);
}
}