介绍

为什么使用锁

锁的出现是为了解决资源争用问题,在单进程环境下的资源争夺可以使用 JDK里的锁实现.

为什么使用分布式锁?

顾名思义,分布式锁是为了分布式环境下的资源争用问题.

Zookeeper是如何实现分布式锁的?

基于Zookeeper的分布式锁都是依赖于zk节点路径唯一的机制来实现的.

什么意思呢?

就是在zk中,在分布式锁的场景下 对于同一个路径,只能有一个客户端能创建成功,其它的都创建失败.(这个不难理解,在平时系统中也没见过有哪2个文件地址完全相同)

下面就说一下zk分布式锁2种实现,没错本篇就是干的不能再干的干货!!!

第一种分布式锁

第一种实现是利用的zk的临时节点, 在争抢锁的时候,所有的客户端都尝试创建一个临时节点(代表锁住的资源),只有一个客户端会创建成功,创建成功的客户端得到锁,其它的客户端则监听(利用zk的watch)该节点的状态改变并且进入阻塞,节点改变后 zk server 会通知剩下的客户端,剩下的客户端停止阻塞并且重新争抢锁.

zk中有持久节点和临时节点,为什么使用临时节点呢?

如果使用的是持久节点,则这个节点在客户端下线后,依旧会一直存在,不会自动删除,导致 其它客户端一直无法争抢到锁 .如果使用的是临时节点的话, 在客户端下线后zk会删除与其相关的临时节点,这样其它客户端就能重新争抢锁 .

    @Override
    public void lock() {
        // 如果获取不到锁，阻塞等待
        if (!tryLock()) {
            // 没获得锁，阻塞自己
            waitForLock();
            // 再次尝试
            lock();
        }
    }

    @Override
    public boolean tryLock() { // 不会阻塞
        // 创建节点
        try {
          // 创建临时节点,zk中的节点(路径)唯一,只有一个会创建成功
      // 为什么使用临时节点: 客户端掉线后会自动删除节点(释放锁)
            client.createEphemeral(lockPath);
        } catch (ZkNodeExistsException e) {
            return false;
        }
        return true;
    }

    /**
   * 争抢不到锁的话,等待锁的释放
     */
    private void waitForLock() {
        CountDownLatch cdl = new CountDownLatch(1);
        IZkDataListener listener = new IZkDataListener() {
            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                System.out.println("收到节点被删除的消息,停止等待,重新争夺锁");
                cdl.countDown();
            }
            @Override
            public void handleDataChange(String dataPath, Object data)
                    throws Exception {
            }
        };

        // 监听
        client.subscribeDataChanges(lockPath, listener);
        // 判断锁节点是否存在，存在的话表明有别人
        if (this.client.exists(lockPath)) {
            try {
                // 等待接收到消息后,继续往下执行
                cdl.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 取消监听消息
        client.unsubscribeDataChanges(lockPath, listener);
    }

实现简单,但是会有 羊群效应 ,节点的删除都会通知所有的客户端,并且所有的客户端会 取消监听 + 重新一起争夺锁 + 争夺失败 + 再次开启监听 ,如此循环,资源耗费多,并且这种耗费是可以避免的,那么如何避免呢?就是下面第二种的 改进版分布式锁 .

第二种分布式锁

这一种分布式锁的实现是利用zk的临时顺序节点,每一个客户端在争夺锁的时候都由zk分配一个顺序号(sequence),客户端则按照这个顺序去获取锁.

lock跟前面的一样,不过lockPath(锁住的资源)是一个持久节点,客户端在该持久节点下面创建临时顺序节点,获取到顺序号后,根据自己是否是最小的顺序号来获取锁,顺序号最小则获取锁,序号不为最小则监听(watch)前一个顺序号,当前一个顺序号被删除的时候表明锁被释放了,则会通知下一个客户端.

下面贴出跟第一种实现不同的代码

    /**
     * 尝试加锁
     *
     * @return
     */
    @Override
    public boolean tryLock() {
        // 创建临时顺序节点
        if (this.currentPath == null) {
            // 在lockPath节点下面创建临时顺序节点
            currentPath = this.client.createEphemeralSequential(LockPath + "/", "aaa");
        }
        // 获得所有的子节点
        List<String> children = this.client.getChildren(LockPath);

        // 排序list
        Collections.sort(children);

        // 判断当前节点是否是最小的,如果是最小的节点,则表明此这个client可以获取锁
        if (currentPath.equals(LockPath + "/" + children.get(0))) {
            return true;
        } else {
            // 如果不是当前最小的sequence,取到前一个临时节点
            // 1.单独获取临时节点的顺序号
            // 2.查找这个顺序号在children中的下标
            // 3.存储前一个节点的完整路径
            int curIndex = children.indexOf(currentPath.substring(LockPath.length() + 1));
            beforePath = LockPath + "/" + children.get(curIndex - 1);
        }
        return false;
    }

    private void waitForLock() {
        CountDownLatch cdl = new CountDownLatch(1);
        // 注册watcher
        IZkDataListener listener = new IZkDataListener() {
            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                System.out.println("监听到前一个节点被删除了");
                cdl.countDown();
            }
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
            }
        };

        // 监听前一个临时节点
        client.subscribeDataChanges(this.beforePath, listener);

        // 前一个节点还存在,则阻塞自己
        if (this.client.exists(this.beforePath)) {
            try {
                // 直至前一个节点释放锁,才会继续往下执行
                cdl.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        // 醒来后,表明前一个临时节点已经被删除,此时客户端可以获取锁 && 取消watcher监听
        client.unsubscribeDataChanges(this.beforePath, listener);
    }

实现比第一种复杂一点,但是更加的合理,少做了很多不必要的操作,只唤醒了后面一个客户端.