Zookeeper节点锁实现

╰+攻爆jí腚メ 2022-05-16 04:47 250阅读 0赞

# Zookeeper节点锁实现理论分析 #

1，该功能基于zookeeper有序节点的小DEMO实现

2，新增节点后尝试对该节点加锁

3，加锁失败后，进入线程等待， 并对上一节点进行监听

4，上一节点状态变更后唤醒线程获取锁执行任务

5，执行完成后，释放锁

# Zookeeper节点锁编码变现 #

1，jar包引入

<dependency>
          <groupId>org.apache.curator</groupId>
          <artifactId>curator-framework</artifactId>
          <version>4.0.0</version>
        </dependency>

2，定义类实现锁及事件接口，重写方法后定义必要的成员变量

\* 基于zookeeper根节点下的/locks节点来实现锁机制

public class ZookeeperLockDemo implements Lock, Watcher {
    
        // ZK链接
        private ZooKeeper zooKeeper = null;
    
        // 根节点
        private final String ROUTE_LOCK = "/locks";
    
        // 等待锁-->即前一个锁
        private String WAIT_LOCK;
    
        // 当前锁节点
        private String CURR_LOCK;
    
        // 线程沉睡唤醒
        private CountDownLatch countDownLatch;

3，通过构造方法初始化链接及根节点，基于zookeeper集群实现

public ZookeeperLockDemo() throws IOException, KeeperException, InterruptedException {
            String host = "192.168.91.128:2181,192.168.91.129:2181,192.168.91.130:2181";
            // this引用当前process方法实现监听
            zooKeeper = new ZooKeeper(host, 4000, this);
            // 判断当前节点是否存在
            Stat stat = zooKeeper.exists(ROUTE_LOCK, false);
            // 创建持久化节点
            if (null == stat) {
                zooKeeper.create(ROUTE_LOCK, "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        }

4，对当前链接节点加锁

public void lock() {
            if (tryLock()) {
                System.out.println(Thread.currentThread().getName() + "-->" + CURR_LOCK + "获得锁成功");
            }
            waitForLock(WAIT_LOCK);
        }

5，加锁判断

public boolean tryLock() {
            try {
                // 创建临时有序节点
                CURR_LOCK = zooKeeper.create(ROUTE_LOCK + "/", "0".getBytes(),
                        ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
                System.out.println(Thread.currentThread().getName() + "-->" + CURR_LOCK + "尝试竞争锁");
                // 获取根节点下的所有子节点
                List<String> lstChildren = zooKeeper.getChildren(ROUTE_LOCK, false);
                // 节点排序
                SortedSet<String> set = new TreeSet<String>();
                for (String child : lstChildren) {
                    set.add(ROUTE_LOCK + "/" + child);
                }
                // 获取有序集合中第一个元素
                String firstElement = set.first();
                // 获取当前节点的所有队列靠前节点
                SortedSet<String> headSet = ((TreeSet<String>) set).headSet(CURR_LOCK);
                // 竞争锁
                if (CURR_LOCK.equals(firstElement)) {
                    return true;
                }
                // 最后一个节点为当前节点的等待节点
                if (!headSet.isEmpty()) {
                    WAIT_LOCK = headSet.last();
                }
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return false;
        }

6，tryLock()失败后，对上一节点锁进行监听，并进行等待

private boolean waitForLock(String waitLock) {
            try {
                // 监听当前节点上一个节点
                Stat stat = zooKeeper.exists(waitLock, true);
                if (stat != null) {
                    System.out.println(Thread.currentThread().getName() + "等待锁 :" + ROUTE_LOCK + "/" + waitLock + "释放");
                    countDownLatch = new CountDownLatch(1);
                    countDownLatch.await();
                    System.out.println(Thread.currentThread().getName() + "获得锁成功");
                }
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return true;
        }

7，节点任务执行完成后，对锁进行释放（此处为删除节点）

public void unlock() {
            System.out.println(Thread.currentThread().getName() + "释放锁 : " + CURR_LOCK);
            try {
                zooKeeper.delete(CURR_LOCK, -1);
                CURR_LOCK = null;
                zooKeeper.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (KeeperException e) {
                e.printStackTrace();
            }
        }

8，exist()事件监听，监听到节点不存在时，CountDownLatch进行countDown()操作

public void process(WatchedEvent watchedEvent) {
            if (null != countDownLatch) {
                countDownLatch.countDown();
            }
        }

9，CountDownLatch参数为0后，自动唤醒线程，为当前节点加锁

private boolean waitForLock(String waitLock) {
            try {
                // 监听当前节点上一个节点
                Stat stat = zooKeeper.exists(waitLock, true);
                if (stat != null) {
                    System.out.println(Thread.currentThread().getName() + "等待锁 :" + ROUTE_LOCK + "/" + waitLock + "释放");
                    // countDownLatch参数为1，线程等待
                    countDownLatch = new CountDownLatch(1);
                    countDownLatch.await();
                    // countDownLatch参数为0后，自动唤醒线程执行
                    System.out.println(Thread.currentThread().getName() + "获得锁成功");
                }
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return true;
        }

10，执行完成后，进行锁释放，下一节点对当前节点进行监听，继续获取锁

11，客户端调用

public class LockDemoTest {
        public static void main(String[] args) throws Exception {
            final CountDownLatch countDownLatch = new CountDownLatch(10);
            for (int i = 0; i < 10; i++) {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            countDownLatch.await();
                            ZookeeperLockDemo zookeeperLockDemo = new ZookeeperLockDemo();
                            zookeeperLockDemo.lock();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } catch (IOException e) {
                            e.printStackTrace();
                        } catch (KeeperException e) {
                            e.printStackTrace();
                        }
                    }
                }, "Thread-->" + i).start();
                countDownLatch.countDown();
            }
            System.in.read();
        }
    }

12， 执行效果

![70][]

\* 手动删除节点9，制造锁释放场景

![70 1][]

\* 观察控制台会发现0节点获取锁

![70 2][]

\* 可以通过获取锁后直接释放来观察结果

private boolean waitForLock(String waitLock) {
            try {
                // 监听当前节点上一个节点
                Stat stat = zooKeeper.exists(waitLock, true);
                if (stat != null) {
                    System.out.println(Thread.currentThread().getName() + "等待锁 :" + ROUTE_LOCK + "/" + waitLock + "释放");
                    // countDownLatch参数为1，线程等待
                    countDownLatch = new CountDownLatch(1);
                    countDownLatch.await();
                    // countDownLatch参数为0后，自动唤醒线程执行
                    System.out.println(Thread.currentThread().getName() + "获得锁成功");
                    Thread.sleep(2000);
                    System.out.println(System.currentTimeMillis());
                    unlock();
                }
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return true;
        }

![70 3][]

\* 最后，基于原生的zookeeper链接，多线程连接判断存在漏洞，第一次执行会报错

[70]: /images/20220516/c4450b11e4b24ef6abb7c9b4851a22a6.png
[70 1]: /images/20220516/3262b4d8d70f45a0a9a1305d7068fcb1.png
[70 2]: /images/20220516/0d0c77db162c405ba26a02d6cad1df89.png
[70 3]: /images/20220516/eb7ca554f6d44c868f64db10095e1448.png