Zookeeper--08---zk实现分布式锁、案例

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

zk实现分布式锁

1.zk中锁的种类：

• 读锁：⼤家都可以读，要想上读锁的前提：之前的锁没有写锁
• 写锁：只有得到写锁的才能写。要想上写锁的前提是，之前没有任何锁。

2.zk如何上读锁

1.创建⼀个临时序号节点，节点的数据是read，表示是读锁

2.获取当前zk中序号⽐⾃⼰⼩的所有节点

3.判断最⼩节点是否是读锁：

• 如果不是读锁的话，则上锁失败，为最⼩节点设置监听。阻塞等待，zk的watch机制
  会当最⼩节点发⽣变化时通知当前节点，于是再执⾏第⼆步的流程
• 如果是读锁的话，则上锁成功
  

3.zk如何上写锁

1.创建⼀个临时序号节点，节点的数据是write，表示是 写锁

2.获取zk中所有的⼦节点

3.判断⾃⼰是否是最⼩的节点：

• 如果是，则上写锁成功
• 如果不是，说明前⾯还有锁，则上锁失败，监听最⼩的节点，如果最⼩节点有变化， 则回到第⼆步。
  

4.⽺群效应

如果⽤上述的上锁⽅式，只要有节点发⽣变化，就会触发其他节点的监听事件，这样的话对
zk的压⼒⾮常⼤，——⽺群效应。

可以调整成链式监听。解决这个问题。

5.curator实现读写锁

importorg.apache.curator.framework.CuratorFramework;importorg.apache.curator.framework.recipes.locks.InterProcessLock;importorg.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;importorg.junit.jupiter.api.Test;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;@SpringBootTestpublicclassTestReadWriteLock{@AutowiredprivateCuratorFrameworkclient;@TestvoidtestGetReadLock()throwsException{// 读写锁InterProcessReadWriteLockinterProcessReadWriteLock=newInterProcessReadWriteLock(client,"/lock1");// 获取读锁对象InterProcessLockinterProcessLock=interProcessReadWriteLock.readLock();System.out.println("等待获取读锁对象!");// 获取锁interProcessLock.acquire();for(inti=1;i<=100;i++){Thread.sleep(3000);System.out.println(i);}// 释放锁interProcessLock.release();System.out.println("等待释放锁!");}@TestvoidtestGetWriteLock()throwsException{// 读写锁InterProcessReadWriteLockinterProcessReadWriteLock=newInterProcessReadWriteLock(client,"/lock1");// 获取写锁对象InterProcessLockinterProcessLock=interProcessReadWriteLock.writeLock();System.out.println("等待获取写锁对象!");// 获取锁interProcessLock.acquire();for(inti=1;i<=100;i++){Thread.sleep(3000);System.out.println(i);}// 释放锁interProcessLock.release();System.out.println("等待释放锁!");}}

分布式锁案例

案例分析

• 比如说 "进程 1"在使用该资源的时候，会先去获得锁在使用该资源的时候，保持独占，这样其他进程就无法访问该资源，
• "进程1"用完该资源以后就将锁释放掉，让其他进程来获得锁，
• 那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。

1. 接收到请求后，在/locks节点下创建一个临时顺序节点
2. 判断自己是不是当前节点下最小的节点：是，获取到锁；不是，对前一个节点进行监听
3. 获取到锁，处理完业务后，delete节点释放锁，然后下面的节点将收到通知，重复第二步判断

依赖：

<dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.5.7</version></dependency>

分布式锁原理–序号节点

持久序号节点

临时序号节点

分布式锁实现

• 获取连接
• 对zk加锁
• 对zk解锁

importorg.apache.zookeeper.*;importorg.apache.zookeeper.data.Stat;importjava.io.IOException;importjava.util.Collections;importjava.util.List;importjava.util.concurrent.CountDownLatch;publicclassDistributedLock{privatefinalStringconnectString="hadoop102:2181,hadoop103:2181,hadoop104:2181";privatefinalintsessionTimeout=2000;privatefinalZooKeeperzk;privateCountDownLatchconnectLatch=newCountDownLatch(1);privateCountDownLatchwaitLatch=newCountDownLatch(1);privateStringwaitPath;privateStringcurrentMode;publicDistributedLock()throwsIOException,InterruptedException,KeeperException{// 获取连接zk=newZooKeeper(connectString,sessionTimeout,newWatcher(){@Overridepublicvoidprocess(WatchedEventwatchedEvent){// connectLatch 如果连接上zk 可以释放if(watchedEvent.getState()==Event.KeeperState.SyncConnected){connectLatch.countDown();}// waitLatch 需要释放if(watchedEvent.getType()==Event.EventType.NodeDeleted&&watchedEvent.getPath().equals(waitPath)){waitLatch.countDown();}}});// 等待zk正常连接后，往下走程序connectLatch.await();// 判断根节点/locks是否存在Statstat=zk.exists("/locks",false);if(stat==null){// 创建一下根节点zk.create("/locks","locks".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);}}// 对zk加锁publicvoidzklock(){// 创建对应的临时带序号节点try{currentMode=zk.create("/locks/"+"seq-",null,ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);// wait一小会, 让结果更清晰一些Thread.sleep(10);// 判断创建的节点是否是最小的序号节点，如果是获取到锁；如果不是，监听他序号前一个节点List<String>children=zk.getChildren("/locks",false);// 如果children 只有一个值，那就直接获取锁； 如果有多个节点，需要判断，谁最小if(children.size()==1){return;}else{Collections.sort(children);// 获取节点名称 seq-00000000StringthisNode=currentMode.substring("/locks/".length());// 通过seq-00000000获取该节点在children集合的位置intindex=children.indexOf(thisNode);// 判断if(index==-1){System.out.println("数据异常");}elseif(index==0){// 就一个节点，可以获取锁了return;}else{// 需要监听 他前一个节点变化waitPath="/locks/"+children.get(index-1);zk.getData(waitPath,true,newStat());// 等待监听waitLatch.await();return;}}}catch(KeeperExceptione){e.printStackTrace();}catch(InterruptedExceptione){e.printStackTrace();}}// 解锁publicvoidunZkLock(){// 删除节点try{zk.delete(this.currentMode,-1);}catch(InterruptedExceptione){e.printStackTrace();}catch(KeeperExceptione){e.printStackTrace();}}}

测试

importorg.apache.zookeeper.KeeperException;importjava.io.IOException;publicclassDistributedLockTest{publicstaticvoidmain(String[]args)throwsInterruptedException,IOException,KeeperException{finalDistributedLocklock1=newDistributedLock();finalDistributedLocklock2=newDistributedLock();newThread(newRunnable(){@Overridepublicvoidrun(){try{lock1.zklock();System.out.println("线程1 启动，获取到锁");Thread.sleep(5*1000);lock1.unZkLock();System.out.println("线程1 释放锁");}catch(InterruptedExceptione){e.printStackTrace();}}}).start();newThread(newRunnable(){@Overridepublicvoidrun(){try{lock2.zklock();System.out.println("线程2 启动，获取到锁");Thread.sleep(5*1000);lock2.unZkLock();System.out.println("线程2 释放锁");}catch(InterruptedExceptione){e.printStackTrace();}}}).start();}}

对比单体模式下—ReentrantLock

Curator框架实现分布式锁案例

依赖

<dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>4.3.0</version></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>4.3.0</version></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-client</artifactId><version>4.3.0</version></dependency>

获取客户端连接

privatestaticCuratorFrameworkgetCuratorFramework(){ExponentialBackoffRetrypolicy=newExponentialBackoffRetry(3000,3);CuratorFrameworkclient=CuratorFrameworkFactory.builder().connectString("hadoop102:2181,hadoop103:2181,hadoop104:2181").connectionTimeoutMs(2000).sessionTimeoutMs(2000).retryPolicy(policy).build();// 启动客户端client.start();System.out.println("zookeeper 启动成功");returnclient;}

测试案例

importorg.apache.curator.framework.CuratorFramework;importorg.apache.curator.framework.CuratorFrameworkFactory;importorg.apache.curator.framework.recipes.locks.InterProcessMutex;importorg.apache.curator.retry.ExponentialBackoffRetry;publicclassCuratorLockTest{publicstaticvoidmain(String[]args){// 创建分布式锁1InterProcessMutexlock1=newInterProcessMutex(getCuratorFramework(),"/locks");// 创建分布式锁2InterProcessMutexlock2=newInterProcessMutex(getCuratorFramework(),"/locks");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();}privatestaticCuratorFrameworkgetCuratorFramework(){ExponentialBackoffRetrypolicy=newExponentialBackoffRetry(3000,3);CuratorFrameworkclient=CuratorFrameworkFactory.builder().connectString("hadoop102:2181,hadoop103:2181,hadoop104:2181").connectionTimeoutMs(2000).sessionTimeoutMs(2000).retryPolicy(policy).build();// 启动客户端client.start();System.out.println("zookeeper 启动成功");returnclient;}}

Redis分布式锁的实现

Redis–12–Redis分布式锁的实现