在Java中操作Zookeeper，使用Zookeeper实现分布式锁

浅浅的花香味﹌ 2023-02-14 02:10 12阅读 0赞

### 目录 ###

*   *   *  在java中操作zk
         *   *  依赖
             *  连接到zk server
             *  检测节点是否存在
             *  创建节点
             *  删除节点
             *  获取子节点列表
             *  获取节点数据
             *  设置、更新节点数据
             *  设置acl权限
             *  查看acl权限
             *  添加监听
             *  移除监听
         *  使用zk实现分布式锁的3种方式

### 在java中操作zk ###

#### 依赖 ####

<dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.6.0</version>
    </dependency>

#### 连接到zk server ####

//连接字符串,zkServer的ip、port，如果是集群逗号分隔
    String connectStr = "192.168.1.1:2181";
    
    //zookeeper对象就是一个zkCli
    ZooKeeper zooKeeper = null;
    
    try { 
    //初始次数为1。后面要在内部类中使用，三种写法：1、写成外部类成员变量，不用加final；2、作为函数局部变量，放在try外面，写成final；3、写在try中，不加final
        CountDownLatch countDownLatch = new CountDownLatch(1);
        //超时时间ms，监听器
        zooKeeper = new ZooKeeper(connectStr, 5000, new Watcher() { 
            public void process(WatchedEvent watchedEvent) { 
                //如果状态变成已连接
                if (watchedEvent.getState().equals(Event.KeeperState.SyncConnected)) { 
                    System.out.println("连接成功");
                    //次数-1
                    countDownLatch.countDown();
                }
            }
        });
        //等待，次数为0时才会继续往下执行。等待监听器监听到连接成功，才能操作zk
        countDownLatch.await();
    } catch (IOException | InterruptedException e) { 
        e.printStackTrace();
    }
    
    
    //...操作zk。后面的demo都是写在此处的
    
    
    //关闭连接
    try { 
        zooKeeper.close();
    } catch (InterruptedException e) { 
        e.printStackTrace();
    }

#### 检测节点是否存在 ####

// 检测节点是否存在
    
    // 同步方式
    Stat exists = null;
    try { 
        //如果存在，返回节点状态Stat；如果不存在，返回null。第二个参数是watch
        exists = zooKeeper.exists("/mall",false);
    } catch (KeeperException | InterruptedException e) { 
        e.printStackTrace();
    }
    if (exists==null){ 
        System.out.println("节点不存在");
    }
    else { 
        System.out.println("节点存在");
    }
    
    
    //异步回调
    zooKeeper.exists("/mall",false, new AsyncCallback.StatCallback() { 
        //第二个是path znode路径，第三个是ctx 后面传入实参，第四个是znode的状态
        public void processResult(int i, String s, Object o, Stat stat) { 
            //如果节点不存在，返回的stat是null
            if (stat==null){ 
                System.out.println("节点不存在");
            }
            else{ 
                System.out.println("节点存在");
            }
        }
    // 传入ctx，Object类型
    },null);

操作后，服务端会返回处理结果，返回的void、null也算处理结果。

同步指的是当前线程阻塞，等待服务端返回数据，收到返回的数据才继续往下执行；异步回调指的是，把对结果（返回的数据）的处理写在回调函数中，当前线程继续往下执行，收到返回的数据时自动调用回调函数来处理。

#### 创建节点 ####

//创建节点
    
    //同步方式
    try { 
        //数据要写成byte[]，不携带数据写成null；默认acl权限使用ZooDefs.Ids.OPEN_ACL_UNSAFE；最后一个是节点类型，P是永久，E是临时，S是有序
        zooKeeper.create("/mall", "abcd".getBytes(),  ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("已创建节点/mall");
        //如果节点已存在，会抛出异常
    } catch (KeeperException | InterruptedException e) { 
    　　　　 System.out.println("创建节点/mall失败，请检查节点是否已存在");
        e.printStackTrace();
    }
    
    
    //异步回调
    zooKeeper.create("/mall", "abcd".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, new AsyncCallback.Create2Callback(){ 
        //第二个path，第三个ctx，第四个节点状态
        public void processResult(int i, String s, Object o, String s1, Stat stat) { 
            //回调方式不抛出异常，返回的stat是创建节点的状态，如果节点已存在，返回的stat是null
            if (stat==null){ 
                System.out.println("创建节点/mall失败，请检查节点是否已存在");
            }
            else { 
                System.out.println("节点创建成功");
            }
        }
        //ctx实参
    },null);

#### 删除节点 ####

//删除节点
    
    //同步方式
    try { 
        //第二个参数是版本号，-1表示可以是任何版本
        zooKeeper.delete("/mall1",-1);
        System.out.println("成功删除节点/mall");
    } catch (InterruptedException | KeeperException e) { 
        System.out.println("删除节点/mall失败");
        e.printStackTrace();
    }
    
    
    //异步回调
    zooKeeper.delete("/mall2", -1, new AsyncCallback.VoidCallback() { 
        //第二个是path，第三个是ctx
        public void processResult(int i, String s, Object o) { 
            
        }
    //
    //ctx实参
    },null);

delete()只能删除没有子节点的znode，如果该znode有子节点会抛出异常。

没有提供递归删除子节点的方法，如果要删除带有子节点的znode，需要自己实现递归删除。可以先getChildren()获取子节点列表，遍历列表依次删除子节点，再删除父节点。

#### 获取子节点列表 ####

//获取子节点列表，List<String>，比如/mall/user,/mall/order，返回的是["user"、"order"]
    
    //同步方式
    List<String> children = null;
    try { 
        //第二个参数是watch
        children = zooKeeper.getChildren("/mall", false);
    } catch (KeeperException | InterruptedException e) { 
        e.printStackTrace();
    }
    System.out.println("子节点列表：" + children);
    
    
    //异步
    zooKeeper.getChildren("/mall", false, new AsyncCallback.ChildrenCallback() { 
        //第二个起依次是：path、ctx、返回的子节点列表
        public void processResult(int i, String s, Object o, List<String> list) { 
            System.out.println("子节点列表：" + list);
        }
    //ctx实参
    }, null);

只获取子节点，不获取孙节点。

watch可以写布尔值，要添加监听就写true，不监听写false；也可以写成Watcher对象，new一个Watcher对象表示要监听，null表示不监听。

#### 获取节点数据 ####

//获取节点数据，返回byte[]
    
    //同步方式
    byte[] data = null;
    try { 
        //第二个参数是watch，第三个是stat
        data = zooKeeper.getData("/mall", false, null);
    } catch (KeeperException | InterruptedException e) { 
        e.printStackTrace();
    }
    //调用new String()时要判断data是否为null，如果是null会抛NPE
    if (data==null){ 
        System.out.println("该节点没有数据");
    }
    else{ 
        System.out.println("节点数据："+new String(data));
    }
    
    
    //异步回调
    zooKeeper.getData("/mall", false, new AsyncCallback.DataCallback() { 
        //第二个起依次是：path、ctx、返回的节点数据、节点状态
        public void processResult(int i, String s, Object o, byte[] bytes, Stat stat) { 
            //不必判断bytes是否是null，如果节点没有数据，不会调用回调函数；执行到此，说明bytes不是null
            System.out.println("节点数据：" + new String(bytes) );
        }
        //ctx实参
    }, null);

#### 设置、更新节点数据 ####

//设置|更新节点据
    
    //同步方式
    try { 
        //最后一个参数是版本号
        zooKeeper.setData("/mall", "1234".getBytes(), -1);
        System.out.println("设置节点数据成功");
    } catch (KeeperException | InterruptedException e) { 
        System.out.println("设置节点数据失败");
        e.printStackTrace();
    }
    
    
    //异步回调
    zooKeeper.setData("/mall", "1234".getBytes(), -1, new AsyncCallback.StatCallback() { 
        //第二个是path，第三个是ctx
        public void processResult(int i, String s, Object o, Stat stat) { 
    
        }
    // ctx
    },null);

#### 设置acl权限 ####

//设置acl权限
        
    //第一个参数指定权限,第二个是Id对象
    ACL acl = new ACL(ZooDefs.Perms.ALL, new Id("auth", "chy:abcd"));
    
    List<ACL> aclList = new ArrayList<>();
    aclList.add(acl);
    
    //如果List中只有一个ACL对象，也可以这样写
    //List<ACL> aclList = Collections.singletonList(auth);
        
    //验证权限，需写在设置权限之前。如果之前没有设置权限，也需要先验证本次即将设置的用户
    zooKeeper.addAuthInfo("digest","chy:abcd".getBytes());
    
    
    //方式一 setAcl
    try { 
        //第二个参数是List<ACL>，第三个参数是版本号
        zooKeeper.setACL("/mall", aclList, -1);
        System.out.println("设置权限成功");
    } catch (KeeperException | InterruptedException e) { 
        System.out.println("设置权限失败");
        e.printStackTrace();
    }
    
    
    //方式二 在创建节点时设置权限
    try { 
        zooKeeper.create("/mall","abcd".getBytes(),aclList,CreateMode.PERSISTENT);
        System.out.println("已创建节点并设置权限");
    } catch (KeeperException | InterruptedException e) { 
        e.printStackTrace();
    }

设置权限之后，连接zk server进行操作时，需要先验证用户。  
此处未写对应的异步回调。

#### 查看acl权限 ####

//查看acl权限
        
    //设置权限之后，以后操作时需要先验证用户，一次session中验证一次即可
    zooKeeper.addAuthInfo("digest","chy:abcd".getBytes());
    
    
    //同步方式
    try { 
        List<ACL> aclList = zooKeeper.getACL("/mall", null);
        System.out.println("acl权限："+aclList);
    } catch (KeeperException | InterruptedException e) { 
        System.out.println("获取acl权限失败");
        e.printStackTrace();
    }
    
    
    //异步回调
    zooKeeper.getACL("/mall3", null, new AsyncCallback.ACLCallback() { 
        //第二个起：path、ctx、获取到的List<ACL>、节点状态
        public void processResult(int i, String s, Object o, List<ACL> list, Stat stat) { 
            //就算没有手动设置acl权限，默认也是有值的
            System.out.println("acl权限："+list);
        }
    //ctx实参
    },null);

#### 添加监听 ####

//添加监听 方式一
    try { 
        CountDownLatch countDownLatch = new CountDownLatch(1);
    
        zooKeeper.getData("/mall", new Watcher() { 
            public void process(WatchedEvent watchedEvent) { 
                //watcher会监听该节点所有的事件，不管发生什么事件都会调用process()来处理，需要先判断一下事件类型
                if (watchedEvent.getType().equals(Event.EventType.NodeDataChanged)){ 
                    System.out.println("节点数据改变了");
                    //会一直监听，如果只监听一次数据改变，将下面这句代码取消注释即可
                    //countDownLatch.countDown();
                }
            }
        }, null);
        //默认watcher是一次性的，如果要一直监听，需要借助CountDownLatch
        countDownLatch.await();
    } catch (KeeperException | InterruptedException e) { 
        e.printStackTrace();
    }

exists()、getData()、getChildren()都具有添加监听的功能，用法类似。

会递归监听子孙节点，子孙节点的数据改变也算NodeDataChanged，子孙节点的创建|删除也算NodeCreated|NodeDeleted。

//添加监听 方式二 可指定是否递归监听子孙节点 
    try { 
        CountDownLatch countDownLatch1 = new CountDownLatch(1);
        zooKeeper.addWatch("/mall", new Watcher() { 
            @Override
            public void process(WatchedEvent watchedEvent) { 
                if (watchedEvent.getType().equals(Event.EventType.NodeDataChanged)){ 
                    System.out.println("节点数据改变了");
                    //如果只监听一次数据改变，将下面这句代码注释掉
                    //countDownLatch1.countDown();
                }
            }
        //监听模式，PERSISTENT是不监听子孙节点，PERSISTENT_RECURSIVE是递归监听子孙节点
        }, AddWatchMode.PERSISTENT_RECURSIVE);
        countDownLatch1.await();
    } catch (KeeperException | InterruptedException e) { 
        e.printStackTrace();
    }

countDownLatch1.await(); 会阻塞线程，最好启动一条新线程来监听。

#### 移除监听 ####

//移除监听 方式一
    try { 
        zooKeeper.addWatch("/mall",null,AddWatchMode.PERSISTENT);
        System.out.println("已移除监听");
    } catch (KeeperException | InterruptedException e) { 
        e.printStackTrace();
    }

就是上面添加监听的那些方法，watch|watcher参数，如果是boolean类型，设置为false即可关闭监听；如果是Watcher类型，可设置为null覆盖之前设置的监听。

会移除整个监听，不需要传入监听对象watcher。

//移除监听 方式二
    try { 
        //第二个参数是Watcher，原来添加的那个Watcher监听对象，不能是null
        //第三个参数指定要移除监听的哪部分，Any是移除整个监听，Data是移除对数据的监听，Children是移除对子节点的递归监听
        //最后一个参数指定未连接到zkServe时，是否移除本地监听部分
        zooKeeper.removeWatches("/mall",watcher, Watcher.WatcherType.Children,true);
    } catch (InterruptedException | KeeperException e) { 
        e.printStackTrace();
    }

监听由2部分组成，一部分在zkServer上，事件发生时通知对应的zkCli；一部分在zkCli（本地），收到zkServer的通知时做出一些处理。最后一个参数指定未连接到zkServer时，是否移除本地（zkCli）监听部分。如果移除了本地监听，就算zkServer进行通知，zkCli也不会做出反应。

这种方式功能更全，可以指定移除监听的哪个部分，但需要传入watcher对象，添加监听时要用一个变量来保存watcher对象。

### 使用zk实现分布式锁的3种方式 ###

**第1种**  
创建一个临时znode，如果创建成功，则获取到锁，操作完删除znode释放锁；  
如果创建失败（抛出异常），当前线程休眠一小会儿，之后重试，设置一个计数器，重试指定次数后还没有获取到锁就放弃。

缺点：未获取到锁时会重试多次，浪费资源

**第2种**  
在第一种的基础修改，如果创建失败、节点已存在，给znode加一个watcher，监听节点删除事件（释放锁），当前线程休眠，节点删除事件发生时唤醒线程。

缺点：可能发生惊群现象。如果多个线程同时等待锁，释放锁后等待锁的线程都会被唤醒，但只有一个线程可以获取到锁。

**第3种**  
在第二种的基础上改，要获取锁时先在指定节点下创建有序的临时znode作为子节点，获取指定节点的子节点数量

*  如果只有1个子节点，则当前线程获取到锁，操作完删除自己创建的节点或者关闭连接自动删除；
 *  如果子节点数量>1，则获取第二小的节点，监听它的节点删除事件，在回调函数中写操作。

优点：避免了惊群现象；获取锁的顺序和等待顺序一致。

无论使用哪一种，都是创建临时节点，避免获取到锁后机器故障，不能及时释放锁。

自己写代码实现zk的分布式锁有点麻烦，可以使用现成的轮子Curator。

如果方便使用redis，尽量使用redis来实现分布式锁，redis与springboot的整合更好，实现分布式锁也更简单。