ZooKeeper学习总结(八)Zookeeper分布式锁

た入场券 2024-02-19 15:09 18阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzU0OTU3OA_size_16_color_FFFFFF_t_70][]

分布式系统中，多台机器并行去实现同一个功能 。jdk自带的锁已经无法满足这个情况了。

**分布式锁的特性：**进程之间的锁机制已经不再是线程之间的锁机制了。

1.可重入特性，客户端在获取到锁后，在调用方法内，有层递调用，并且所需的锁是同一吧，可无需在重新获取锁。

2.网络中断或者宕机，自动删除锁，防置死锁现象。

3.分布式中，每次只有一个方法或者变量同一时间被一个客户端的线程操作。

4.具备阻塞锁（没有获取到锁，则继续等待获取锁）的特性，与非阻塞锁（没有获取到锁，则直接返回获取锁失败）的特性。

zookeeper=文件系统+通知机制

![2018122516540839][]

ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，它内部是一个分层的文件系统目录树结构，规定同一个目录下只能存在唯一文件名。同时他还具备有监听器watch(1）节点创建；2）节点删除；3）节点数据修改；4）子节点变更)，并且在zookeeper中存在很多节点，其中临时性节点（会话结束或者会话超时后，zookeeper会自动删除该节点）与有序性节点就很满足分布式锁的锁失效机制，防置死锁的发生。

根据zookeeper实现分布式锁：

1.在zookeeper中先创建一个根节点目录，当作分布式锁的顶级父目录（lock）.

2.客户端会在该lock下面创建一个临时顺序节点。比如第一个客户端在/lock/lock-0000000，第二客户端在/lock/lock-0000001等以此类推。

3.当一个客户端运行的时候首先获取锁目录下的所有子节点，然后在获取比自己小的兄弟节点，如果没有则说明其顺序号最小，获取锁。如果不是最小节点，则设置监听(watcher)比自己顺序号小的节点（只关注比自己次小的节点是为了防止发生“羊群效应”）

4.当客户端处理完，释放锁，删除自己的节点，另一个客户端会监听（watch）到变更事件，判断自己是否是最小节点，如果是，则获取锁，以此类推。

羊群效应:如果不监听比自己次序小的节点，在变更事件发生的时候，会同时所有的客户端，会阻塞其他操作。

节点删除并且监听事件发生是原子操作。不会造成该节点后面的节点获取不到监听事件的发生。

![201812261301021.png][]

翻译：

数据访问

存储在命名空间中每个znode的数据以原子方式读取和写入。读取获取与znode关联的所有数据字节，写入替换所有数据。

分布式锁实现：使用 Apache 开源的curator 方便来实现 Zookeeper 分布式锁。

curator API：

public void acquire() throws Exception;//获取锁
     public void release() throws Exception;//释放锁

Linux安装的zookeeper集群是 zookeeper-3.4.13因而项目中需要引入curator 2.12.0版本，如果不是则会报错：

KeeperException$UnimplementedException异常。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzU0OTU3OA_size_16_color_FFFFFF_t_70 1][]

官方解释：[点击][Link 1]

pom:

<dependency>
         <groupId>org.apache.curator</groupId>
         <artifactId>curator-recipes</artifactId>
         <version>2.12.0</version>
    </dependency>

源码分析：

InterProcessLock接口：

public interface InterProcessLock{
        /**
         * 获取互斥锁 - 阻塞直到它可用
         */
        public void acquire() throws Exception;
    
        /**
         * 获取互斥锁 - 阻塞直到它可用或给定时间到期
         */
        public boolean acquire(long time, TimeUnit unit) throws Exception;
    
        /**
         * 释放锁
         */
        public void release() throws Exception;
    
        /**
         * 如果此JVM中的线程获取了互斥锁，则返回true
         */
        boolean isAcquiredInThisProcess();
    }

获取锁：

@Override
    public boolean acquire(long time, TimeUnit unit) throws Exception{
            return internalLock(time, unit);
    }
    private boolean internalLock(long time, TimeUnit unit) throws Exception{
                //获取当前线程
                Thread currentThread = Thread.currentThread();
                //获取当前线程可重入数据凭证，使用该凭证可以获取可重入锁
                InterProcessMutex.LockData lockData = threadData.get(currentThread);
                if ( lockData != null )
                {
                    //原子递增一个当前值，记录重入次数，后面锁释放会用到
                    lockData.lockCount.incrementAndGet();
                    return true;
                }
                //尝试连接zookeeper获取锁
                String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
                if ( lockPath != null )
                {
                    //创建可重入锁数据凭证，用于记录当前线程重入次数
                    InterProcessMutex.LockData newLockData = new InterProcessMutex.LockData(currentThread, lockPath);
                    threadData.put(currentThread, newLockData);
                    return true;
                }
    
                return false;
     }

Zookeeper获取锁实现：

String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
        {
            //系统时间
            final long      startMillis = System.currentTimeMillis();
            //时间单位不存在，则自动转为秒
            final Long      millisToWait = (unit != null) ? unit.toMillis(time) : null;
            //本地锁节点数组
            final byte[]    localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes;
            //重试次数
            int             retryCount = 0;
    
            String          ourPath = null;
            boolean         hasTheLock = false;
            boolean         isDone = false;
            //
            while ( !isDone )
            {
                isDone = true;
    
                try
                {
                    //在锁节点下创建临时且有序的子节点 /curator/lock/_c_99d5c3d3-474a-4988-b670-828a0ef722b4-lock-0000000023
                    ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
                    //如果当前子节点序号最小，获得锁则直接返回，否则阻塞等待前一个子节点删除通知(release释放锁)
                    hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
                }
                catch ( KeeperException.NoNodeException e )
                {
                    
                    if ( 
                   //异常处理，如果找不到节点，这可能发生在session过期等时，可以使用重试机制
                  client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, 
                 System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
                    {
                        isDone = false;
                    }
                    else
                    {
                        throw e;
                    }
                }
            }
            //如果获取锁则返回当前锁子节点路径
            if ( hasTheLock )
            {
                return ourPath;
            }
    
            return null;
        }

返回锁：

private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception
        {
            boolean     haveTheLock = false;
            boolean     doDelete = false;
            try
            {
                if ( revocable.get() != null )
                {
                    client.getData().usingWatcher(revocableWatcher).forPath(ourPath);
                }
                //自旋获取锁
                while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock )
                {
                    //获取所有子节点集合
                    List<String> children = getSortedChildren();
    
                    //判断当前子节点是否为最小节点
                    String              sequenceNodeName = ourPath.substring(basePath.length() + 1); ///curator/lock/_c_99d5c3d3-474a-4988-b670-828a0ef722b4-lock-0000000023
    
                    PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
    
                    //满足最小节点则获取锁
                    if ( predicateResults.getsTheLock() )
                    {
                        haveTheLock = true;
                    }
                    else {
                        //获取前一个节点，用于监听
                        String  previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
                        synchronized(this) {
                            try {
                                //使用getData()接口而不是checkExists()是因为，如果前一个子节点已经被删除了那么会抛出异常而且不会设置事件监听器，
                                // 而checkExists虽然也可以获取到节点是否存在的信息但是同时设置了监听器，这个监听器其实永远不会触发。造成浪费，资源泄露
    //使用getData()代替checkExists()以避免留下不必要的监视器，这是一种资源泄漏类型                            client.getData().usingWatcher(watcher).forPath(previousSequencePath);
                                if ( millisToWait != null ) {
                                    millisToWait -= (System.currentTimeMillis() - startMillis);
                                    startMillis = System.currentTimeMillis();
                                    //如果设置了获取锁等待时间
                                    if ( millisToWait <= 0 ) {
                                        doDelete = true;    // 超时则删除子节点 释放锁避免死锁
                                        break;
                                    }
                                    //等待超时时间
                                    wait(millisToWait);
                                } else {
                                     //一直等待
                                    wait();
                                }
                            } catch ( KeeperException.NoNodeException e ) {
                               //如果前一个子节点已经被删除则deException，只需要自旋获取一次即可
                            }
                        }
                    }
                }
            }
            catch ( Exception e )
            {
                ThreadUtils.checkInterrupted(e);
                doDelete = true;
                throw e;
            }
            finally
            {
                //是否 超时
                if ( doDelete )
                {
                    deleteOurPath(ourPath);//删除节点
                }
            }
            return haveTheLock;
        }

释放锁：

@Override
        public void release() throws Exception
        {
            //获取当前线程
            Thread currentThread = Thread.currentThread();
            LockData lockData = threadData.get(currentThread);
            //没有获取锁，则释放锁会报错
            if ( lockData == null )
            {
                throw new IllegalMonitorStateException("You do not own the lock: " + basePath);
            }
            //重入锁次数
            int newLockCount = lockData.lockCount.decrementAndGet();
            if ( newLockCount > 0 )
            {
                return;
            }
            if ( newLockCount < 0 )
            {
                throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + basePath);
            }
            try
            {   //释放锁
                internals.releaseLock(lockData.lockPath);
            }
            finally
            {   //移除当前线程中可重入锁凭证
                threadData.remove(currentThread);
            }
        }

![2018122615570630.png][]

案例：

package com.example.zookeeper.demo;
    
    import org.apache.curator.RetryPolicy;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.framework.recipes.locks.InterProcessMutex;
    import org.apache.curator.retry.ExponentialBackoffRetry;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class ZookeeperLockUtils {
    
        private Lock lock=new ReentrantLock(true);//公平锁
    
        private static String address = "94.191.86.252:2181,94.191.86.252:2182,94.191.86.252:2183";
        public static CuratorFramework client;
        private static InterProcessMutex interProcessMutex;
        //使用static代码块实现单例
        static{
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
            client = CuratorFrameworkFactory.newClient(address, retryPolicy);
            client.start();
            interProcessMutex=new InterProcessMutex(client, "/curator/lock");
        }
        //私有构造 保证外界无法直接实例化
        private ZookeeperLockUtils(){
    
        }
        public static InterProcessMutex getMutex(){
            return interProcessMutex;
        }
        //获得了锁
        public static boolean acquire(long time, TimeUnit unit){
            try {
                return getMutex().acquire(time,unit);
            } catch (Exception e) {
                e.printStackTrace();
                return false;
            }
        }
        //释放锁
        public static void release(){
            try {
                getMutex().release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) {
    
            ExecutorService executorService = Executors.newFixedThreadPool(5);
            for(int i=0;i<10000;i++){
                Runnable task = new Runnable() {
                    @Override
                    public void run() {
                        getInfo();
                    }
                };
                executorService.execute(task);
            }
        }
    
        public static void getInfo(){
            boolean flag = false;
            try {
                //尝试获取锁，最多等待5秒
                flag = ZookeeperLockUtils.getMutex().acquire(5, TimeUnit.SECONDS);
                Thread currentThread = Thread.currentThread();
                if(flag){
                    System.out.println("线程"+currentThread.getId()+"获取锁成功");
                }else{
                    System.out.println("线程"+currentThread.getId()+"获取锁失败");
                }
                System.out.println("业务逻辑");
            } catch (Exception e) {
                e.printStackTrace();
            } finally{
                if(flag){
                    try {
                        ZookeeperLockUtils.getMutex().release();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    
    
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzU0OTU3OA_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/6910de3615fe4e0bbc69c62c380a3515.png
[2018122516540839]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/ebfafecd192c4a83a1e4b6a57e66e499.png
[201812261301021.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/47b083c6c6b041658e674bc6b8ff1aa9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzU0OTU3OA_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/f43a9386fe95435dbc43e0dc03d079e9.png
[Link 1]: http://curator.apache.org/zk-compatibility.html
[2018122615570630.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/3104439309b64d728a0c1474f9c47a13.png