摘要：分享牛原创，zookeeper使用，zookeeper锁在实际项目开发中还是很常用的，在这里我们介绍一下zookeeper分布式锁的使用，以及我们如何zookeeper分布式锁的原理。zookeeper节点理解。
zookeeper分布式锁有什么用呢？首先要明白锁是一个什么东西？举个通俗的例子，把门锁着了，外面的人进不去，里面的人可以随时出来，出来之后，还可以继续加锁。比如我们项目中，主要有供应商系统锁库存这种情况，锁库存的时候不能让其他的人去修改库存信息。这里就需要使用的时候加锁。当然了也可以使用数据库或者redis版本锁的概念，根据版本去区分到底如何锁库。
首先我们看一下zookeeper节点类型。
zookeeper节点类型分为以下四种：
1.1.1. 节点说明
public enum CreateMode {
/**
* The znode will not be automatically deleted upon client's disconnect.
*/
PERSISTENT (0, false, false),
/**
* The znode will not be automatically deleted upon client's disconnect,
* and its name will be appended with a monotonically increasing number.
*/
PERSISTENT_SEQUENTIAL (2, false, true),
/**
* The znode will be deleted upon the client's disconnect.
*/
EPHEMERAL (1, true, false),
/**
* The znode will be deleted upon the client's disconnect, and its name
* will be appended with a monotonically increasing number.
*/
EPHEMERAL_SEQUENTIAL (3, true, true);
}
从持久化的层次划分：
1.持久化节点：不删除节点永远存在。
2.非持久节点，换言之就是临时节点，临时节点就是客户端连接的时候创建，客户端挂起的时候，临时节点自动删除。
从排序层次划分：
1.持久有序。
2.持久无序，
3.临时有序。
4.临时无序。
这里需要注意持久化节点可以创建子节点。非持久化节点不能创建子节点。这里可以自己去使用命令去测试。
非持久节点就是创建的时候存在，消失的时候，节点自动删除，所以我们利用这个特性，实现我们的需求，比如，我可以程序启动的时候在指定的持久节点，创建临时节点，当程序挂掉的时候，临时节点消失，我们可以一直监控指定父节点中的子节点集合，就可以监控程序的健康状态。
1.1.2. zookeeper 分布式锁实现
接下来，在上面理解节点的基础之上，我们可以去实现zookeeper 分布式锁。具体怎么实现呢？思路如下：
我们可以创建一个持久节点，在程序中我们每次创建临时子节点，然后我们遍历持久节点下面的子节点，因为临时节点我们设置的时候是有序的。所以我们可以加锁的时候，创建了一个临时有序节点，当我们加锁完成自己的业务之后，释放锁，然后，这个删除临时节点，所以设计的核心点就是:
1.创建父节点。持久节点
2.创建有序的临时子节点。
3.删除临时节点，当业务完成的时候。
4.每次需要判断自身的临时节点，是否是最小的。为什么要判断是最小的呢？因为不是最小的话，说明前面还有一些节点在加锁执行中，所以我们这个节点不能加锁执行。
5.怎么让自身节点监听执行呢？因为如果自身节点是最小的，可以直接执行，如果不是最小的。要监听前面的节点是否已经删除，如果其他的前面的节点都删除了。则自己就可以加锁执行业务代码了。
下面开始书写我们的代码吧？
package com.shareniu.zkTest;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
public class ShareniuDistributedLock implements Watcher {
private int threadId;
// 主要区分线程
private static String PREFIX_OF_THREAD = null;
// 子节点的前缀
private static final String EPHEMERAL_SEQUENTIAL_PATH = "/shareniuLock/sub";
// 父节点
private static final String PARENT_PATH = "/shareniuLock";
protected static final String CONNECTION_STRING = "101.201.xx.xx:2181";
protected static final int SESSION_TIMEOUT = 10000;
//开启的线程的数量
private static final int THREAD_NUM = 5;
// 创建临时节点自身
private String selfPath;
// zk连接对象
public ShareniuDistributedLock(int threadId) {
this.threadId = threadId;
PREFIX_OF_THREAD = "【第"+threadId+"个线程】";
}
private ZooKeeper zooKeeper = null;
private CountDownLatch cdl = new CountDownLatch(1);
private static final CountDownLatch threadSemaphore = new CountDownLatch(THREAD_NUM);
private String waitPath;
public static void main(String[] args) {
for(int i=0; i < THREAD_NUM; i++){
final int threadId = i+1;
new Thread(){
@Override
public void run() {
try{
ShareniuDistributedLock dc = new ShareniuDistributedLock(threadId);
dc.createConnection(CONNECTION_STRING, SESSION_TIMEOUT);
//GROUP_PATH不存在的话，由一个线程创建即可；
synchronized (threadSemaphore){
dc.createPath(PARENT_PATH, "该节点由线程" + threadId + "创建", true);
}
dc.getLock();
} catch (Exception e){
e.printStackTrace();
}
}
}.start();
}
try {
threadSemaphore.await();
System.out.println("所有线程运行结束!");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 实现监听器中的方法拿到WatchedEvent对象
*/
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( PREFIX_OF_THREAD + "成功连接上ZK服务器" );
cdl.countDown();
}else if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
try {
if(checkMinPathOfChilde()){
getLockByShelf();
}
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}else if ( Event.KeeperState.Disconnected == keeperState ) {
System.out.println( PREFIX_OF_THREAD + "与ZK服务器断开连接" );
} else if ( Event.KeeperState.AuthFailed == keeperState ) {
System.out.println( PREFIX_OF_THREAD + "权限检查失败" );
} else if ( Event.KeeperState.Expired == keeperState ) {
System.out.println( PREFIX_OF_THREAD + "会话失效" );
}
}
/**
* 关闭ZK连接
*/
public void releaseConnection() {
if (this.zooKeeper != null) {
try {
this.zooKeeper.close();
} catch (InterruptedException e) {
}
}
System.out.println(PREFIX_OF_THREAD + "释放连接");
}
/***
* 创建连接
*
* @param connectString
* 连接的服务器字符串
* @param sessionTimeout
* 超时时间
* @throws IOException
* @throws InterruptedException
*/
public void createConnection(String connectString, int sessionTimeout)
throws IOException, InterruptedException {
zooKeeper = new ZooKeeper(connectString, sessionTimeout, this);
System.out.println("打开连接........");
// 因为打开连接只需要一次，这里为了防止并发，使用的CountDownLatch对象
// 程序一直等待，直到cdl.countDown();方法的调用
cdl.await();
}
/**
* 创建节点
*
* @param path
* 路径
* @param data
* 内容
* @param needWatch
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public boolean createPath(String path, String data, boolean needWatch)
throws KeeperException, InterruptedException {
// 判断节点是否存在存在就不要创建了。
Stat exists = zooKeeper.exists(path, needWatch);
if (exists == null) {
// 节点不存在
System.out.println(PREFIX_OF_THREAD
+ "节点创建成功, Path: "
+ this.zooKeeper.create(path, data.getBytes(),
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT)
+ ", content: " + data);
}
return true;
}
/**
* 获取锁
*
* @throws KeeperException
* @throws InterruptedException
*/
private void getLock() throws KeeperException, InterruptedException {
// 创建临时节点
selfPath = zooKeeper.create(EPHEMERAL_SEQUENTIAL_PATH, null,
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
System.out.println("创建临时节点：" + selfPath);
if (checkMinPathOfChilde()) {
getLockByShelf();
}
}
/**
* 获取到锁了
*
* @throws InterruptedException
* @throws KeeperException
*/
private void getLockByShelf() throws KeeperException, InterruptedException {
// 获取到锁了 开始 执行代码 删除节点 释放连接
if (zooKeeper.exists(this.selfPath, false) == null) {
System.out.println(PREFIX_OF_THREAD + "本节点已不在了...");
return;
} else {
// 节点存在
System.out.println(PREFIX_OF_THREAD + "获取锁成功....");
// 休息一下
Thread.sleep(1000);
System.out.println(PREFIX_OF_THREAD + "删除临时节点：" + selfPath);
// 删除的时候，版本是-1就是所有的都删除
zooKeeper.delete(this.selfPath, -1);
releaseConnection();
// 释放锁 其他的程序可以 继续打开连接
cdl.countDown();
}
}
/**
* 获取锁
*
* @return
* @throws InterruptedException
* @throws KeeperException
*/
private boolean checkMinPathOfChilde() throws KeeperException,
InterruptedException {
// 获取父节点中的所有子节点
List<String> subNodes = zooKeeper.getChildren(PARENT_PATH, false);
// 临时节点是有序的，那就排序找最小的吧
Collections.sort(subNodes);
int index = subNodes
.indexOf(selfPath.substring(PARENT_PATH.length() + 1));
switch (index) {
case -1: {
System.out.println(PREFIX_OF_THREAD + "节点已不在了..." + selfPath);
return false;
}
case 0: {
System.out.println(PREFIX_OF_THREAD + "自己可以获取锁执行代码了" + selfPath);
return true;
}
default: {
this.waitPath = PARENT_PATH + "/" + subNodes.get(index - 1);
System.out.println(PREFIX_OF_THREAD + "前面的节点" + waitPath);
try {
zooKeeper.getData(waitPath, true, new Stat());
return false;
} catch (KeeperException e) {
if (zooKeeper.exists(waitPath, false) == null) {
System.out.println(PREFIX_OF_THREAD + "本节点前面的节点："
+ waitPath + "");
// 递归找吧
return checkMinPathOfChilde();
} else {
throw e;
}
}
}
}
}
}
1.1.3. 程序的输出
程序的输出如下：
打开连接........
打开连接........
打开连接........
打开连接........
打开连接........
【第5个线程】成功连接上ZK服务器
【第5个线程】成功连接上ZK服务器
【第5个线程】成功连接上ZK服务器
【第5个线程】成功连接上ZK服务器
【第5个线程】成功连接上ZK服务器
创建临时节点：/shareniuLock/sub0000000001
【第5个线程】自己可以获取锁执行代码了/shareniuLock/sub0000000001
创建临时节点：/shareniuLock/sub0000000002
【第5个线程】获取锁成功....
【第5个线程】前面的节点/shareniuLock/sub0000000001
创建临时节点：/shareniuLock/sub0000000003
创建临时节点：/shareniuLock/sub0000000004
【第5个线程】前面的节点/shareniuLock/sub0000000002
【第5个线程】前面的节点/shareniuLock/sub0000000003
创建临时节点：/shareniuLock/sub0000000005
【第5个线程】前面的节点/shareniuLock/sub0000000004
【第5个线程】删除临时节点：/shareniuLock/sub0000000001
【第5个线程】释放连接
【第5个线程】自己可以获取锁执行代码了/shareniuLock/sub0000000002
【第5个线程】获取锁成功....
【第5个线程】删除临时节点：/shareniuLock/sub0000000002
【第5个线程】释放连接
【第5个线程】自己可以获取锁执行代码了/shareniuLock/sub0000000003
【第5个线程】获取锁成功....
【第5个线程】删除临时节点：/shareniuLock/sub0000000003
【第5个线程】释放连接
【第5个线程】自己可以获取锁执行代码了/shareniuLock/sub0000000004
【第5个线程】获取锁成功....
【第5个线程】删除临时节点：/shareniuLock/sub0000000004
【第5个线程】释放连接
【第5个线程】自己可以获取锁执行代码了/shareniuLock/sub0000000005
【第5个线程】获取锁成功....
【第5个线程】删除临时节点：/shareniuLock/sub0000000005
【第5个线程】释放连接
1.1.4. zk节点的查看
下面我们看一下zk的节点，
在此证明，节点是有序的，释放锁的时候，会删除临时节点。
分享牛原创(尊重原创 转载对的时候第一行请注明，转载出处来自分享牛http://blog.csdn.net/qq_30739519)
