基于zookeeper的分布式锁实现

xiaowei8782088

　　工作中需要写一个定时任务,由于是集群环境,自然而然想到需要通过分布式锁来保证单台执行..相信大家都会想到使用zk来实现对应的分布式锁.下面就简单介绍一下几种实现

准备工作
　　有几个帮助类,先把代码放上来
　　ZKClient 对zk的操作做了一个简单的封装

package zk.lock;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import zk.util.ZKUtil;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
/**
* User: zhenghui
* Date: 14-3-26
* Time: 下午8:50
* 封装一个zookeeper实例.
*/
public class ZKClient implements Watcher {
private ZooKeeper zookeeper;
private CountDownLatch connectedSemaphore = new CountDownLatch(1);

public ZKClient(String connectString, int sessionTimeout) throws Exception {
zookeeper = new ZooKeeper(connectString, sessionTimeout, this);
System.out.println("connecting zk server");
if (connectedSemaphore.await(10l, TimeUnit.SECONDS)) {
System.out.println("connect zk server success");
} else {
System.out.println("connect zk server error.");
throw new Exception("connect zk server error.");
}
}
public void close() throws InterruptedException {
if (zookeeper != null) {
zookeeper.close();
}
}
public void createPathIfAbsent(String path, boolean isPersistent) throws Exception {
CreateMode createMode = isPersistent ? CreateMode.PERSISTENT : CreateMode.EPHEMERAL;
path = ZKUtil.normalize(path);
if (!this.exists(path)) {
zookeeper.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, createMode);
}
}
public boolean exists(String path) throws Exception {
path = ZKUtil.normalize(path);
Stat stat = zookeeper.exists(path, null);
return stat != null;
}
public String getData(String path) throws Exception {
path = ZKUtil.normalize(path);
try {
byte[] data = zookeeper.getData(path, null, null);
return new String(data);
} catch (KeeperException e) {
if (e instanceof KeeperException.NoNodeException) {
throw new Exception("Node does not exist,path is [" + e.getPath() + "].", e);
} else {
throw new Exception(e);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new Exception(e);
}
}
@Override
public void process(WatchedEvent event) {
if (event == null) return;
// 连接状态
Watcher.Event.KeeperState keeperState = event.getState();
// 事件类型
Watcher.Event.EventType eventType = event.getType();
// 受影响的path
//        String path = event.getPath();
if (Watcher.Event.KeeperState.SyncConnected == keeperState) {
// 成功连接上ZK服务器
if (Watcher.Event.EventType.None == eventType) {
System.out.println("zookeeper connect success");
connectedSemaphore.countDown();
}
}
//下面可以做一些重连的工作.
else if (Watcher.Event.KeeperState.Disconnected == keeperState) {
System.out.println("zookeeper Disconnected");
} else if (Watcher.Event.KeeperState.AuthFailed == keeperState) {
System.out.println("zookeeper AuthFailed");
} else if (Watcher.Event.KeeperState.Expired == keeperState) {
System.out.println("zookeeper Expired");
}
}
}

　　ZKUtil 针对zk路径的一个工具类

package zk.util;
/**
* User: zhenghui
* Date: 14-3-26
* Time: 下午9:56
*/
public class ZKUtil {
public static final String SEPARATOR = "/";
/**
* 转换path为zk的标准路径 以/开头,最后不带/
*/
public static String normalize(String path) {
String temp = path;
if(!path.startsWith(SEPARATOR)) {
temp = SEPARATOR + path;
}
if(path.endsWith(SEPARATOR)) {
temp = temp.substring(0, temp.length()-1);
return normalize(temp);
}else {
return temp;
}
}
/**
* 链接两个path,并转化为zk的标准路径
*/
public static String contact(String path1,String path2){
if(path2.startsWith(SEPARATOR)) {
path2 = path2.substring(1);
}
if(path1.endsWith(SEPARATOR)) {
return normalize(path1 + path2);
} else {
return normalize(path1 + SEPARATOR + path2);
}
}
/**
* 字符串转化成byte类型
*/
public static byte[] toBytes(String data) {
if(data == null || data.trim().equals("")) return null;
return data.getBytes();
}
}

　　NetworkUtil 获取本机IP的工具方法

package zk.util;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.util.Enumeration;
/**
* User: zhenghui
* Date: 14-4-1
* Time: 下午4:47
*/
public class NetworkUtil {
static private final char COLON = ':';
/**
* 获取当前机器ip地址
* 据说多网卡的时候会有问题.
*/
public static String getNetworkAddress() {
Enumeration<NetworkInterface> netInterfaces;
try {
netInterfaces = NetworkInterface.getNetworkInterfaces();
InetAddress ip;
while (netInterfaces.hasMoreElements()) {
NetworkInterface ni = netInterfaces
.nextElement();
Enumeration<InetAddress> addresses=ni.getInetAddresses();
while(addresses.hasMoreElements()){
ip = addresses.nextElement();
if (!ip.isLoopbackAddress()
&& ip.getHostAddress().indexOf(COLON) == -1) {
return ip.getHostAddress();
}
}
}
return "";
} catch (Exception e) {
return "";
}
}
}

　　--------------------------- 正文开始  -----------------------------------
　　这种实现非常简单,具体的流程如下
　　

 对应的实现如下

package zk.lock;

import zk.util.NetworkUtil;
import zk.util.ZKUtil;
/**
* User: zhenghui
* Date: 14-3-26
* Time: 下午8:37
* 分布式锁实现.
*
* 这种实现的原理是,创建某一个任务的节点,比如 /lock/tasckname 然后获取对应的值,如果是当前的Ip,那么获得锁,如果不是,则没获得
* .如果该节点不存在,则创建该节点,并把改节点的值设置成当前的IP
*/
public class DistributedLock01 {
private ZKClient zkClient;

public static final String LOCK_ROOT = "/lock";
private String lockName;

public DistributedLock01(String connectString, int sessionTimeout,String lockName) throws Exception {
//先创建zk链接.
this.createConnection(connectString,sessionTimeout);
this.lockName = lockName;
}
public boolean tryLock(){
String path = ZKUtil.contact(LOCK_ROOT,lockName);
String localIp = NetworkUtil.getNetworkAddress();
try {
if(zkClient.exists(path)){
String ownnerIp = zkClient.getData(path);
if(localIp.equals(ownnerIp)){
return true;
}
} else {
zkClient.createPathIfAbsent(path,false);
if(zkClient.exists(path)){
String ownnerIp = zkClient.getData(path);
if(localIp.equals(ownnerIp)){
return true;
}
}
}
} catch (Exception e) {
e.printStackTrace();
}
return false;
}

/**
* 创建zk连接
*
*/
protected void createConnection(String connectString, int sessionTimeout) throws Exception {
if(zkClient != null){
releaseConnection();
}
zkClient = new ZKClient(connectString,sessionTimeout);
zkClient.createPathIfAbsent(LOCK_ROOT,true);
}
/**
* 关闭ZK连接
*/
protected void releaseConnection() throws InterruptedException {
if (zkClient != null) {
zkClient.close();
}
}
}


总结
　　网上有很多文章,大家的方法大多数都是创建一个root根节点,每一个trylock的客户端都会在root下创建一个 EPHEMERAL_SEQUENTIAL 的子节点,同时设置root的child 变更watcher(为了避免羊群效应,可以只添加前一个节点的变更通知) .如果创建的节点的序号是最小,则获取到锁,否则继续等待root的child 变更..这种方式我个人觉得有点繁琐,实现起来有点麻烦.具体的现有实现可以查看
　　https://svn.apache.org/repos/asf/zookeeper/trunk/src/recipes/lock/
　　其他的,想了半天没啥好说的..over