zookeeper系列（五）实战分布式锁
zookeeper系列（一）zookeeper必知
zookeeper系列（二）实战master选举
zookeeper系列（三）实战数据发布订阅
zookeeper系列（四）实战负载均衡
zookeeper系列（五）实战分布式锁
zookeeper系列（六）实战分布式队列
zookeeper系列（七）实战分布式命名服务
zookeeper系列（八）zookeeper运维java
分布式锁
咱们常说的锁是单进程多线程锁，在多线程并发编程中，用于线程之间的数据同步，保护共享资源的访问。而分布式锁，指在分布式环境下，保护跨进程、跨主机、跨网络的共享资源，实现互斥访问，保证一致性。node
架构图
左侧是zookeeper集群，locker是数据节点，node_1到node_n表明一系列的顺序节点。编程
右侧client_1至client_n表明客户端，Service表明须要互斥访问的服务。segmentfault
总实现思路，是在获取锁的时候在locker节点下建立顺序节点，在释放锁的时候，把本身建立的节点删除。网络
流程图
类图
代码实现
public interface DistributedLock {
/*
* 获取锁，若是没有获得就等待
*/
public void acquire() throws Exception;
/*
* 获取锁，直到超时
*/
public boolean acquire(long time, TimeUnit unit) throws Exception;
/*
* 释放锁
*/
public void release() throws Exception;
}
public class SimpleDistributedLockMutex extends BaseDistributedLock implements
DistributedLock {
//锁名称前缀,成功建立的顺序节点如lock-0000000000,lock-0000000001,...
private static final String LOCK_NAME = "lock-";
// zookeeper中locker节点的路径
private final String basePath;
// 获取锁之后本身建立的那个顺序节点的路径
private String ourLockPath;
private boolean internalLock(long time, TimeUnit unit) throws Exception {
ourLockPath = attemptLock(time, unit);
return ourLockPath != null;
}
public SimpleDistributedLockMutex(ZkClientExt client, String basePath){
super(client,basePath,LOCK_NAME);
this.basePath = basePath;
}
// 获取锁
public void acquire() throws Exception {
if ( !internalLock(-1, null) ) {
throw new IOException("链接丢失!在路径:'"+basePath+"'下不能获取锁!");
}
}
// 获取锁，能够超时
public boolean acquire(long time, TimeUnit unit) throws Exception {
return internalLock(time, unit);
}
// 释放锁
public void release() throws Exception {
releaseLock(ourLockPath);
}
}
public class BaseDistributedLock {
private final ZkClientExt client;
private final String path;
private final String basePath;
private final String lockName;
private static final Integer MAX_RETRY_COUNT = 10;
public BaseDistributedLock(ZkClientExt client, String path, String lockName){
this.client = client;
this.basePath = path;
this.path = path.concat("/").concat(lockName);
this.lockName = lockName;
}
// 删除成功获取锁以后所建立的那个顺序节点
private void deleteOurPath(String ourPath) throws Exception{
client.delete(ourPath);
}
// 建立临时顺序节点
private String createLockNode(ZkClient client, String path) throws Exception{
return client.createEphemeralSequential(path, null);
}
// 等待比本身次小的顺序节点的删除
private boolean waitToLock(long startMillis, Long millisToWait, String ourPath) throws Exception{
boolean haveTheLock = false;
boolean doDelete = false;
try {
while ( !haveTheLock ) {
// 获取/locker下的通过排序的子节点列表
List<String> children = getSortedChildren();
// 获取刚才本身建立的那个顺序节点名
String sequenceNodeName = ourPath.substring(basePath.length()+1);
// 判断本身排第几个
int ourIndex = children.indexOf(sequenceNodeName);
if (ourIndex < 0){ // 网络抖动，获取到的子节点列表里可能已经没有本身了
throw new ZkNoNodeException("节点没有找到: " + sequenceNodeName);
}
// 若是是第一个，表明本身已经得到了锁
boolean isGetTheLock = ourIndex == 0;
// 若是本身没有得到锁，则要watch比咱们次小的那个节点
String pathToWatch = isGetTheLock ? null : children.get(ourIndex - 1);
if ( isGetTheLock ){
haveTheLock = true;
} else {
// 订阅比本身次小顺序节点的删除事件
String previousSequencePath = basePath .concat( "/" ) .concat( pathToWatch );
final CountDownLatch latch = new CountDownLatch(1);
final IZkDataListener previousListener = new IZkDataListener() {
public void handleDataDeleted(String dataPath) throws Exception {
latch.countDown(); // 删除后结束latch上的await
}
public void handleDataChange(String dataPath, Object data) throws Exception {
// ignore
}
};
try {
//订阅次小顺序节点的删除事件，若是节点不存在会出现异常
client.subscribeDataChanges(previousSequencePath, previousListener);
if ( millisToWait != null ) {
millisToWait -= (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
if ( millisToWait <= 0 ) {
doDelete = true; // timed out - delete our node
break;
}
latch.await(millisToWait, TimeUnit.MICROSECONDS); // 在latch上await
} else {
latch.await(); // 在latch上await
}
// 结束latch上的等待后，继续while从新来过判断本身是否第一个顺序节点
}
catch ( ZkNoNodeException e ) {
//ignore
} finally {
client.unsubscribeDataChanges(previousSequencePath, previousListener);
}
}
}
}
catch ( Exception e ) {
//发生异常须要删除节点
doDelete = true;
throw e;
} finally {
//若是须要删除节点
if ( doDelete ) {
deleteOurPath(ourPath);
}
}
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;
}
// 获取/locker下的通过排序的子节点列表
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();
}
}
protected void releaseLock(String lockPath) throws Exception{
deleteOurPath(lockPath);
}
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 {
// 在/locker下建立临时的顺序节点
ourPath = createLockNode(client, path);
// 判断本身是否得到了锁，若是没有得到那么等待直到得到锁或者超时
hasTheLock = waitToLock(startMillis, millisToWait, ourPath);
} catch ( ZkNoNodeException e ) { // 捕获这个异常
if ( retryCount++ < MAX_RETRY_COUNT ) { // 重试指定次数
isDone = false;
} else {
throw e;
}
}
}
if ( hasTheLock ) {
return ourPath;
}
return null;
}
}
public class TestDistributedLock {
public static void main(String[] args) {
final ZkClientExt zkClientExt1 = new ZkClientExt("192.168.1.105:2181", 5000, 5000, new BytesPushThroughSerializer());
final SimpleDistributedLockMutex mutex1 = new SimpleDistributedLockMutex(zkClientExt1, "/Mutex");
final ZkClientExt zkClientExt2 = new ZkClientExt("192.168.1.105:2181", 5000, 5000, new BytesPushThroughSerializer());
final SimpleDistributedLockMutex mutex2 = new SimpleDistributedLockMutex(zkClientExt2, "/Mutex");
try {
mutex1.acquire();
System.out.println("Client1 locked");
Thread client2Thd = new Thread(new Runnable() {
public void run() {
try {
mutex2.acquire();
System.out.println("Client2 locked");
mutex2.release();
System.out.println("Client2 released lock");
} catch (Exception e) {
e.printStackTrace();
}
}
});
client2Thd.start();
Thread.sleep(5000);
mutex1.release();
System.out.println("Client1 released lock");
client2Thd.join();
} catch (Exception e) {
e.printStackTrace();
}
}
}
public class ZkClientExt extends ZkClient {
public ZkClientExt(String zkServers, int sessionTimeout, int connectionTimeout, ZkSerializer zkSerializer) {
super(zkServers, sessionTimeout, connectionTimeout, zkSerializer);
}
@Override
public void watchForData(final String path) {
retryUntilConnected(new Callable<Object>() {
public Object call() throws Exception {
Stat stat = new Stat();
_connection.readData(path, stat, true);
return null;
}
});
}
}
zookeeper系列（一）zookeeper必知
zookeeper系列（二）实战master选举
zookeeper系列（三）实战数据发布订阅
zookeeper系列（四）实战负载均衡
zookeeper系列（五）实战分布式锁
zookeeper系列（六）实战分布式队列
zookeeper系列（七）实战分布式命名服务
zookeeper系列（八）zookeeper运维session