Zookeeper API

痛定思痛。 2022-04-24 09:34 183阅读 0赞

**Zookeeper类是线程安全的。**

使用客户端来创建一个和zk服务端连接的句柄，这就是一个会话（session）。Session一旦建立，状态就是连接中（CONNECTING）状态，然后客户端会尝试去连接zk服务端，连接成功之后状态变成已连接（CONNECTED）。一般正常情况下只会有这两个状态。不过，还是会发生一些无法恢复的错误/故障，比如：session过期，认证失败，或者客户端关闭连接，这种情况下，session状态会变成关闭（CLOSED）状态。

# 1、Watcher（ 数据变更的通知） #

ZooKeeper提供了分布式数据的**发布/订阅**功能。一个典型的发布/订阅模型系统定义了一种一对多的订阅关系，能够让多个订阅者同时监听某一个主题对象，当这个主题对象自身状态变化时，会通知所有订阅者，使他们能够做出相应的处理。在ZooKeeper中，引入了Watcher机制来实现这种分布式的通知功能。ZooKeeper允许客户端向服务端注册一个Watcher机制来实现这种分布式的通知功能。ZooKeeper允许客户端向服务端注册一个Watcher监听，当服务端的一些指定事件触发了这个Watcher，那么就会向指定客户端发送一个事件通知来实现分布式的通知功能。

整个Watcher注册与通知过程如下图所示。

![20190422214540904.png][]

从上图中，我们可以看到，ZooKeeper的Watcher机制主要包括客户端线程、客户端WatchManager和ZooKeeper服务器三部分。在具体工作流程上，简单地讲，客户端在向ZooKeeper服务器注册Watcher的同时，会将Watcher对象存储在客户端的WatchManager中。当ZooKeeper服务器端触发Watcher事件后，会向客户端发送通知，客户端线程从WatchManager中取出对应的Watcher对象来执行回调逻辑。

### Watcher接口 ###

在ZooKeeper中，接口类Watcher用于表示一个标准的事件处理器，其定义了事件通知相关的逻辑，包含KeeperState和EventType两个枚举类，分别代表了通知状态和事件类型，同时定义了事件的回调方法：process(WatchedEvent event)。

### Watcher事件 ###

同一个事件类型在不同的通知状态中代表的含义有所不同，下表列举了常见的通知状态和事件类型。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70][]

上表中列举了ZooKeeper中最常见的几个通知状态和事件类型。其中，针对NodeDataChanged事件，此处说的变更包括节点的数据内容和数据的版本号dataVersion。因此即使使用相同的数据内容来更新，还是会触发这个事件通知，因为对于ZooKeeper来说，无论数据内容是否变更，一旦有客户端调用了数据更新的接口，且更新成功，就会更新dataVersion值。

NodeChildrenChanged事件会在数据节点的子节点列表发生变更的时候被触发，这里说的子节点列表变化特指子节点个数和组合情况的变更，即新增子节点或删除子节点，而子节点内容的变化是不会触发这个事件的。

对于AuthFailed这个事件，需要注意的地方是，他的触发条件并不是简简单单因为当前客户端会话没有权限，而是授权失败。我们首先通过下面的两个例子来看看AuthFailed事件。

### 回调方法process() ###

process方法是Watcher接口中的一个回调方法，当ZooKeeper向客户端发送一个Watcher事件通知时，客户单就会对相应的process方法进行回调，从而实现对事件的处理。process方法的定义如下：

*  **abstract public void process(WatchedEvent event);**

这个回掉方法的定义非常简单，我们重点看下方法的参数定义：WatchedEvent。

WatchedEvent包含了每一个事件的三个基本方法：通知状态（keeperState）、事件类型（eventType）和节点路径（path），其数据结构如下图所示。ZooKeeper使用WatchedEvent对象来封装服务端事件并传递给Watcher，从而方便回调方法process对服务端事件进行处理。

提到WatchedEvent，不得不讲下WatcherEvent实体。笼统的讲，两者表示的是同一个事务，都是对一个服务端事件的封装。不同的是，WatchedEvent是一个逻辑事件，用于服务端和客户端程序执行过程中所需的逻辑对象，而WatcherEvent因为实现了序列化接口，因此可以用于网络传输，其数据结构如下图所示。

服务端在生成WatchedEvent事件之后，会调用getWrapper方法将自己包装成一个可序列化的WatcherEvent事件，以便通过网络传输到客户端。客户端在接收到服务端的这个事件对象后，首先会将WatcherEvent事件还原成一个WatchedEvent事件，并传递给process方法处理，回调方法process根据入参就能够解析出完整的服务端事件了。

需要注意的一点是，无论是WatchedEvent还是WatcherEvent，其对ZooKeeper服务端事件的封装都是及其简单的。举个例子来说，当/zk-book这个节点的数据发生变更时，服务端会发送给客户端一个“ZNode数据内容变更”事件，客户端只能收到如下信息：

![20190422215357475.png][]

从上面展示的信息中，我们可以看到，客户端无法直接从该事件中获取到对应数据节点的原始数据内容以及变更后的新数据内容，而是需要客户端再次主动去重新获取数据——这也是ZooKeeper Watcher机制的一个非常重要的特性。

# **2、Zookeeper API使用** #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70 1][]

**1、jar包引入，演示版本为3.4.6，非maven项目，可以下载jar包导入到项目中**

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

**2、创建zookeeper连接**

ZooKeeper(java.lang.String connectString, int sessionTimeout, org.apache.zookeeper.Watcher watcher)

*   *  connectString：zookeeper服务地址，例如“192.168.117.128:2181”
     *  sessionTimeout ：超时时间，单位为毫秒
     *  watcher：实现org.apache.zookeeper.Watcher接口的实现类，需实现process(WatchedEvent watchedEvent) 方法

ZooKeeper zooKeeper = new ZooKeeper("192.168.117.128:2181",5000, new MyWatcher());

示例代码：

import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooKeeper;
    import java.io.IOException;
    /**
     * zookeeper连接
     */
    public class CreateSession implements Watcher{
        private static ZooKeeper zooKeeper;
    
        @Override
        public void process(WatchedEvent watchedEvent) {
    
            if(watchedEvent.getState().equals(Event.KeeperState.SyncConnected)) {
                doBus();
            }
            System.out.println("接收内容："+watchedEvent.toString());
        }
    
        private void doBus() {
            System.out.println("做业务！");
        }
    
        public static void main(String[] args) {
            try {
                zooKeeper = new ZooKeeper("192.168.117.128:2181",5000, new CreateSession());
                System.out.println(zooKeeper.getState());
                Thread.sleep(5000);
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

**3、同步创建**

create(java.lang.String path, byte[] data, java.util.List<org.apache.zookeeper.data.ACL> acl, 
    org.apache.zookeeper.CreateMode createMode)

*  path：创建节点路径，需保证父节点已存在（有序的）
 *  data：节点数据
 *  acl: 权限列表
    
     *  提供默认的权限OPEN\_ACL\_UNSAFE、CREATOR\_ALL\_ACL、READ\_ACL\_UNSAFE
        
         *  OPEN\_ACL\_UNSAFE：完全开放
         *  CREATOR\_ALL\_ACL：创建该znode的连接拥有所有权限
         *  READ\_ACL\_UNSAFE：所有的客户端都可读

**     自定义权限**

ACL aclIp = new ACL(ZooDefs.Perms.READ,new Id("ip","127.0.0.1"));
                    ACL aclDigest = new ACL(ZooDefs.Perms.READ| ZooDefs.Perms.WRITE,
                            new Id("digest", DigestAuthenticationProvider.generateDigest("id:pass")));

**createMode:节点类型**

*  PERSISTENT：持久化节点
 *  PERSISTENT\_SEQUENTIAL：持久化有序节点
 *  EPHEMERAL：临时节点（连接断开自动删除）
 *  EPHEMERAL\_SEQUENTIAL：临时有序节点（连接断开自动删除）

代码：

public class Zookeeper_Create_API_Sync_Usage implements Watcher {
        private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
        
        public static void main(String[] args) throws Exception {
            ZooKeeper zookeeper = new ZooKeeper("127.0.0.1:2181", 5000, new Zookeeper_Create_API_Sync_Usage());
            System.out.println(zookeeper.getState());
            connectedSemaphore.await();
            
            String path1 = zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);    
            System.out.println("Success create znode: " + path1);
            
            String path2 = zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);    
            System.out.println("Success create znode: " + path2);
        }
    
        public void process(WatchedEvent event) {
            if (KeeperState.SyncConnected == event.getState()) {
                connectedSemaphore.countDown();
            }
        }
    }

运行结果如下

CONNECTING 
    Success create znode: /zk-test-ephemeral- 
    Success create znode: /zk-test-ephemeral-0000000043

结果表明已经成功创建了临时节点和临时顺序节点，在创建顺序节点时，系统会在后面自动增加一串数字。

**② 异步方式**

使用异步方式于同步方式的区别在于节点的创建过程（包括网络通信和服务端的节点创建过程）是异步的，在同步接口调用过程中，开发者需要关注接口抛出异常的可能，但是在异步接口中，**接口本身不会抛出异常，所有异常都会在回调函数中通过Result Code来体现**。

public class Zookeeper_Create_API_ASync_Usage implements Watcher {
        private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    
        public static void main(String[] args) throws Exception {
            ZooKeeper zookeeper = new ZooKeeper("127.0.0.1:2181", 5000, new Zookeeper_Create_API_ASync_Usage());
            System.out.println(zookeeper.getState());
            connectedSemaphore.await();
    
            zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
                    new IStringCallback(), "I am context. ");
    
            zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
                    new IStringCallback(), "I am context. ");
    
            zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,
                    new IStringCallback(), "I am context. ");
            Thread.sleep(Integer.MAX_VALUE);
        }
    
        public void process(WatchedEvent event) {
            if (KeeperState.SyncConnected == event.getState()) {
                connectedSemaphore.countDown();
            }
        }
    }
    
    class IStringCallback implements AsyncCallback.StringCallback {
        public void processResult(int rc, String path, Object ctx, String name) {
            System.out.println("Create path result: [" + rc + ", " + path + ", " + ctx + ", real path name: " + name);
        }
    }

运行结果如下

CONNECTING 
    Create path result: [0, /zk-test-ephemeral-, I am context. , real path name: /zk-test-ephemeral- 
    Create path result: [-110, /zk-test-ephemeral-, I am context. , real path name: null 
    Create path result: [0, /zk-test-ephemeral-, I am context. , real path name: /zk-test-ephemeral-0000000045

结果表明已经成功使用异步方式创建了相应节点。

**4、 删除节点　**

只允许删除叶子节点，即一个节点如果有子节点，那么该节点将无法直接删除，必须先删掉其所有子节点。同样也有同步和异步两种方式。

public class Delete_API_Sync_Usage implements Watcher {
        private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
        private static ZooKeeper zk;
    
        public static void main(String[] args) throws Exception {
            String path = "/zk-book";
            zk = new ZooKeeper("127.0.0.1:2181", 5000,
                    new Delete_API_Sync_Usage());
            connectedSemaphore.await();
    
            zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println("success create znode: " + path);
            zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println("success create znode: " + path + "/c1");
            try {
                zk.delete(path, -1);
            } catch (Exception e) {
                System.out.println("fail to delete znode: " + path);
            }
            
            zk.delete(path + "/c1", -1);
            System.out.println("success delete znode: " + path + "/c1");
            zk.delete(path, -1);        
            System.out.println("success delete znode: " + path);
    
            Thread.sleep(Integer.MAX_VALUE);
        }
    
        public void process(WatchedEvent event) {
            if (KeeperState.SyncConnected == event.getState()) {
                if (EventType.None == event.getType() && null == event.getPath()) {
                    connectedSemaphore.countDown();
                }
            }
        }
    }

运行结果如下

success create znode: /zk-book 
    success create znode: /zk-book/c1 
    fail to delete znode: /zk-book 
    success delete znode: /zk-book/c1 
    success delete znode: /zk-book

结果表明若节点有子节点，则无法将其删除，必须先删除其所有子节点。

**5、子节点获取**

读取节点的子节点列表，同样可以使用同步和异步的方式进行操作。

public class Zookeeper_GetChildren_API_Sync_Usage implements Watcher {
        private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
        private static ZooKeeper zk = null;
    
        public static void main(String[] args) throws Exception {
            String path = "/zk-book-1";
            zk = new ZooKeeper("127.0.0.1:2181", 5000, new Zookeeper_GetChildren_API_Sync_Usage());
            connectedSemaphore.await();
    
            zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println("success create znode: " + path);
            zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            System.out.println("success create znode: " + path + "/c1");
            List<String> childrenList = zk.getChildren(path, true);
            System.out.println(childrenList);
    
            zk.create(path + "/c2", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            System.out.println("success create znode: " + path + "/c2");
            Thread.sleep(1000);
            zk.create(path + "/c3", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            System.out.println("success create znode: " + path + "/c3");
            Thread.sleep(Integer.MAX_VALUE);
        }
    
        public void process(WatchedEvent event) {
            if (KeeperState.SyncConnected == event.getState()) {
                if (EventType.None == event.getType() && null == event.getPath()) {
                    connectedSemaphore.countDown();
                } else if (event.getType() == EventType.NodeChildrenChanged) {
                    try {
                        System.out.println("ReGet Child:" + zk.getChildren(event.getPath(), true));
                    } catch (Exception e) {
                    }
                }
            }
        }
    }

运行结果如下

success create znode: /zk-book-1
    success create znode: /zk-book-1/c1
    [c1]
    success create znode: /zk-book-1/c2
    ReGet Child:[c1, c2]
    success create znode: /zk-book-1/c3
    ReGet Child:[c3, c1, c2]

值得注意的是，**Watcher通知是一次性的，即一旦触发一次通知后，该Watcher就失效了**，因此客户端需要反复注册Watcher，即程序中在process里面又注册了Watcher，否则，将无法获取c3节点的创建而导致子节点变化的事件。

**6、数据节点获取**

对于节点的数据获取，同样存在同步和异步两种方式。

public class GetData_API_Sync_Usage implements Watcher {
        private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
        private static ZooKeeper zk = null;
        private static Stat stat = new Stat();
    
        public static void main(String[] args) throws Exception {
            String path = "/zk-book";
            zk = new ZooKeeper("127.0.0.1:2181", 5000, 
                    new GetData_API_Sync_Usage());
            connectedSemaphore.await();
            zk.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            System.out.println("success create znode: " + path);
            
            System.out.println("the data of znode " + path + " is : " + new String(zk.getData(path, true, stat)));
            System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion());
            
            zk.setData(path, "123".getBytes(), -1);
    
            Thread.sleep(Integer.MAX_VALUE);
        }
    
        public void process(WatchedEvent event) {
            if (KeeperState.SyncConnected == event.getState()) {
                if (EventType.None == event.getType() && null == event.getPath()) {
                    connectedSemaphore.countDown();
                } else if (event.getType() == EventType.NodeDataChanged) {
                    try {
                        System.out.println("the data of znode " + event.getPath() + " is : " + new String(zk.getData(event.getPath(), true, stat)));
                        System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion());
                    } catch (Exception e) {
                    }
                }
            }
        }
    }

运行结果如下

success create znode: /zk-book
    the data of /zk-book is : 123
    czxID: 2924, mzxID: 2924, version: 0
    the data of /zk-book is : 123
    czxID: 2924, mzxID: 2925, version: 1

结果表明可以使用getData函数获取节点的数据。

**7、更新数据（乐观锁）**

在更新数据时，setData方法存在一个version参数，其用于指定节点的数据版本，表明本次更新操作是针对指定的数据版本进行的，但是，在getData方法中，并没有提供根据指定数据版本来获取数据的接口，那么，这里为何要指定数据更新版本呢，这里方便理解，可以等效于CAS（compare and swap），对于值V，每次更新之前都会比较其值是否是预期值A，只有符合预期，才会将V原子化地更新到新值B。Zookeeper的setData接口中的version参数可以对应预期值，表明是针对哪个数据版本进行更新，假如一个客户端试图进行更新操作，它会携带上次获取到的version值进行更新，而如果这段时间内，Zookeeper服务器上该节点的数据已经被其他客户端更新，那么其数据版本也会相应更新，而客户端携带的version将无法匹配，无法更新成功，因此**可以有效地避免分布式更新的并发问题**。

public class SetData_API_Sync_Usage implements Watcher {
        private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
        private static ZooKeeper zk;
    
        public static void main(String[] args) throws Exception {
            String path = "/zk-book";
            zk = new ZooKeeper("127.0.0.1:2181", 5000, new SetData_API_Sync_Usage());
            connectedSemaphore.await();
    
            zk.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            System.out.println("success create znode: " + path);
            zk.getData(path, true, null);
    
            Stat stat = zk.setData(path, "456".getBytes(), -1);
            System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion());
            Stat stat2 = zk.setData(path, "456".getBytes(), stat.getVersion());
            System.out.println("czxID: " + stat2.getCzxid() + ", mzxID: " + stat2.getMzxid() + ", version: " + stat2.getVersion());
            try {
                zk.setData(path, "456".getBytes(), stat.getVersion());
            } catch (KeeperException e) {
                System.out.println("Error: " + e.code() + "," + e.getMessage());
            }
            Thread.sleep(Integer.MAX_VALUE);
        }
    
        public void process(WatchedEvent event) {
            if (KeeperState.SyncConnected == event.getState()) {
                if (EventType.None == event.getType() && null == event.getPath()) {
                    connectedSemaphore.countDown();
                }
            }
        }
    }

运行结果如下

success create znode: /zk-book
    czxID: 2936, mzxID: 2937, version: 1
    czxID: 2936, mzxID: 2938, version: 2
    Error: BADVERSION,KeeperErrorCode = BadVersion for /zk-book

结果表明由于携带的数据版本不正确，而无法成功更新节点。其中，setData中的version参数设置-1含义为客户端需要基于数据的最新版本进行更新操作。

**8、检测节点是否存在**

在调用接口时注册Watcher的话，还可以对节点是否存在进行监听，一旦节点被创建、被删除、数据更新，都会通知客户端。

public class Exist_API_Sync_Usage implements Watcher {
        private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
        private static ZooKeeper zk;
    
        public static void main(String[] args) throws Exception {
            String path = "/zk-book";
            zk = new ZooKeeper("127.0.0.1:2181", 5000, //
                    new Exist_API_Sync_Usage());
            connectedSemaphore.await();
    
            zk.exists(path, true);
    
            zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            zk.setData(path, "123".getBytes(), -1);
    
            zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println("success create znode: " + path + "/c1");
    
            zk.delete(path + "/c1", -1);
            zk.delete(path, -1);
    
            Thread.sleep(Integer.MAX_VALUE);
        }
    
        public void process(WatchedEvent event) {
            try {
                if (KeeperState.SyncConnected == event.getState()) {
                    if (EventType.None == event.getType() && null == event.getPath()) {
                        connectedSemaphore.countDown();
                    } else if (EventType.NodeCreated == event.getType()) {
                        System.out.println("success create znode: " + event.getPath());
                        zk.exists(event.getPath(), true);
                    } else if (EventType.NodeDeleted == event.getType()) {
                        System.out.println("success delete znode: " + event.getPath());
                        zk.exists(event.getPath(), true);
                    } else if (EventType.NodeDataChanged == event.getType()) {
                        System.out.println("data changed of znode: " + event.getPath());
                        zk.exists(event.getPath(), true);
                    }
                }
            } catch (Exception e) {
            }
        }
    }

运行结果如下

success create znode: /zk-book
    data changed of znode: /zk-book
    success create znode: /zk-book/c1
    success delete znode: /zk-book

**结果表明：**

*  无论节点是否存在，都可以通过exists接口注册Watcher。
 *  注册的Watcher，对节点创建、删除、数据更新事件进行监听。
 *  对于指定节点的子节点的各种变化，不会通知客户端。

参考：[https://www.cnblogs.com/leesf456/p/6028416.html][https_www.cnblogs.com_leesf456_p_6028416.html]

[20190422214540904.png]: /images/20220216/6b05caef1ebd42c3a37275dc999b26af.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70]: /images/20220216/c7e0161e8fb84bf7a5b7bd2082cb6e9e.png
[20190422215357475.png]: /images/20220216/646fbb7813ce49dd97975c80d0f9840b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70 1]: /images/20220216/afe4b882ceed4565bfaf4793b1c2d2f5.png
[https_www.cnblogs.com_leesf456_p_6028416.html]: https://www.cnblogs.com/leesf456/p/6028416.html