今天儿给大伙说说Zookeeper分布式锁

💛分布式解决方案源码,请帮我点个star哦!
💛原文地址为https://www.cnblogs.com/haixiang/p/13112710.html,转载请注明出处!

zookeeper客户端选型#

  • 原生zookeeper客户端,有watcher一次性、无超时重连机制等一系列问题
  • ZkClient,解决了原生客户端一些问题,一些存量老系统中还在使用
  • curator,提供了各种应用场景(封装了分布式锁,计数器等),新项目首选

分布式锁使用场景#

在单体项目中jvm中的锁即可完成需要,但是微服务、分布式环境下,同一个服务可能部署在多台服务器上,多个jvm之间无法通过常用的jvm锁来完成同步操作,需要借用分布式锁来完成上锁、释放锁。例如在订单服务中,我们需要根据日期来生成订单号流水,就有可能产生相同的时间日期,从而出现重复订单号。(jdk8使用LocalDateTime线程安全,不会存在这样的问题)

zookeeper分布式锁实现原理#

  • zookeeper中规定,在同一时刻,不能有多个客户端创建同一个节点,我们可以利用这个特性实现分布式锁。zookeeper临时节点只在session生命周期存在,session一结束会自动销毁。
  • watcher机制,在代表锁资源的节点被删除,即可以触发watcher解除阻塞重新去获取锁,这也是zookeeper分布式锁较其他分布式锁方案的一大优势。

基于临时节点方案#

第一种方案实现较为简单,逻辑就是谁创建成功该节点,谁就持有锁,创建失败的自己进行阻塞,A线程先持有锁,B线程获取失败就会阻塞,同时对/lockPath设置监听,A线程执行完操作后删除节点,触发监听器,B线程此时解除阻塞,重新去获取锁。

我们模仿原生jdk的lock接口设计,采用模板方法设计模式来编写分布式锁,这样的好处是扩展性强,我们可以快速切换到redis分布式锁、数据库分布式锁等实现方式。

创建Lock接口

Copy<strong>public</strong> <strong>interface</strong> <strong>Lock</strong> {
    /**
     * 获取锁
     */
    <strong>void</strong> <strong>getLock</strong>() <strong>throws</strong> Exception;

    /**
     * 释放锁
     */
    <strong>void</strong> <strong>unlock</strong>() <strong>throws</strong> Exception;
}

AbstractTemplateLock抽象类

Copy<code><strong>public</strong> <strong>abstract</strong> <strong>class</strong> <strong>AbstractTemplateLock</strong> <strong>implements</strong> <strong>Lock</strong> {
    @Override
    <strong>public</strong> <strong>void</strong> <strong>getLock</strong>() {
        <strong>if</strong> (tryLock()) {
            System.out.println(Thread.currentThread().getName() + "获取锁成功");
        } <strong>else</strong> {
            //等待
            waitLock();//事件监听 如果节点被删除则可以重新获取
            //重新获取
            getLock();
        }
    }
    <strong>protected</strong> <strong>abstract</strong> <strong>void</strong> <strong>waitLock</strong>();
    <strong>protected</strong> <strong>abstract</strong> <strong>boolean</strong> <strong>tryLock</strong>();
    <strong>protected</strong> <strong>abstract</strong> <strong>void</strong> <strong>releaseLock</strong>();
    @Override
    <strong>public</strong> <strong>void</strong> <strong>unlock</strong>() {
        releaseLock();
    }
}
</code>

zookeeper分布式锁逻辑

Copy<code>@Slf4j
<strong>public</strong> <strong>class</strong> <strong>ZkTemplateLock</strong> <strong>extends</strong> <strong>AbstractTemplateLock</strong> {
    <strong>private</strong> <strong>static</strong> <strong>final</strong> String zkServers = "127.0.0.1:2181";
    <strong>private</strong> <strong>static</strong> <strong>final</strong> <strong>int</strong> sessionTimeout = 8000;
    <strong>private</strong> <strong>static</strong> <strong>final</strong> <strong>int</strong> connectionTimeout = 5000;

    <strong>private</strong> <strong>static</strong> <strong>final</strong> String lockPath = "/lockPath";


    <strong>private</strong> ZkClient client;

    <strong>public</strong> <strong>ZkTemplateLock</strong>() {
        client = <strong>new</strong> ZkClient(zkServers, sessionTimeout, connectionTimeout);
        log.info("zk client 连接成功:{}",zkServers);
    }

    @Override
    <strong>protected</strong> <strong>void</strong> <strong>waitLock</strong>() {
        CountDownLatch latch = <strong>new</strong> CountDownLatch(1);

        IZkDataListener listener = <strong>new</strong> IZkDataListener() {
            @Override
            <strong>public</strong> <strong>void</strong> <strong>handleDataDeleted</strong>(String dataPath) <strong>throws</strong> Exception {
                System.out.println("监听到节点被删除");
                latch.countDown();
            }
            @Override
            <strong>public</strong> <strong>void</strong> <strong>handleDataChange</strong>(String dataPath, Object data) <strong>throws</strong> Exception {}
        };
        //完成 watcher 注册
        client.subscribeDataChanges(lockPath, listener);

        //阻塞自己
        <strong>if</strong> (client.exists(lockPath)) {
            <strong>try</strong> {
                latch.await();
            } <strong>catch</strong> (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //取消watcher注册
        client.unsubscribeDataChanges(lockPath, listener);
    }

    @Override
    <strong>protected</strong> <strong>boolean</strong> <strong>tryLock</strong>() {
        <strong>try</strong> {
            client.createEphemeral(lockPath);
            System.out.println(Thread.currentThread().getName()+"获取到锁");
        } <strong>catch</strong> (Exception e) {
            log.error("创建失败");
            <strong>return</strong> <strong>false</strong>;
        }
        <strong>return</strong> <strong>true</strong>;
    }

    @Override
    <strong>public</strong> <strong>void</strong> <strong>releaseLock</strong>() {
       client.delete(<strong>this</strong>.lockPath);
    }
}

</code>

缺点

每次去竞争锁,都只会有一个线程拿到锁,当线程数庞大时会发生“惊群”现象,zookeeper节点可能会运行缓慢甚至宕机。这是因为其他线程没获取到锁时都会监听/lockPath节点,当A线程释放完毕,海量的线程都同时停止阻塞,去争抢锁,这种操作十分耗费资源,且性能大打折扣。

基于临时顺序节点方案#

临时顺序节点与临时节点不同的是产生的节点是有序的,我们可以利用这一特点,只让当前线程监听上一序号的线程,每次获取锁的时候判断自己的序号是否为最小,最小即获取到锁,执行完毕就删除当前节点继续判断谁为最小序号的节点。

临时顺序节点操作源码

Copy<code>@Slf4j
<strong>public</strong> <strong>class</strong> <strong>ZkSequenTemplateLock</strong> <strong>extends</strong> <strong>AbstractTemplateLock</strong> {
    <strong>private</strong> <strong>static</strong> <strong>final</strong> String zkServers = "127.0.0.1:2181";
    <strong>private</strong> <strong>static</strong> <strong>final</strong> <strong>int</strong> sessionTimeout = 8000;
    <strong>private</strong> <strong>static</strong> <strong>final</strong> <strong>int</strong> connectionTimeout = 5000;
    <strong>private</strong> <strong>static</strong> <strong>final</strong> String lockPath = "/lockPath";
    <strong>private</strong> String beforePath;
    <strong>private</strong> String currentPath;
    <strong>private</strong> ZkClient client;

    <strong>public</strong> <strong>ZkSequenTemplateLock</strong>() {
        client = <strong>new</strong> ZkClient(zkServers);
        <strong>if</strong> (!client.exists(lockPath)) {
            client.createPersistent(lockPath);

        }
        log.info("zk client 连接成功:{}",zkServers);

    }

    @Override
    <strong>protected</strong> <strong>void</strong> <strong>waitLock</strong>() {
        CountDownLatch latch = <strong>new</strong> CountDownLatch(1);
        IZkDataListener listener = <strong>new</strong> IZkDataListener() {
            @Override
            <strong>public</strong> <strong>void</strong> <strong>handleDataDeleted</strong>(String dataPath) <strong>throws</strong> Exception {
                System.out.println("监听到节点被删除");
                latch.countDown();
            }
            @Override
            <strong>public</strong> <strong>void</strong> <strong>handleDataChange</strong>(String dataPath, Object data) <strong>throws</strong> Exception {}
        };
        //给排在前面的节点增加数据删除的watcher,本质是启动另一个线程去监听上一个节点
        client.subscribeDataChanges(beforePath, listener);

        //阻塞自己
        <strong>if</strong> (client.exists(beforePath)) {
            <strong>try</strong> {
                System.out.println("阻塞"+currentPath);
                latch.await();
            } <strong>catch</strong> (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //取消watcher注册
        client.unsubscribeDataChanges(beforePath, listener);
    }

    @Override
    <strong>protected</strong> <strong>boolean</strong> <strong>tryLock</strong>() {
        <strong>if</strong> (currentPath == <strong>null</strong>) {
            //创建一个临时顺序节点
            currentPath = client.createEphemeralSequential(lockPath + "/", "lock-data");
            System.out.println("current:" + currentPath);
        }

        //获得所有的子节点并排序。临时节点名称为自增长的字符串
        List<String> childrens = client.getChildren(lockPath);
        //排序list,按自然顺序排序
        Collections.sort(childrens);
        <strong>if</strong> (currentPath.equals(lockPath + "/" + childrens.get(0))) {
            <strong>return</strong> <strong>true</strong>;
        } <strong>else</strong> {
            //如果当前节点不是排第一,则获取前面一个节点信息,赋值给beforePath
            <strong>int</strong> curIndex = childrens.indexOf(currentPath.substring(lockPath.length() + 1));
            beforePath = lockPath + "/" + childrens.get(curIndex - 1);
        }
        System.out.println("beforePath"+beforePath);
        <strong>return</strong> <strong>false</strong>;
    }

    @Override
    <strong>public</strong> <strong>void</strong> <strong>releaseLock</strong>() {
        System.out.println("delete:" + currentPath);
        client.delete(currentPath);
    }
}

</code>

Curator分布式锁工具#

curator提供了以下种类的锁:

  • 共享可重入锁(Shared Reentrant Lock):全局同步锁,同一时间不会有两个客户端持有一个锁
  • 共享锁:与共享可重入锁类似,但是不可重入(有时候会因为这个原因造成死锁)
  • 共享可重入读写锁
  • 共享信号量
  • Multi Shared Lock:管理多种锁的容器实体

我们采用第一种Shared Reentrant Lock中的InterProcessMutex来完成上锁、释放锁的的操作

Copy<code><strong>public</strong> <strong>class</strong> <strong>ZkLockWithCuratorTemplate</strong> <strong>implements</strong> <strong>Lock</strong> {
    // zk host地址
    <strong>private</strong> String host = "localhost";

    // zk自增存储node
    <strong>private</strong> String lockPath = "/curatorLock";

    // 重试休眠时间
    <strong>private</strong> <strong>static</strong> <strong>final</strong> <strong>int</strong> SLEEP_TIME_MS = 1000;
    // 最大重试1000次
    <strong>private</strong> <strong>static</strong> <strong>final</strong> <strong>int</strong> MAX_RETRIES = 1000;
    //会话超时时间
    <strong>private</strong> <strong>static</strong> <strong>final</strong> <strong>int</strong> SESSION_TIMEOUT = 30 * 1000;
    //连接超时时间
    <strong>private</strong> <strong>static</strong> <strong>final</strong> <strong>int</strong> CONNECTION_TIMEOUT = 3 * 1000;
		//curator核心操作类
    <strong>private</strong> CuratorFramework curatorFramework;

    InterProcessMutex lock;

   <strong>public</strong> <strong>ZkLockWithCuratorTemplate</strong>() {
       curatorFramework = CuratorFrameworkFactory.builder()
               .connectString(host)
               .connectionTimeoutMs(CONNECTION_TIMEOUT)
               .sessionTimeoutMs(SESSION_TIMEOUT)
               .retryPolicy(<strong>new</strong> ExponentialBackoffRetry(SLEEP_TIME_MS, MAX_RETRIES))
               .build();
       curatorFramework.start();
       lock = <strong>new</strong> InterProcessMutex (curatorFramework, lockPath);
    }

    @Override
    <strong>public</strong> <strong>void</strong> <strong>getLock</strong>() <strong>throws</strong> Exception {
        //5s后超时释放锁
         lock.acquire(5, TimeUnit.SECONDS);
    }

    @Override
    <strong>public</strong> <strong>void</strong> <strong>unlock</strong>() <strong>throws</strong> Exception {
        lock.release();
    }
}

</code>

源码以及测试类地址

Copyhttps://github.com/Motianshi/distribute-tool
© 版权声明
THE END
喜欢就支持以下吧
点赞47 分享
评论区 抢沙发
图片正在生成中,请稍后...