Zookeeper学习(九):ZooKeeper 实现分布式锁
2016-03-21 11:25
573 查看
1.什么是分布式锁
一般的锁:一般我们说的锁是但进程多线程的锁,在多线程并发编程中,用于线程之间的数据同步,保护共享资源的访问
分布式锁:分布式锁指的是在分布式环境下,保护跨进程,跨主机,跨网络的共享资源,实现互斥访问,保证一致性
2.分布式锁的架构图
3.分布式锁的算法流程
一般的锁:一般我们说的锁是但进程多线程的锁,在多线程并发编程中,用于线程之间的数据同步,保护共享资源的访问
分布式锁:分布式锁指的是在分布式环境下,保护跨进程,跨主机,跨网络的共享资源,实现互斥访问,保证一致性
2.分布式锁的架构图
3.分布式锁的算法流程
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; }
package com.jike.lock; 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 String path; //zookeeper中locker节点的路径 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 ) { //获取lock节点下的所有节点 List<String> children = getSortedChildren(); String sequenceNodeName = ourPath.substring(basePath.length()+1); //获取当前节点的在所有节点列表中的位置 int ourIndex = children.indexOf(sequenceNodeName); //节点位置小于0,说明没有找到节点 if ( ourIndex<0 ){ throw new ZkNoNodeException("节点没有找到: " + sequenceNodeName); } //节点位置大于0说明还有其他节点在当前的节点前面,就需要等待其他的节点都释放 boolean isGetTheLock = ourIndex == 0; 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(); } 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<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); } /** * 尝试获取锁 * @param time * @param unit * @return * @throws Exception */ 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 { 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; } }
package com.jike.lock; 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(); } } }
相关文章推荐
- ionic 里使用 iframe 可能遇到的问题
- 百度地图-定位
- 有符号数和无符号数
- 为什么要使用SLF4J而不是Log4J
- ****如何优雅的用Axure装逼?高保真原型心得分享
- iOS_SN_CoreData数据迁移
- iOS/Android/Web Url Encode空格處理 原文連結:http://read01.com/3gDO.html
- Eclipse远程调试HDP源代码
- GDC 2016 神秘海域4中使用Substance制作Texture
- 美好的一天(Weather Live)
- 华为设计总监:年轻设计师如何进步?
- PHP计算两个经纬度之间的距离
- PHP socket实现 writeInt 和 writeLong 和 writeUTF 以及和Java端对接
- MySQL 5.6 for Windows 配置安装
- 330. Patching Array
- Android Dalvik虚拟机初识
- iOS微信支付
- 十天学会web标准(DIV+CSS)系列(四)纵向导航菜单及二级弹出菜单
- 本地如何无密码登录linux服务器
- flume+Kafka+Storm实时框架