Zookeeper 分布式锁原理深入分析及实战

我会带着你远行 2023-10-02 18:33 8阅读 0赞

今天我们分析Zookeeper 分布式锁原理及实战，分布式锁在分布式服务中一定会用到的，且ZK的分布式锁相对比较稳定，可用性比较好，因此我们深入分析一下。

1、ZK分布式锁原理图：

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_20_color_FFFFFF_t_70_g_se_x_16][]

1.1，争抢锁，只有一个人能获得锁

1.2，获得锁的客户端出问题，临时节点（session）就会自动删除，不需要设置 过期时间，也不会造成死锁问题。

1.3，获得锁的人成功了，使用后自动释放锁。

2、锁被释放、删除，其他客户端怎么知道，如图ZK的监听机制：

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

2.1：（可以想到的方案1）主动轮询，心跳检测等 弊端：延迟，压力

2.2：（可以想到的方案2）watch： 解决延迟问题 弊端：压力

2.3：（ZK的解决方案）sequence+watch：watch谁？watch前一个，最小的获得锁~！一旦， 最小的释放了锁，成本：zk只给第二个发事件回调，因此压力小，延时小

3、分布式锁业务代码测试：

3.1、客户端连接Zk集群：

import org.apache.zookeeper.ZooKeeper;
    
    import java.util.concurrent.CountDownLatch;
    
    /**
     * @author: nandao
     * @create: 2021-08-20 11:20
     */
    public class ZKUtils {
    
        private  static ZooKeeper zk;
    
        private static String address = "192.168.18.11:2181,192.168.18.12:2181,192.168.18.13:2181,192.168.18.14:2181/testLock";
    
        private static DefaultWatch watch = new DefaultWatch();
    
        private static CountDownLatch init  =  new CountDownLatch(1);
        public static ZooKeeper  getZK(){
    
            try {
                zk = new ZooKeeper(address,1000,watch);
                watch.setCc(init);
                init.await();
    
            } catch (Exception e) {
                e.printStackTrace();
            }
            return zk;
        }
    }

3.2、获取锁、释放锁、回调接口的工具类：

import org.apache.zookeeper.*;
    import org.apache.zookeeper.data.Stat;
    
    import java.util.Collections;
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    
    /**
     * @author: nandao
     * @create: 2021-08-20 11:21
     */
    public class WatchCallBack   implements Watcher  ,AsyncCallback.StringCallback ,AsyncCallback.Children2Callback ,AsyncCallback.StatCallback {
    
        ZooKeeper zk ;
        String threadName;
        CountDownLatch cc = new CountDownLatch(1);
        String pathName;
    
        public String getPathName() {
            return pathName;
        }
    
        public void setPathName(String pathName) {
            this.pathName = pathName;
        }
    
        public String getThreadName() {
            return threadName;
        }
    
        public void setThreadName(String threadName) {
            this.threadName = threadName;
        }
    
        public ZooKeeper getZk() {
            return zk;
        }
    
        public void setZk(ZooKeeper zk) {
            this.zk = zk;
        }
    
        public void tryLock(){
            try {
                System.out.println(threadName + "  create....");
    //            if(zk.getData("/"))
                zk.create("/lock",threadName.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL,this,"abc");
                cc.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        public void unLock(){
            try {
                zk.delete(pathName,-1);
                System.out.println(threadName + " over work....");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (KeeperException e) {
                e.printStackTrace();
            }
        }
    
    
        @Override
        public void process(WatchedEvent event) {
            //如果第一个客户端那个锁释放了，其实只有第二个收到了回调事件！
            //如果，不是第一个客户端，某一个挂了，也能造成他后边的收到这个通知，从而让他后边那个跟去watch挂掉这个哥们前边的。
            switch (event.getType()) {
                case None:
                    break;
                case NodeCreated:
                    break;
                case NodeDeleted:
                    zk.getChildren("/",false,this ,"sdf");
                    break;
                case NodeDataChanged:
                    break;
                case NodeChildrenChanged:
                    break;
            }
        }
    
        @Override
        public void processResult(int rc, String path, Object ctx, String name) {
            if(name != null ){
                System.out.println(threadName  +"  create node : " +  name );
                pathName =  name ;
                zk.getChildren("/",false,this ,"sdf");
            }
        }
    
        //getChildren  call back
        @Override
        public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
            //一定能看到自己前边的。。
    //        System.out.println(threadName+"look locks.....");
    //        for (String child : children) {
    //            System.out.println(child);
    //        }
            Collections.sort(children);
            int i = children.indexOf(pathName.substring(1));
            //是不是第一个
            if(i == 0){
                //yes
                System.out.println(threadName +" i am first....");
                try {
                    zk.setData("/",threadName.getBytes(),-1);
                    cc.countDown();
    
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }else{
                //no
                zk.exists("/"+children.get(i-1),this,this,"sdf");
            }
        }
    
        @Override
        public void processResult(int rc, String path, Object ctx, Stat stat) {
            //暂时不写
        }
    }

3.3、获取ZK的连接、获取锁、释放锁的业务代码：

import org.apache.zookeeper.ZooKeeper;
    import org.junit.After;
    import org.junit.Before;
    import org.junit.Test;
    
    /**
     * @author: nandao
     * @create: 2021-08-20 11:21
     */
    public class TestLock {
    
        ZooKeeper zk ;
    
        @Before
        public void conn (){
            zk  = ZKUtils.getZK();
        }
    
        @After
        public void close (){
            try {
                zk.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        @Test
        public void lock(){
            for (int i = 0; i < 10; i++) {
                new Thread(){
                    @Override
                    public void run() {
                        WatchCallBack watchCallBack = new WatchCallBack();
                        watchCallBack.setZk(zk);
                        String threadName = Thread.currentThread().getName();
                        watchCallBack.setThreadName(threadName);
                        //每一个线程：
                        //抢锁
                        watchCallBack.tryLock();
                        //干活
                        System.out.println(threadName+" working...");
    //                    try {
    //                        Thread.sleep(1000);
    //                    } catch (InterruptedException e) {
    //                        e.printStackTrace();
    //                    }
                        //释放锁
                        watchCallBack.unLock();
                    }
                }.start();
            }
            while(true){
    
            }
        }
    }

核心伪代码已经贴出来了，大家一定要多测试多思考，定会很快理解。

[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/a3962916d1c54c469265e45c0543d09a.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBAbmFuZGFvMTU4,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/cc0fceb2e2cd4037b36e0ab123822e87.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBAbmFuZGFvMTU4,size_20,color_FFFFFF,t_70,g_se,x_16