zookeeper分布式锁说明 第一步，使用zookeeper节点名称唯一性，用于分布式锁: 关于zookeeper集群的搭建，可以参考我之前写的一篇文章: ZooKeeper1 利用虚拟机搭建自己的ZooKeeper集群 zookeeper抽
第一步，使用zookeeper节点名称唯一性，用于分布式锁:
关于zookeeper集群的搭建，可以参考我之前写的一篇文章: ZooKeeper1 利用虚拟机搭建自己的ZooKeeper集群
zookeeper抽象出来的节点结构是一个和文件系统类似的小型的树状的目录结构，同时zookeeper机制规定：同一个目录下只能有一个唯一的文件名。例如：我们在zookeeper的根目录下，由两个客户端同时创建一个名为/myDistributeLock，只有一个客户端可以成功。
上述方案和memcached的add()方法、redis的setnx()方法实现分布式锁有着相同的思路。这样的方案实现起来如果不考虑搭建和维护zookeeper集群的成本，由于正确性和可靠性是zookeeper机制自己保证的，实现还是比较简单的。
第二步，使用zookeeper临时顺序节点，用于分布式锁:
在讨论这套方案之前，我们有必要先“吹毛求疵”般的说明一下使用zookeeper节点名称唯一性来做分布式锁这个方案的缺点。比如，当许多线程在等待一个锁时，如果锁得到释放的时候，那么所有客户端都被唤醒，但是仅仅有一个客户端得到锁。在这个过程中，大量的线程根本没有获得锁的可能性，但是也会引起大量的上下文切换，这个系统开销也是不小的，对于这样的现象有一个专业名词，称之为“惊群效应”。
我们首先说明一下zookeeper的顺序节点、临时节点和watcher机制:
所谓顺序节点，假如我们在/myDisLocks/目录下创建3个节点，zookeeper集群会按照发起创建的顺序来创建节点，节点分别为/myDisLocks/0000000001、/myDisLocks/0000000002、/myDisLocks/0000000003。
所谓临时节点，临时节点由某个客户端创建，当客户端与zookeeper集群断开连接，则该节点自动被删除。
所谓对于watcher机制，大家可以参考Apache ZooKeeper Watcher机制源码解释。当然如果你之前不知道watcher机制是个什么东东，不建议你直接去看前边我提供的文章链接，这样你极有可能忘掉我们的讨论主线，即分布式锁的实现方案，而陷入到watcher机制的源码实现中。所以你也可以先看看下面的具体方案，猜测一下watcher是用来干嘛的，我这里先总结一句话做个引子: 所谓watcher机制，你可以简单一点儿理解成任何一个连接zookeeper的客户端可以通过watcher机制关注自己感兴趣的节点的增删改查，当这个节点发生增删改查的操作时，会“广播”自己的消息，所有对此感兴趣的节点可以在收到这些消息后，根据自己的业务需要执行后续的操作。
具体的使用步骤如下:
1. 每个业务线程调用create()方法创建名为“/myDisLocks/thread”的节点，需要注意的是，这里节点的创建类型需要设置为EPHEMERAL_SEQUENTIAL，即节点类型为临时顺序节点。此时/myDisLocks节点下会出现诸如/myDisLocks/thread0000000001、/myDisLocks/thread0000000002、/myDisLocks/thread0000000003这样的子节点。
2. 每个业务线程调用getChildren(“myDisLocks”)方法来获取/myDisLocks这个节点下所有已经创建的子节点。
3. 每个业务线程获取到所有子节点的路径之后，如果发现自己在步骤1中创建的节点的尾缀编号是所有节点中序号最小的，那么就认为自己获得了锁。
4. 如果在步骤3中发现自己并非是所有子节点中序号最小的，说明自己还没有获取到锁。使用watcher机制监视比自己创建节点的序列号小的节点（比自己创建的节点小的最大节点），进入等待。比如，如果当前业务线程创建的节点是/myDisLocks/thread0000000003，那么在没有获取到锁的情况下，他只需要监视/myDisLocks/thread0000000002的情况。只有当/myDisLocks/thread0000000002获取到锁并释放之后，当前业务线程才启动获取锁，这样可以避免一个业务线程释放锁之后，其他所有线程都去竞争锁，引起不必要的上下文切换，最终造成“惊群现象”。
5. 释放锁的过程相对比较简单，就是删除自己创建的那个子节点即可。
注意: 这个方案实现的分布式锁还带着一点儿公平锁的味道！为什么呢？我们在利用每个节点的序号进行排队以此来避免进群现象时，实际上所有业务线程获得锁的顺序就是自己创建节点的顺序，也就是哪个业务线程先来，哪个就可以最快获得锁。
下面贴出我自己实现的上述方案的代码:
分布式锁2
public class LockWatcher implements Watcher {
/** 成员变量 **/
private ZooKeeper zk = null;
// 当前业务线程竞争锁的时候创建的节点路径
private String selfPath = null;
// 当前业务线程竞争锁的时候创建节点的前置节点路径
private String waitPath = null;
// 确保连接zk成功；只有当收到Watcher的监听事件之后，才执行后续的操作，否则请求阻塞在createConnection()创建ZK连接的方法中
private CountDownLatch connectSuccessLatch = new CountDownLatch(1);
// 标识线程是否执行完任务
private CountDownLatch threadCompleteLatch = null;
/** ZK的相关配置常量 **/
private static final String LOCK_ROOT_PATH = "/myDisLocks";
private static final String LOCK_SUB_PATH = LOCK_ROOT_PATH + "/thread";
public LockWatcher(CountDownLatch latch) {
this.threadCompleteLatch = latch;
}
@Override
public void process(WatchedEvent event) {
if (event == null) {
return;
}
// 通知状态
Event.KeeperState keeperState = event.getState();
// 事件类型
Event.EventType eventType = event.getType();
// 根据通知状态分别处理
if (Event.KeeperState.SyncConnected == keeperState) {
if ( Event.EventType.None == eventType ) {
System.out.println(Thread.currentThread().getName() + "成功连接上ZK服务器");
// 此处代码的主要作用是用来辅助判断当前线程确实已经连接上ZK
connectSuccessLatch.countDown();
}else if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
System.out.println(Thread.currentThread().getName() + "收到情报，排我前面的家伙已挂，我准备再次确认我是不是最小的节点！？");
try {
if(checkMinPath()){
getLockSuccess();
}
} catch (Exception e) {
e.printStackTrace();
}
}
} else if ( Event.KeeperState.Disconnected == keeperState ) {
System.out.println(Thread.currentThread().getName() + "与ZK服务器断开连接");
} else if ( Event.KeeperState.AuthFailed == keeperState ) {
System.out.println(Thread.currentThread().getName() + "权限检查失败");
} else if ( Event.KeeperState.Expired == keeperState ) {
System.out.println(Thread.currentThread().getName() + "会话失效");
}
}
/**
* @author zhangyi03
* @date 2017-5-23 下午6:07:03
* @description 创建ZK连接
* @param connectString ZK服务器地址列表
* @param sessionTimeout Session超时时间
* @throws IOException
* @throws InterruptedException
*/
public void createConnection(String connectString, int sessionTimeout) throws IOException, InterruptedException {
zk = new ZooKeeper(connectString, sessionTimeout, this);
// connectSuccessLatch.await(1, TimeUnit.SECONDS) 正式实现的时候可以考虑此处是否采用超时阻塞
connectSuccessLatch.await();
}
/**
* @author zhangyi03
* @date 2017-5-23 下午6:15:48
* @description 创建ZK节点
* @param path 节点path
* @param data 初始数据内容
* @param needWatch
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public boolean createPersistentPath(String path, String data, boolean needWatch) throws KeeperException, InterruptedException {
if(zk.exists(path, needWatch) == null){
String result = zk.create( path,data.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
System.out.println(Thread.currentThread().getName() + "创建节点成功, path: " + result + ", content: " + data);
}
return true;
}
/**
* @author zhangyi03
* @date 2017-5-23 下午6:24:46
* @description 获取分布式锁
* @throws KeeperException
* @throws InterruptedException
*/
public void getLock() throws Exception {
selfPath = zk.create(LOCK_SUB_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(Thread.currentThread().getName() + "创建锁路径:" + selfPath);
if(checkMinPath()){
getLockSuccess();
}
}
/**
* @author zhangyi03
* @date 2017-5-23 下午7:02:41
* @description 获取锁成功
* @throws KeeperException
* @throws InterruptedException
*/
private void getLockSuccess() throws KeeperException, InterruptedException {
if(zk.exists(selfPath, false) == null){
System.err.println(Thread.currentThread().getName() + "本节点已不在了...");
return;
}
System.out.println(Thread.currentThread().getName() + "获取锁成功，开始处理业务数据！");
Thread.sleep(2000);
System.out.println(Thread.currentThread().getName() + "处理业务数据完成，删除本节点：" + selfPath);
zk.delete(selfPath, -1);
releaseConnection();
threadCompleteLatch.countDown();
}
/**
* @author zhangyi03
* @date 2017-5-23 下午7:06:46
* @description 关闭ZK连接
*/
private void releaseConnection() {
if (zk != null) {
try {
zk.close();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println(Thread.currentThread().getName() + "释放ZK连接");
}
/**
* @author zhangyi03
* @date 2017-5-23 下午6:57:14
* @description 检查自己是不是最小的节点
* @param selfPath
* @return
* @throws KeeperException
* @throws InterruptedException
*/
private boolean checkMinPath() throws Exception {
List
subNodes = zk.getChildren(LOCK_ROOT_PATH, false);
// 根据元素按字典序升序排序
Collections.sort(subNodes);
System.err.println(Thread.currentThread().getName() + "创建的临时节点名称:" + selfPath.substring(LOCK_ROOT_PATH.length()+1));
int index = subNodes.indexOf(selfPath.substring(LOCK_ROOT_PATH.length()+1));
System.err.println(Thread.currentThread().getName() + "创建的临时节点的index:" + index);
switch (index){
case -1: {
System.err.println(Thread.currentThread().getName() + "创建的节点已不在了..." + selfPath);
return false;
}
case 0:{
System.out.println(Thread.currentThread().getName() + "子节点中，我果然是老大" + selfPath);
return true;
}
default:{
// 获取比当前节点小的前置节点,此处只关注前置节点是否还在存在，避免惊群现象产生
waitPath = LOCK_ROOT_PATH +"/"+ subNodes.get(index - 1);
System.out.println(Thread.currentThread().getName() + "获取子节点中，排在我前面的节点是:" + waitPath);
try {
zk.getData(waitPath, true, new Stat());
return false;
} catch (Exception e) {
if (zk.exists(waitPath, false) == null) {
System.out.println(Thread.currentThread().getName() + "子节点中，排在我前面的" + waitPath + "已失踪，该我了");
return checkMinPath();
} else {
throw e;
}
}
}
}
}
}
分布式锁1
public class MyDistributedLockByZK {
/** 线程池 **/
private static ExecutorService executorService = null;
private static final int THREAD_NUM = 5;
private static int threadNo = 0;
private static CountDownLatch threadCompleteLatch = new CountDownLatch(THREAD_NUM);
/** ZK的相关配置常量 **/
private static final String CONNECTION_STRING = "192.168.224.170:2181";
private static final int SESSION_TIMEOUT = 10000;
// 此变量在LockWatcher中也有一个同名的静态变量，正式使用的时候，提取到常量类中共同维护即可。
private static final String LOCK_ROOT_PATH = "/myDisLocks";
public static void main(String[] args) {
// 定义线程池
executorService = Executors.newFixedThreadPool(THREAD_NUM, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
String name = String.format("第[%s]个测试线程", ++threadNo);
Thread ret = new Thread(Thread.currentThread().getThreadGroup(), r, name, 0);
ret.setDaemon(false);
return ret;
}
});
// 启动线程
if (executorService != null) {
startProcess();
}
}
/**
* @author zhangyi03
* @date 2017-5-23 下午5:57:27
* @description 模拟并发执行任务
*/
public static void startProcess() {
Runnable disposeBusinessRunnable= new Thread(new Runnable() {
public void run() {
String threadName = Thread.currentThread().getName();
LockWatcher lock = new LockWatcher(threadCompleteLatch);
try {
/** 步骤1: 当前线程创建ZK连接 **/
lock.createConnection(CONNECTION_STRING, SESSION_TIMEOUT);
/** 步骤2: 创建锁的根节点 **/
// 注意，此处创建根节点的方式其实完全可以在初始化的时候由主线程单独进行根节点的创建，没有必要在业务线程中创建。
// 这里这样写只是一种思路而已，不必局限于此
synchronized (MyDistributedLockByZK.class){
lock.createPersistentPath(LOCK_ROOT_PATH, "该节点由" + threadName + "创建", true);
}
/** 步骤3: 开启锁竞争并执行任务 **/
lock.getLock();
} catch (Exception e) {
e.printStackTrace();
}
}
});
for (int i = 0; i < THREAD_NUM; i++) {
executorService.execute(disposeBusinessRunnable);
}
executorService.shutdown();
try {
threadCompleteLatch.await();
System.out.println("所有线程运行结束!");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}