ZooKeeper实现分布式锁 ZooKeeper实现分布式锁

【ZooKeeper实现分布式锁】1.什么是分布式锁
一般的锁：一般我们说的锁是但进程多线程的锁，在多线程并发编程中，用于线程之间的数据同步，保护共享资源的访问
分布式锁：分布式锁指的是在分布式环境下，保护跨进程，跨主机，跨网络的共享资源，实现互斥访问，保证一致性
2.分布式锁的架构图

文章图片
20160320214246073.png 3.分布式锁的算法流程

文章图片
20160320221558695.png

package com.jike.lock; import java.util.concurrent.TimeUnit; public interface DistributedLock {/* * 获取锁，如果没有得到就等待 */ public void acquire() throws Exception; /* * 获取锁，直到超时 */ public boolean acquire(long time, TimeUnit unit) throws Exception; /* * 释放锁 */ public void release() throws Exception; }

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 ZkClientExt client; private final Stringpath; //zookeeper中locker节点的路径 private final StringbasePath; private final StringlockName; private static final IntegerMAX_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{booleanhaveTheLock = false; booleandoDelete = false; try { while ( !haveTheLock ) { //获取lock节点下的所有节点 List children = getSortedChildren(); String sequenceNodeName = ourPath.substring(basePath.length()+1); //获取当前节点的在所有节点列表中的位置 intourIndex = children.indexOf(sequenceNodeName); //节点位置小于0,说明没有找到节点 if ( ourIndex<0 ){ throw new ZkNoNodeException("节点没有找到: " + sequenceNodeName); }//节点位置大于0说明还有其他节点在当前的节点前面，就需要等待其他的节点都释放 boolean isGetTheLock = ourIndex == 0; StringpathToWatch = isGetTheLock ? null : children.get(ourIndex - 1); if ( isGetTheLock ){haveTheLock = true; }else{ /** * 获取当前节点的次小的节点，并监听节点的变化 */ StringpreviousSequencePath = basePath .concat( "/" ) .concat( pathToWatch ); final CountDownLatchlatch = new CountDownLatch(1); final IZkDataListener previousListener = new IZkDataListener() {public void handleDataDeleted(String dataPath) throws Exception { latch.countDown(); }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); } else { latch.await(); } } 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; }List getSortedChildren() throws Exception { try{List children = client.getChildren(basePath); Collections.sort ( children, new Comparator() { 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); }/** * 尝试获取锁 * @param time * @param unit * @return * @throws Exception */ protected String attemptLock(long time, TimeUnit unit) throws Exception{final longstartMillis = System.currentTimeMillis(); final LongmillisToWait = (unit != null) ? unit.toMillis(time) : null; StringourPath = null; booleanhasTheLock = false; booleanisDone = false; intretryCount = 0; //网络闪断需要重试一试 while ( !isDone ) { isDone = true; try { 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; }}

import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer; 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(); }} }

原文链接：https://blog.csdn.net/ZuoAnYinXiang/article/details/50938430