zk系列三：zookeeper实战之分布式锁实现( 二 ) _生活百科

提示：代码中注释的代码块可以关注下，原本是直接阻塞式编程，将获取所有子节点并释放锁的操作直接写在getChildren方法的回调里，后来发现当节点被删除时我们还要重新抢锁，那么代码就冗余了，于是结合响应式编程的思想，将这段核心代码放到getChildren方法的回调里，这样代码简洁了并且可以让业务更只关注于getChildren这件事了

2、测试代码编写线程安全问题复现

package com.darling.service.zookeeper.lock;import lombok.SneakyThrows;import lombok.extern.slf4j.Slf4j;import org.junit.Test;/** * @description:开启是个线程给i做递减操作，未加锁的情况下会有线程安全问题 * @author: dll * @date: Created in 2022/11/8 8:32 * @version: * @modified By: */@Slf4jpublic class ZkLockTest02 {private int i = 10;@Testpublic void test() throws InterruptedException {for (int n = 0; n < 10; n++) {new Thread(new Runnable() {@SneakyThrows@Overridepublic void run() {Thread.sleep(100);incre();}}).start();}Thread.sleep(5000);log.info("i = {}",i);}/*** i递减 线程不安全*/public void incre(){//i.incrementAndGet();log.info("当前线程：{},i = {}",Thread.currentThread().getName(),i--);}}

上面代码运行结果如下：

文章插图

使用上面封装的ZkLockHelper实现的分布式锁

package com.darling.service.zookeeper.lock;import lombok.SneakyThrows;import lombok.extern.slf4j.Slf4j;import org.apache.zookeeper.ZooKeeper;import org.junit.After;import org.junit.Before;import org.junit.Test;/** * @description: 使用zk实现的分布式锁解决线程安全问题 * @author: dll * @date: Created in 2022/11/8 8:32 * @version: * @modified By: */@Slf4jpublic class ZkLockTest03 {ZooKeeper zkClient;@Beforepublic void conn (){zkClient= ZkUtil.getZkClient();}@Afterpublic void close (){try {zkClient.close();} catch (InterruptedException e) {e.printStackTrace();}}private int i = 10;@Testpublic void test() throws InterruptedException {for (int n = 0; n < 10; n++) {new Thread(new Runnable() {@SneakyThrows@Overridepublic void run() {Thread.sleep(100);ZkLockHelper zkHelper = new ZkLockHelper();// 这里给zkHelper设置threadName是为了后续调试的时候日志打印，便于观察存在的问题String threadName = Thread.currentThread().getName();zkHelper.setThreadName(threadName);zkHelper.setZkClient(zkClient);// tryLock上锁zkHelper.tryLock();incre();log.info("线程{}正在执行业务代码...",threadName);try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}// 释放锁zkHelper.unLock();}}).start();}while (true) {}}/*** i递减 线程不安全*/public void incre(){//i.incrementAndGet();log.info("☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆当前线程：{},i = {}",Thread.currentThread().getName(),i--);}}