【基础篇】详解Zookeeper客户端Curator

清疚 2022-12-20 06:03 108阅读 0赞

### 一、前言 ###

Zookeeper被广泛应用于分布式环境下各种应用程序的协调，而Curator无疑是Zookeeper客户端中的瑞士军刀，解决了很多Zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册Watcher和NodeExistsException异常等等。

### 二、依赖 ###

Curator包含了几个包：  
**curator-framework**：对zookeeper的底层api的一些封装  
**curator-client**：提供一些客户端的操作，例如重试策略等  
**curator-recipes**：封装了一些高级特性，如：Cache事件监听、选举、分布式锁、分布式计数器、分布式Barrier等

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

### 三、基本Api ###

#### 3.1 会话创建 ####

1.使用静态工厂方法创建会话

CuratorFramework zkClient = CuratorFrameworkFactory.newClient(zkAddr, 5000, 3000, new ExponentialBackoffRetry(1000, 5));

构造函数主要有4个参数：

<table> 
 <thead> 
  <tr> 
   <th>参数名</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>connectionString</td> 
   <td>服务器列表，格式host1:port1,host2:port2,…</td> 
  </tr> 
  <tr> 
   <td>retryPolicy</td> 
   <td>重试策略</td> 
  </tr> 
  <tr> 
   <td>sessionTimeoutMs</td> 
   <td>会话超时时间，单位毫秒</td> 
  </tr> 
  <tr> 
   <td>connectionTimeoutMs</td> 
   <td>连接创建超时时间，单位毫秒</td> 
  </tr> 
 </tbody> 
</table>

2.使用Fluent风格创建会话

CuratorFramework zkClient = CuratorFrameworkFactory.builder().connectString(zkAddr)
            .sessionTimeoutMs(5000)
            .connectionTimeoutMs(3000)
            .retryPolicy(new ExponentialBackoffRetry(1000, 5))
            .build();

3.创建包含隔离命名空间的会话  
为了实现不同的Zookeeper业务之间的隔离，需要为每个业务分配一个独立的命名空间（NameSpace），即指定一个Zookeeper的根路径，例如（下面的例子）当客户端指定了独立命名空间为“/config”，那么该客户端对Zookeeper上的数据节点的操作都是基于该目录进行的。在多个应用共用一个Zookeeper集群的场景下，这对于实现不同应用之间的相互隔离十分有意义。

作者：zhrowable  
链接：https://www.jianshu.com/p/70151fc0ef5d  
来源：简书  
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

#### 3.2 启动客户端 ####

当创建会话成功，得到实例然后可以直接调用其start( )方法启动客户端。

zkClient.start();

#### 3.3 数据节点操作 ####

##### 3.3.1 创建数据节点 #####

Zookeeper的节点模式总共有三种，持久节点，临时节点，顺序节点，通过组合我们可以实现以下4种节点类型：

> PERSISTENT：持久节点  
> PERSISTENT\_SEQUENTIAL：持久顺序节点  
> EPHEMERAL：临时节点  
> EPHEMERAL\_SEQUENTIAL：临时顺序节点

// 创建一个节点，初始内容为空
    // 如果没有设置节点属性，节点创建模式默认为持久化节点，内容默认为空
    zkClient.create().forPath("test");
    
    // 创建一个节点，附带初始化内容
    zkClient.create().forPath("test", "contect".getBytes());
    
    // 创建一个节点，指定创建模式（临时节点），内容为空
    zkClient.create().withMode(CreateMode.EPHEMERAL).forPath("test");
    
    // 创建一个节点，指定创建模式（临时节点），附带初始化内容
    zkClient.create().withMode(CreateMode.EPHEMERAL).forPath("test", "contect".getBytes());
    
    // 创建一个节点，指定创建模式（临时节点），附带初始化内容，并且自动递归创建父节点
    zkClient.create().creatingParentContainersIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("test", "contect".getBytes());

自动递归创建父节点非常有用，一般情况创建一个子节点必须先判断它的父节点是否存在，如果不存在直接创建会抛出NoNodeException，使用creatingParentContainersIfNeeded()能够自动递归创建所有所需的父节点。

##### 3.3.2 删除数据节点 #####

// 删除一个节点
    // 此方法只能删除叶子节点，否则会抛出异常。
    zkClient.delete().forPath("test");
    
    // 删除一个节点，并且递归删除其所有的子节点
    zkClient.delete().deletingChildrenIfNeeded().forPath("test");
    
    // 删除一个节点，强制指定版本进行删除
    zkClient.delete().withVersion(1000).forPath("test");
    
    // 删除一个节点，强制保证删除
    // guaranteed()是一个保障措施，只要客户端会话有效，会在后台持续进行删除操作，直到成功。
    zkClient.delete().guaranteed().forPath("test");
    
    // 上面的多个流式接口是可以自由组合的，例如：
    zkClient.delete().guaranteed().deletingChildrenIfNeeded().withVersion(1000).forPath("test");

##### 3.3.3 读取数据节点 #####

// 读取一个节点的数据内容
    zkClient.getData().forPath("test");
    
    // 读取一个节点的数据内容，同时获取到该节点的stat
    Stat stat = new Stat();
    zkClient.getData().storingStatIn(stat).forPath("test");
    
    //获取某个节点的所有子节点路径
    zkClient.getChildren().forPath("test");

通过传递Stat可以获取到读取的节点状态信息（例如版本号），zookeeper内部根据版本号区别当前的更新是不是最新，所以带版本号更新可避免并行操作带来的数据不一致问题。

##### 3.3.3 更新数据节点 #####

// 更新一个节点的数据内容
    // 该接口会返回一个Stat实例
    client.setData().forPath("test", "contect".getBytes());
    
    // 更新一个节点的数据内容，强制指定版本进行更新
    client.setData().withVersion(1000 ).forPath("test", "contect".getBytes());
    
    // 检查节点是否存在
    client.checkExists().forPath("test");

#### 3.4 事务 ####

CuratorFramework的实例包含inTransaction( )接口方法，调用此方法开启一个ZooKeeper事务。可以复合create, setData, check, and/or delete 等操作然后调用commit()作为一个原子操作提交。例如：

zkClient.inTransaction().check().forPath("test")
            .and().create().withMode(CreateMode.PERSISTENT).forPath("test", "contect".getBytes())
            .and().setData().forPath("test", "newContect".getBytes())
            .and().commit();

#### 3.5 异步接口 ####

上面提到的创建、删除、更新、读取等方法都是同步的，Curator提供异步接口，引入了BackgroundCallback接口用于处理异步接口调用之后服务端返回的结果信息。BackgroundCallback接口中一个重要的回调值为CuratorEvent，里面包含事件类型、响应吗和节点的详细信息。

ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 5, 300, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(10000), new ThreadFactory() { 
        @Override
        public Thread newThread(Runnable r) { 
            return new Thread(r, "zkSyncThreadPool");
        }
    });
    zkClient.create()
            .creatingParentsIfNeeded()
            .withMode(CreateMode.EPHEMERAL)
            .inBackground((curatorFramework, curatorEvent) -> { 
                System.out.println(String.format("eventType:%s,resultCode:%s",curatorEvent.getType(),curatorEvent.getResultCode()));
            }, executor)
            .forPath("test");

如果inBackground()方法不指定executor，那么会默认使用Curator的EventThread去进行异步处理。

**相关链接：**

*  [【高级篇】详解Zookeeper客户端Curator][Zookeeper_Curator]
 *  [【进阶篇】详解Zookeeper客户端Curator][Zookeeper_Curator 1]
 *  [《跟着实例学习ZooKeeper的用法》博客系列][ZooKeeper]

[Zookeeper_Curator]: https://blog.csdn.net/lveex/article/details/109636635
[Zookeeper_Curator 1]: https://blog.csdn.net/lveex/article/details/111193697
[ZooKeeper]: https://colobu.com/tags/Curator/