==============================

Curator Java 客户端 CRUD 使用

==============================

Curator 是 Apache 下的开源项目, Spring Cloud 也采用了该库, 可以其功能强大和权威性.

Curator 项目包含多个 artifact, 一般情况下, 我们只需要引入 curator-recipes 依赖即可, artifact 针对不同的场景提供高级封装, 可简化使用zk的复杂性.

Curator 4.0 适合于 zk 3.5版, 但目前zk3.5仍是beta版本, 对于zk 3.4版推荐使用 Curator 2.12.0 版.

 

Curator所做的改进：

摘自: https://iamjohnnyzhuang.github.io/architecture/2016/08/10/%E4%B8%BB%E5%A4%87%E5%88%87%E6%8D%A2%E7%9A%84%E6%80%9D%E8%80%83.html

重试机制:提供可插拔的重试机制, 它将给捕获所有可恢复的异常配置一个重试策略, 并且内部也提供了几种标准的重试策略(比如指数补偿).

连接状态监控: Curator初始化之后会一直的对zk连接进行监听, 一旦发现连接状态发生变化, 将作出相应的处理.

zk客户端实例管理:Curator对zk客户端到server集群连接进行管理. 并在需要的情况, 重建zk实例, 保证与zk集群的可靠连接

各种使用场景支持:Curator实现zk支持的大部分使用场景支持(甚至包括zk自身不支持的场景), 这些实现都遵循了zk的最佳实践, 并考虑了各种极端情况.

 

pom 依赖:

         
      
       org.apache.curator
          
      
       curator-recipes
          
      
       2.12.0
      
     
         
          
      
       org.slf4j
          
      
       slf4j-simple
          
      
       1.7.7
      
     

Curator大量使用了链式调用风格, 遇到forPath()才会真正触发zk调用, 主要的链式调用有:

   client.create().forPath()

   client.delete().forPath()

   client.setData().forPath()

   client.getData().forPath()

   client.checkExists().forPath()

下面 main() 代码中, 已经展现几乎所有的Curator 基础操作.

public static void main(String[] args) throws Exception {        // 多个服务器用逗号分隔        String zkUrl = "localhost:2181";        // 重连server机制, 最多重连3次, 每次重连间隔会加长, 初次重连的间隔为1000毫秒        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);        /**         * 重连机制有: ExponentialBackoffRetry: 重试指定的次数, 且每一次重试之间停顿的时间逐渐增加 RetryNtimes:         * 指定最大重试次数的重试策略 RetryOneTimes: 仅重试一次 RetryUntilElapsed:一直重试直到达到规定的时间         */        // CuratorFramework 是一个线程安全的类, 可以完成zk所有操作        CuratorFramework client = CuratorFrameworkFactory.newClient(zkUrl, retryPolicy);        // 打开客户端        client.start();        // 检查节点是否存在        if (client.checkExists().forPath("/head") != null) {            // 删除节点            client.delete().forPath("/head");        }        // 新建节点, 节点数据为空        client.create().forPath("/head", new byte[0]);        // 重新赋值,需要将String转成 byte 数组        client.setData().forPath("/head", "ABC".getBytes());        // 获取节点的版本等信息, 返回一个 Stat实例        Stat stat = client.checkExists().forPath("/head");        System.out.println(stat.getCzxid());        // 获取节点值, getData().forPath()结果为 byte 数组, 可以转成String类型        String value = new String(client.getData().forPath("/head"));        System.out.println(String.format("/head data is :%s", value));        // 创建临时序列节点        client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/head/child");        if (client.checkExists().forPath("/a") != null) {            // 级联删除节点            client.delete().guaranteed().deletingChildrenIfNeeded().forPath("/a");        }        // 级联创建节点, 创建之前不需要检查上级节点是否存在        client.create().creatingParentsIfNeeded().forPath("/a/b/c");        client.close();    }