package com.ljq.lock;import java.util.concurrent.TimeUnit;public interface DistributedLock {/*** 获取锁，如果没有得到锁就一直等待* * @throws Exception*/public void acquire() throws Exception;/*** 获取锁，如果没有得到锁就一直等待直到超时* * @param time 超时时间* @param unit time参数时间单位* * @return 是否获取到锁* @throws Exception*/public boolean acquire(long time, TimeUnit unit) throws Exception;/*** 释放锁* * @throws Exception*/public void release() throws Exception;
}

package com.ljq.lock;import java.io.IOException;
import java.util.concurrent.TimeUnit;import org.I0Itec.zkclient.ZkClient;public class SimpleDistributedLock extends BaseDistributedLock implements DistributedLock {/** 用于保存Zookeeper中实现分布式锁的节点，如名称为locker：/locker，* 该节点应该是持久节点，在该节点下面创建临时顺序节点来实现分布式锁*/private final String basePath;/** 锁名称前缀，locker下创建的顺序节点例如都以lock-开头，这样便于过滤无关节点* 这样创建后的节点类似：lock-00000001，lock-000000002*/private static final String LOCK_NAME = "lock-";/* 用于保存某个客户端在locker下面创建成功的顺序节点，用于后续相关操作使用（如判断） */private String ourLockPath;/*** 传入Zookeeper客户端连接对象，和basePath* * @param client*            Zookeeper客户端连接对象* @param basePath*            basePath是一个持久节点*/public SimpleDistributedLock(ZkClient client, String basePath) {/** 调用父类的构造方法在Zookeeper中创建basePath节点，并且为basePath节点子节点设置前缀* 同时保存basePath的引用给当前类属性*/super(client, basePath, LOCK_NAME);this.basePath = basePath;}/*** 用于获取锁资源，通过父类的获取锁方法来获取锁* * @param time 获取锁的超时时间* @param unit 超时时间单位* * @return 是否获取到锁* @throws Exception*/private boolean internalLock(long time, TimeUnit unit) throws Exception {// 如果ourLockPath不为空则认为获取到了锁，具体实现细节见attemptLock的实现ourLockPath = attemptLock(time, unit);return ourLockPath != null;}/*** 获取锁，如果没有得到锁就一直等待* * @throws Exception*/public void acquire() throws Exception {// -1表示不设置超时时间，超时由Zookeeper决定if (!internalLock(-1, null)) {throw new IOException("连接丢失!在路径:'" + basePath + "'下不能获取锁!");}}/*** 获取锁，如果没有得到锁就一直等待直到超时* * @param time 超时时间* @param unit time参数时间单位* * @return 是否获取到锁* @throws Exception*/public boolean acquire(long time, TimeUnit unit) throws Exception {return internalLock(time, unit);}/*** 释放锁*/public void release() throws Exception {releaseLock(ourLockPath);System.out.println(ourLockPath + "锁已释放...");}
}

package com.ljq.lock;import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNoNodeException;public class BaseDistributedLock {private final ZkClient client; //Zookeeper客户端private final String basePath; //用于保存Zookeeper中实现分布式锁的节点，例如/locker节点，该节点是个持久节点，在该节点下面创建临时顺序节点来实现分布式锁private final String path; //同basePath变量一样private final String lockName; //锁名称前缀，/locker下创建的顺序节点，例如以lock-开头，这样便于过滤无关节点private static final Integer MAX_RETRY_COUNT = 10; //最大重试次数public BaseDistributedLock(ZkClient client, String path, String lockName) {this.client = client;this.basePath = path;this.path = path.concat("/").concat(lockName);this.lockName = lockName;}/*** 删除节点* * @param path * @throws Exception*/private void deletePath(String path) throws Exception {client.delete(path);}/*** 创建临时顺序节点* * @param client Zookeeper客户端* @param path 节点路径* @return* @throws Exception*/private String createEphemeralSequential(ZkClient client, String path) throws Exception {return client.createEphemeralSequential(path, null);}/*** 获取锁的核心方法* * @param startMillis 当前系统时间* @param millisToWait 超时时间* @param path * @return* @throws Exception*/private boolean waitToLock(long startMillis, Long millisToWait, String path) throws Exception {boolean haveTheLock = false; //获取锁标志boolean doDelete = false; //删除锁标志try {while (!haveTheLock) {// 获取/locker节点下的所有顺序节点，并且从小到大排序List<String> children = getSortedChildren();// 获取子节点，如：/locker/node_0000000003返回node_0000000003String sequenceNodeName = path.substring(basePath.length() + 1);// 计算刚才客户端创建的顺序节点在locker的所有子节点中排序位置，如果是排序为0，则表示获取到了锁int ourIndex = children.indexOf(sequenceNodeName);/** 如果在getSortedChildren中没有找到之前创建的[临时]顺序节点，这表示可能由于网络闪断而导致* Zookeeper认为连接断开而删除了我们创建的节点，此时需要抛出异常，让上一级去处理* 上一级的做法是捕获该异常，并且执行重试指定的次数，见后面的 attemptLock方法*/if (ourIndex < 0) {throw new ZkNoNodeException("节点没有找到: " + sequenceNodeName);}// 如果当前客户端创建的节点在locker子节点列表中位置大于0，表示其它客户端已经获取了锁// 此时当前客户端需要等待其它客户端释放锁boolean isGetTheLock = ourIndex == 0; //是否得到锁// 如何判断其它客户端是否已经释放了锁？从子节点列表中获取到比自己次小的那个节点，并对其建立监听String pathToWatch = isGetTheLock ? null : children.get(ourIndex - 1); //获取比自己次小的那个节点，如：node_0000000002if (isGetTheLock) {haveTheLock = true;} else {// 如果次小的节点被删除了，则表示当前客户端的节点应该是最小的了，所以使用CountDownLatch来实现等待String previousSequencePath = basePath.concat("/").concat(pathToWatch);final CountDownLatch latch = new CountDownLatch(1);final IZkDataListener previousListener = new IZkDataListener() {/*** 监听指定节点删除时触发该方法*/public void handleDataDeleted(String dataPath)throws Exception {// 次小节点删除事件发生时，让countDownLatch结束等待// 此时还需要重新让程序回到while，重新判断一次！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 nodebreak;}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) {deletePath(path);}}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;}/*** 获取parentPath节点下的所有顺序节点，并且从小到大排序* * @return* @throws Exception*/private 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();}}/*** 释放锁* * @param lockPath* @throws Exception*/protected void releaseLock(String lockPath) throws Exception {deletePath(lockPath);}/*** 尝试获取锁* * @param time* @param unit* @return* @throws Exception*/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 {// createLockNode用于在locker（basePath持久节点）下创建客户端要获取锁的[临时]顺序节点ourPath = createEphemeralSequential(client, path);/*** 该方法用于判断自己是否获取到了锁，即自己创建的顺序节点在locker的所有子节点中是否最小* 如果没有获取到锁，则等待其它客户端锁的释放，并且稍后重试直到获取到锁或者超时*/hasTheLock = waitToLock(startMillis, millisToWait, ourPath);} catch (ZkNoNodeException e) {if (retryCount++ < MAX_RETRY_COUNT) {isDone = false;} else {throw e;}}}System.out.println(ourPath + "锁获取" + (hasTheLock ? "成功" : "失败"));if (hasTheLock) {return ourPath;}return null;}
}

package com.ljq.lock;import org.I0Itec.zkclient.ZkClient;public class LockTest {public static void main(String[] args) throws Exception {ZkClient zkClient = new ZkClient("192.168.2.249:2181", 3000);SimpleDistributedLock simple = new SimpleDistributedLock(zkClient, "/locker");for (int i = 0; i < 10; i++) {try {simple.acquire();System.out.println("正在进行运算操作：" + System.currentTimeMillis());} catch (Exception e) {e.printStackTrace();} finally {simple.release();System.out.println("=================\r\n");}}}
}

/locker/lock-0000000131锁获取成功
正在进行运算操作：1479128867323
/locker/lock-0000000131锁已释放...
=================/locker/lock-0000000132锁获取成功
正在进行运算操作：1479128867424
/locker/lock-0000000132锁已释放...
=================/locker/lock-0000000133锁获取成功
正在进行运算操作：1479128867503
/locker/lock-0000000133锁已释放...
=================/locker/lock-0000000134锁获取成功
正在进行运算操作：1479128867577
/locker/lock-0000000134锁已释放...
=================/locker/lock-0000000135锁获取成功
正在进行运算操作：1479128867670
/locker/lock-0000000135锁已释放...
=================/locker/lock-0000000136锁获取成功
正在进行运算操作：1479128867744
/locker/lock-0000000136锁已释放...
=================/locker/lock-0000000137锁获取成功
正在进行运算操作：1479128867885
/locker/lock-0000000137锁已释放...
=================/locker/lock-0000000138锁获取成功
正在进行运算操作：1479128868108
/locker/lock-0000000138锁已释放...
=================/locker/lock-0000000139锁获取成功
正在进行运算操作：1479128868192
/locker/lock-0000000139锁已释放...
=================/locker/lock-0000000140锁获取成功
正在进行运算操作：1479128868286
/locker/lock-0000000140锁已释放...
=================