Zookeeper 实现分布式锁及面试真题

1. 分布式锁概述什么叫做分布式锁呢比如说进程1在使用该资源的时候会先去获得锁进程1获得锁以后会对该资源保持独占这样其他进程就无法访问该资源进程1用完该资源以后就将锁释放掉让其他进程来获得锁那么通过这个锁机制我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。2. 原生Zookeeper实现分布式锁案例2.1 分布式锁实现原理实现思路接收到请求后在/locks节点下创建一个临时顺序节点判断自己是不是当前节点下最小的节点是获取到锁不是对前一个节点进行监听获取到锁处理完业务后delete节点释放锁然后下面的节点将收到通知重复第二步判断2.2 代码实现packagecom.atguigu.lock2;importorg.apache.zookeeper.*;importorg.apache.zookeeper.data.Stat;importjava.io.IOException;importjava.util.Collections;importjava.util.List;importjava.util.concurrent.CountDownLatch;publicclassDistributedLock{// zookeeper server列表privateStringconnectStringhadoop102:2181,hadoop103:2181,hadoop104:2181;// 超时时间privateintsessionTimeout2000;privateZooKeeperzk;privateStringrootNodelocks;privateStringsubNodeseq-;// 当前client等待的子节点privateStringwaitPath;// 当前client创建的子节点privateStringcurrentNode;// ZooKeeper连接privateCountDownLatchconnectLatchnewCountDownLatch(1);// ZooKeeper节点等待privateCountDownLatchwaitLatchnewCountDownLatch(1);// 和zk服务建立连接并创建根节点publicDistributedLock()throwsIOException,InterruptedException,KeeperException{zknewZooKeeper(connectString,sessionTimeout,newWatcher(){Overridepublicvoidprocess(WatchedEventevent){// 连接建立时打开latch唤醒wait在该latch上的线程if(event.getState()Event.KeeperState.SyncConnected){connectLatch.countDown();}// 发生了waitPath的删除事件if(event.getType()Event.EventType.NodeDeletedevent.getPath().equals(waitPath)){waitLatch.countDown();}}});// 等待连接建立connectLatch.await();// 获取根节点状态Statstatzk.exists(/rootNode,false);// 如果根节点不存在则创建根节点根节点类型为永久节点if(statnull){System.out.println(根节点不存在);zk.create(/rootNode,newbyte[0],ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);}}// 加锁方法publicvoidzkLock(){try{// 在根节点下创建临时顺序节点返回值为创建的节点路径currentNodezk.create(/rootNode/subNode,null,ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);// wait一小会让结果更清晰一些Thread.sleep(10);// 注意没有必要监听/locks的子节点的变化情况ListStringchildrenNodeszk.getChildren(/rootNode,false);// 列表中只有一个子节点那肯定就是currentNode说明client获得锁if(childrenNodes.size()1){return;}else{// 对根节点下的所有临时顺序节点进行从小到大排序Collections.sort(childrenNodes);// 当前节点名称StringthisNodecurrentNode.substring((/rootNode/).length());// 获取当前节点的位置intindexchildrenNodes.indexOf(thisNode);if(index-1){System.out.println(数据异常);}elseif(index0){// index 0说明thisNode在列表中最小当前client获得锁return;}else{// 获得排名比currentNode前1位的节点this.waitPath/rootNode/childrenNodes.get(index-1);// 在waitPath上注册监听器当waitPath被删除时// zookeeper会回调监听器的process方法zk.getData(waitPath,true,newStat());// 进入等待锁状态waitLatch.await();return;}}}catch(KeeperExceptione){e.printStackTrace();}catch(InterruptedExceptione){e.printStackTrace();}}// 解锁方法publicvoidzkUnlock(){try{zk.delete(this.currentNode,-1);}catch(InterruptedException|KeeperExceptione){e.printStackTrace();}}}2.3 分布式锁测试packagecom.atguigu.lock2;importorg.apache.zookeeper.KeeperException;importjava.io.IOException;publicclassDistributedLockTest{publicstaticvoidmain(String[]args)throwsInterruptedException,IOException,KeeperException{// 创建分布式锁1finalDistributedLocklock1newDistributedLock();// 创建分布式锁2finalDistributedLocklock2newDistributedLock();newThread(newRunnable(){Overridepublicvoidrun(){// 获取锁对象try{lock1.zkLock();System.out.println(线程1获取锁);Thread.sleep(5*1000);lock1.zkUnlock();System.out.println(线程1释放锁);}catch(Exceptione){e.printStackTrace();}}}).start();newThread(newRunnable(){Overridepublicvoidrun(){// 获取锁对象try{lock2.zkLock();System.out.println(线程2获取锁);Thread.sleep(5*1000);lock2.zkUnlock();System.out.println(线程2释放锁);}catch(Exceptione){e.printStackTrace();}}}).start();}}观察控制台变化线程1获取锁 线程1释放锁 线程2获取锁 线程2释放锁3. Curator框架实现分布式锁案例3.1 原生Java API开发存在的问题会话连接是异步的需要自己去处理。比如使用CountDownLatchWatch需要重复注册不然就不能生效开发的复杂性还是比较高的不支持多节点删除和创建。需要自己去递归3.2 Curator简介Curator是一个专门解决分布式锁的框架解决了原生Java API开发分布式遇到的问题。3.3 添加依赖dependencygroupIdorg.apache.curator/groupIdartifactIdcurator-framework/artifactIdversion4.3.0/version/dependencydependencygroupIdorg.apache.curator/groupIdartifactIdcurator-recipes/artifactIdversion4.3.0/version/dependencydependencygroupIdorg.apache.curator/groupIdartifactIdcurator-client/artifactIdversion4.3.0/version/dependency3.4 代码实现packagecom.atguigu.lock2;importorg.apache.curator.RetryPolicy;importorg.apache.curator.framework.CuratorFramework;importorg.apache.curator.framework.CuratorFrameworkFactory;importorg.apache.curator.framework.recipes.locks.InterProcessLock;importorg.apache.curator.framework.recipes.locks.InterProcessMutex;importorg.apache.curator.retry.ExponentialBackoffRetry;publicclassCuratorLockTest{privateStringrootNode/locks;// zookeeper server列表privateStringconnectStringhadoop102:2181,hadoop103:2181,hadoop104:2181;// connection超时时间privateintconnectionTimeout2000;// session超时时间privateintsessionTimeout2000;publicstaticvoidmain(String[]args){newCuratorLockTest().test();}// 测试privatevoidtest(){// 创建分布式锁1finalInterProcessLocklock1newInterProcessMutex(getCuratorFramework(),rootNode);// 创建分布式锁2finalInterProcessLocklock2newInterProcessMutex(getCuratorFramework(),rootNode);newThread(newRunnable(){Overridepublicvoidrun(){// 获取锁对象try{lock1.acquire();System.out.println(线程1 获取锁);// 测试锁重入lock1.acquire();System.out.println(线程1 再次获取锁);Thread.sleep(5*1000);lock1.release();System.out.println(线程1 释放锁);lock1.release();System.out.println(线程1 再次释放锁);}catch(Exceptione){e.printStackTrace();}}}).start();newThread(newRunnable(){Overridepublicvoidrun(){// 获取锁对象try{lock2.acquire();System.out.println(线程2 获取锁);// 测试锁重入lock2.acquire();System.out.println(线程2 再次获取锁);Thread.sleep(5*1000);lock2.release();System.out.println(线程2 释放锁);lock2.release();System.out.println(线程2 再次释放锁);}catch(Exceptione){e.printStackTrace();}}}).start();}// 分布式锁初始化publicCuratorFrameworkgetCuratorFramework(){// 重试策略初试时间3秒重试3次RetryPolicypolicynewExponentialBackoffRetry(3000,3);// 通过工厂创建CuratorCuratorFrameworkclientCuratorFrameworkFactory.builder().connectString(connectString).connectionTimeoutMs(connectionTimeout).sessionTimeoutMs(sessionTimeout).retryPolicy(policy).build();// 开启连接client.start();System.out.println(zookeeper 初始化完成...);returnclient;}}观察控制台变化zookeeper 初始化完成... zookeeper 初始化完成... 线程1 获取锁 线程1 再次获取锁 线程1 释放锁 线程1 再次释放锁 线程2 获取锁 线程2 再次获取锁 线程2 释放锁 线程2 再次释放锁4. 企业面试真题4.1 选举机制半数机制超过半数的投票通过即通过。(1) 第一次启动选举规则投票过半数时服务器id大的胜出(2) 第二次启动选举规则比较规则(EPOCH, ZXID, SID)EPOCH大的直接胜出EPOCH相同事务id大的胜出事务id相同服务器id大的胜出4.2 生产集群安装多少zk合适生产经验服务器数量Zookeeper台数10台服务器3台zk20台服务器5台zk100台服务器11台zk200台服务器11台zk分析服务器台数多好处提高可靠性坏处提高通信延时200台服务器部署11台zk的原因继续增加zk节点数量通信延时的负面影响会超过可靠性提升的收益4.3 常用命令命令功能ls查看节点列表get获取节点数据create创建节点delete删除节点总结主题核心要点分布式锁原理临时顺序节点 Watch监听前一个节点删除事件原生实现手动创建节点、排序、监听、等待、释放Curator框架InterProcessMutex自动处理重试、Watch注册、锁重入选举规则第一次过半数myid大非第一次Epoch→ZXID→SID生产部署10台→3台、20台→5台、100/200台→11台常用命令ls、get、create、delete