3.ZooKeeper客户端Curator的进阶使用「第三章 ZooKeeper Java客户端」「架构之路ZooKeeper理论和实战」

在前面我们对于Curator有了一个基本的了解和使用，这一节我们看一下更深入的知识。

一、Curator进阶使用

1.1 Curator构建CuratorFramework推荐方式

       在前面我们说到了，第二种方式会比较好，直接上源码：

  1. /*
  2.     方式二：
  3.      * connectionString zk地址
  4.      * sessionTimeoutMs 会话超时时间
  5.      * connectionTimeoutMs 连接超时时间
  6.      * namespace 每个curatorFramework 可以设置一个独立的命名空间,之后操作都是基于该命名空间，比如操作 /user/message 其实操作的是/curator/user/message
  7.      * retryPolicy 重试策略
  8.  */
  9. String connectString = "127.0.0.1:2181";
  10. RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
  11. CuratorFramework curatorFramework = CuratorFrameworkFactory.builder()
  12.         .connectString(connectString)
  13.         .sessionTimeoutMs(60000)
  14.         .connectionTimeoutMs(15000)
  15.         .namespace("curator")
  16.         .retryPolicy(retryPolicy)
  17.         .build();
  18. curatorFramework.start();

1.2 连接成功的监听

       我们创建连接是否成功可以通过方法进行监听：

  1. curatorFramework.getConnectionStateListenable().addListener(new ConnectionStateListener() {
  2.     @Override
  3.     public void stateChanged(CuratorFramework client, ConnectionState newState) {
  4.         if(newState == ConnectionState.CONNECTED){
  5.             System.out.println("连接成功！");
  6.         }
  7.     }
  8. });

       添加监听的代码需要放到curatorFramework.start();代码之前。

1.3 创建节点

       上面使用create进行节点的创建，但是如果节点存在的话，也是会报错的：

  1.     创建一个 允许所有人访问的 持久节点
  2.     由于使用了namespace，所以最终创建的节点是：/curator/test2
  3.  */
  4. curatorFramework.create()
  5.         .creatingParentsIfNeeded()//递归创建,如果没有父节点,自动创建父节点
  6.         .withMode(CreateMode.PERSISTENT)//节点类型,持久节点
  7.         .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)//设置ACL,和原生API相同
  8.         .forPath("/test2", "hello".getBytes());

说明：由于使用了namespace，所以最终创建的节点是：/curator/test2

 

       假设我们多次创建节点的话，是会保存的，那么怎么办呐？

       可以先判断节点是否存在，在决定是否进行创建：

  1. /*
  2.     创建一个 允许所有人访问的 持久节点
  3.     由于使用了namespace，所以最终创建的节点是：/curator/test2
  4.  */
  5. String path = "/test3";
  6. Stat stat = curatorFramework.checkExists().forPath(path);
  7. if(stat == null){
  8.     curatorFramework.create()
  9.             .creatingParentsIfNeeded()//递归创建,如果没有父节点,自动创建父节点
  10.             .withMode(CreateMode.PERSISTENT)//节点类型,持久节点
  11.             .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)//设置ACL,和原生API相同
  12.             .forPath(path, "hello".getBytes());
  13. }

       protection模式，防止由于异常原因，导致僵尸节点：

  1. curatorFramework.create()
  2.         .creatingParentsIfNeeded()//递归创建,如果没有父节点,自动创建父节点
  3.         .withProtection()//protection模式，防止由于异常原因，导致僵尸节点。
  4.         .withMode(CreateMode.PERSISTENT_SEQUENTIAL)//节点类型,持久节点
  5.         .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)//设置ACL,和原生API相同
  6.         .forPath("/test4/id-", "hello".getBytes());

       当我们创建节点是持久有序节点的时候，当我们提供请求给ZK Server，ZK Server创建成功了之后会给客户端进行反馈，但在反馈的过程中，由于网络原因导致客户端没有收到反馈结果，由于curator有重试的机制，会造次进行发起请求进行创建，那么就会出现之前的节点没有被使用的情况，也就是所谓的僵尸节点。如果我们配置了withProtection()的话，那么就能够解决僵尸节点的问题。

       如果客户端所连接的服务器崩溃了，但还没来得及返回客户端所创建的有序节点的节点名称（即节点序列号），或者客户端只是连接丢失，客户端没接收到所请求操作的响应信息，结果，客户端并不知道所创建的znode节点路径名称。回忆对于有序节点的应用场景，例如，建立一个有序的所有客户端列表。为了解决这个问题，CreateBuilder提供了一个 withProtection 方法来通知Curator客户端，在创建的有序节点前添加一个唯一标识符，如果create操作失败了，客户端就会开始重试操作，而重试操作的一个步骤就是验证是否存在一个节点包含这个唯一标识符。

       那么具体是什么原理呢？

       我看下底层的代码CreateBuilderImpl：

       跟进setProtectdMode方法：

       说白了调用withProtection方法就是初始化了一个protectedId。那这个家伙又是怎么使用的呐？

       我们看下生成的数据：

       这个protectedId被当做了节点path的前缀了。

       通过这里我们就可以分析出来，底层应该是每次都会把protectedId进行拼凑成一个新的path，重试操作的一个步骤就是验证是否存在一个节点包含这个唯一标识符。。






1.4 异步获取节点数据

       获取节点可以使用get进行获取，如果是异步获取的，使用inBackground：

  1. curatorFramework.getData().inBackground(new BackgroundCallback(){
  2.     @Override
  3.     public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
  4.         System.out.println(event.getPath()+":"+new String(event.getData()));
  5.     }
  6. }).forPath("/test3");

       对于inBackground还可以指定我们自己的线程池对象：

public T inBackground(BackgroundCallback callback, Executor executor);

       这个大家可以去尝试一下。

1.5 修改节点

       修改节点就很简单了：

curatorFramework.setData().forPath("/test3", "hello-update".getBytes());

        根据版本进行修改的话：

  1. Stat stat = curatorFramework.checkExists().forPath("/test3");
  2. curatorFramework.setData().withVersion(stat.getVersion()).forPath("/test3", "hello-update1".getBytes());
  3.

 

1.6 删除节点

       删除节点使用delete().forPath()：

  1. /*
  2.     会删除节点：
  3.         /curator/test4
  4.     以及
  5.         /curator/test4下的子节点
  6.  */
  7. curatorFramework.delete().deletingChildrenIfNeeded().forPath("/test4");

 



购买完整视频，请前往：http://www.mark-to-win.com/TeacherV2.html?id=287