Curator客户端框架

蔚落 2022-04-24 10:32 268阅读 0赞

# 一 简介 #

使用过原生API不得不说,有很多的问题,比如：不能递归创建和删除节点、Watcher只能使用一次、还有很多可以解决分布式应用问题的api(比如分布式锁,leader选举等)，但由于ZooKeeper提供的原始API并不是很易用，在其基础上封装一些高级应用又是一件很复杂的事情。

这个时候，Curator出现了，Curator是Netflix公司开源的一个Zookeeper客户端，后捐献给Apache，Curator框架在zookeeper原生API接口上进行了包装，解决了很多ZooKeeper客户端非常底层的细节开发。提供ZooKeeper各种应用场景(recipe， 比如：分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等)的抽象封装，实现了Fluent风格的API接口,是最好用，最流行的zookeeper的客户端。

Apache Curator是一个比较完善的ZooKeeper客户端框架，通过封装的一套高级API 简化了ZooKeeper的操作。通过查看官方文档，可以发现Curator主要解决了三类问题：

*  封装ZooKeeper client与ZooKeeper server之间的连接处理
 *  提供了一套Fluent风格的操作API
 *  提供ZooKeeper各种应用场景(recipe， 比如：**分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列**等)的抽象封装

## Curator主要从以下几个方面降低了zk使用的复杂性： ##

*  **重试机制:**提供可插拔的重试机制, 它将给捕获所有可恢复的异常配置一个重试策略，并且内部也提供了几种标准的重试策略(比如指数补偿)
 *  **连接状态监控:** Curator初始化之后会一直对zk连接进行监听，一旦发现连接状态发生变化将会作出相应的处理
 *  **zk客户端实例管理:**Curator会对zk客户端到server集群的连接进行管理，并在需要的时候重建zk实例，保证与zk集群连接的可靠性
 *  **各种使用场景支持:**Curator实现了zk支持的大部分使用场景（甚至包括zk自身不支持的场景），这些实现都遵循了zk的最佳实践，并考虑了各种极端情况

# 二 基于Curator的ZooKeeper基本用法 #

首先需要在pom.xml中添加如下依赖：

<dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-framework</artifactId>
                <version>4.0.0</version>
            </dependency>
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-recipes</artifactId>
                <version>4.0.0</version>
            </dependency>
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-x-discovery</artifactId>
                <version>4.0.0</version>
            </dependency>
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-test</artifactId>
                <version>4.0.0</version>
                <scope>test</scope>
            </dependency>

### **1、创建会话** ###

Curator是Fluent风格API，如果您以前没有使用它，它可能看起来很奇怪，所以建议您熟悉这种风格。

Curator的创建会话方式与原生的API创建方式区别很大。Curator创建客户端为CuratorFramework，是由CuratorFrameworkFactory工厂类来实现的，CuratorFramework是线程安全的，要连接的每个ZooKeeper集群只需要一个 CuratorFramework对象就可以了。

CuratorFrameworkFactory提供了两种构造CuratorFramework的方法,一种是直接newClient，还有一种方式是使用CuratorFrameworkFactory.builder()，提供了细粒度的控制。

CuratorFrameworkFactory提供了两个默认的直接构造CuratorFramework的方法,如下：

public static CuratorFramework newClient(String connectString, RetryPolicy retryPolicy);
    public static CuratorFramework newClient(String connectString, int sessionTimeoutMs, 
                             int connectionTimeoutMs, RetryPolicy retryPolicy);

其中connectString参数就是zk的地址，参数RetryPolicy提供重试策略的接口，可以让用户实现自定义的重试策略。

*  参数1：connectString，zookeeper服务器地址及端口号，多个zookeeper服务器地址以“,”分隔。
 *  参数2：sessionTimeoutMs，会话超时时间，单位毫秒，默认为60000ms。
 *  参数3：connectionTimeoutMs，连接超时时间，单位毫秒，默认为15000ms。
 *  参数4：retryPolicy，重试连接策略，有四种实现，分别为：ExponentialBackoffRetry（重试指定的次数, 且每一次重试之间停顿的时间逐渐增加）、RetryNtimes（指定最大重试次数的重试策略）、RetryOneTimes（仅重试一次）、RetryUntilElapsed（一直重试直到达到规定的时间）

使用上面方法创建出一个CuratorFramework之后，需要再调用其start()方法完成会话创建。

下面给出创建CuratorFramework例子代码：

/**
     * @description: 创建会话的例子
     */
    public class CreateClientExamples{
    
        public static void main(String[] args) {
    
            String connectionString = "127.0.0.1";
            ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3,Integer.MAX_VALUE);
    
            
            CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(connectionString, retryPolicy);
            curatorFramework.start();
    
            
            CuratorFramework curatorFramework1 = CuratorFrameworkFactory.newClient(connectionString, 60000, 15000, retryPolicy);
            curatorFramework1.start();
    
            /**
             * connectionString zk地址
             * sessionTimeoutMs 会话超时时间
             * connectionTimeoutMs 连接超时时间
             * namespace 每个curatorFramework 可以设置一个独立的命名空间,之后操作都是基于该命名空间，比如操作 /app1/message 其实操作的是/node1/app1/message
             * retryPolicy 重试策略
             */
            CuratorFramework curatorFramework2 = CuratorFrameworkFactory.builder().connectString(connectionString)
                    .sessionTimeoutMs(60000)
                    .connectionTimeoutMs(15000)
                    .namespace("/node1")
                    .retryPolicy(retryPolicy)
                    .build();
            curatorFramework2.start();
        }
    }

### **2、增删改查** ###

/**
     * PERSISTENT：持久化
       PERSISTENT_SEQUENTIAL：持久化并且带序列号
       EPHEMERAL：临时
       EPHEMERAL_SEQUENTIAL：临时并且带序列号
     */
    public class curatorRecipesDemo {
    
        final static String zookeeperAddress = "192.168.149.133:2181,192.168.149.135:2181,192.168.149.134:2181";
    
        public static void main(String[] args) throws Exception {
            CuratorFramework curatorClint =
                    CuratorFrameworkFactory.builder().
                            connectString(zookeeperAddress)//zkClint连接地址
                            .connectionTimeoutMs(2000)//连接超时时间
                            .sessionTimeoutMs(10000)//会话超时时间
                            .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                            //重试策略
                            .namespace("myZookeeperTest")
                            //命名空间,默认节点
                            .build();
    
            curatorClint.start();
    
            curatorClint.create().forPath("/path"); //默认持久化节点,以斜杠开头
            System.out.println(curatorClint.getChildren().forPath("/"));
    
            curatorClint.create().withMode(CreateMode.EPHEMERAL)
                    .forPath("/secondPath","hello,word".getBytes());
            System.out.println("节点secondPath的数据"+new String(curatorClint.getData().forPath("/secondPath")));
    
            curatorClint.setData().forPath("/secondPath","hello,myWorld!".getBytes());
            System.out.println("节点secondPath的数据"+new String(curatorClint.getData().forPath("/secondPath")));
    
            curatorClint.create()
                    .creatingParentContainersIfNeeded()
                    .forPath("/secondPath/second2/second3");//递归创建
    
            List<String> list= curatorClint.getChildren().forPath("/secondPath");//查询节点的所有字节点
            System.out.println(list);
    
            curatorClint.delete().deletingChildrenIfNeeded().forPath("/secondPath/second2");//递归删除
            System.out.println(curatorClint.checkExists().forPath("/secondPath/second2"));//判断节点是否存在
            System.out.println(curatorClint.checkExists().forPath("/secondPath/second2/second3"));//判断节点是否存在
            System.out.println(curatorClint.getChildren().forPath("/secondPath"));
            curatorClint.delete().deletingChildrenIfNeeded().forPath("/secondPath");
    
            //事务操作(curator独有)
            Collection<CuratorTransactionResult> commit = curatorClint.inTransaction().create().forPath("xiao","456".getBytes()).and().setData()
                            .forPath("xiao","123".getBytes()).and().commit();
            for (CuratorTransactionResult result : commit) {
    			System.out.println(result.getForPath()+"--->"+result.getType());
    		}
            //todo guaranteed()如果删除失败,会记录下来,只要会话有效,就会不断的重试,直到删除成功为止
           //todo   Stat stat 对象包含版本id,事物id等信息
    
        }
    
    }

### **3、监听** ###

### NodeCache ###

**用于监听数据节点本身的变化**。提供了两个构造方法：

public NodeCache(CuratorFramework client, String path)
    
    public NodeCache(CuratorFramework client, String path, boolean dataIsCompressed)

其中参数dataIsCompressed表示是否对数据进行压缩，而第一个方法内部实现为调用第二个方法，且dataIsCompressed默认设为false。

对节点的监听需要配合回调函数来进行处理接收到监听事件之后的业务处理。NodeCache通过NodeCacheListener来完成后续处理。具体代码示例如下：

public class CuratorNodeCacheTest {
    
        public static void main(String[] args) throws Exception {
    
            CuratorFramework client = getClient();
            String path = "/p1";
            final NodeCache nodeCache = new NodeCache(client,path);
            nodeCache.start();
            nodeCache.getListenable().addListener(new NodeCacheListener() {
                @Override
                public void nodeChanged() throws Exception {
                    System.out.println("监听事件触发");
                    System.out.println("重新获得节点内容为：" + new String(nodeCache.getCurrentData().getData()));
                }
            });
            client.setData().forPath(path,"456".getBytes());
            client.setData().forPath(path,"789".getBytes());
            client.setData().forPath(path,"123".getBytes());
            client.setData().forPath(path,"222".getBytes());
            client.setData().forPath(path,"333".getBytes());
            client.setData().forPath(path,"444".getBytes());
            Thread.sleep(15000);
    
        }
    
        private static CuratorFramework getClient(){
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
            CuratorFramework client = CuratorFrameworkFactory.builder()
                    .connectString("127.0.0.1:2181")
                    .retryPolicy(retryPolicy)
                    .sessionTimeoutMs(6000)
                    .connectionTimeoutMs(3000)
                    .namespace("demo")
                    .build();
            client.start();
            return client;
        }
    }

执行结果：

监听事件触发
    重新获得节点内容为：123
    监听事件触发
    重新获得节点内容为：333
    监听事件触发
    重新获得节点内容为：444

NodeCache的start方法有一个带Boolean参数的方法，如果设置为true则在首次启动时就会缓存节点内容到Cache中。

经过试验，发现注册监听之后，如果先后多次修改监听节点的内容，部分监听事件会发生丢失现象。其他版本未验证，此版本此处需特别留意。

NodeCache不仅可以监听节点内容变化，还可以监听指定节点是否存在。如果原本节点不存在，那么Cache就会在节点被创建时触发监听事件，如果该节点被删除，就无法再触发监听事件。

### PathChildrenCache ###

PathChildrenCache用于**监听数据节点子节点的变化情况**。当前版本总共提供了7个构造方法，其中2个已经不建议使用了。

public PathChildrenCache(CuratorFramework client, String path, boolean cacheData)
    
    public PathChildrenCache(CuratorFramework client, String path, boolean cacheData, boolean dataIsCompressed, final CloseableExecutorService executorService)
    
    public PathChildrenCache(CuratorFramework client, String path, boolean cacheData, boolean dataIsCompressed, final ExecutorService executorService)
    
    public PathChildrenCache(CuratorFramework client, String path, boolean cacheData, boolean dataIsCompressed, ThreadFactory threadFactory)
    
    public PathChildrenCache(CuratorFramework client, String path, boolean cacheData, ThreadFactory threadFactory)

常见的参数就不再具体说明了。其中cacheData表示是否把节点内容缓存起来，如果为true，那么接收到节点列表变更的同时会将获得节点内容。

ExecutorService 和threadFactory提供了通过线程池的方式来处理监听事件。

PathChildrenCache使用PathChildrenCacheListener（回调接口）来处理监听事件。具体使用方法见代码实例：

public class CuratorPathChildrenCacheTest {
    
        public static void main(String[] args) throws Exception {
    
            CuratorFramework client = getClient();
            String parentPath = "/p1";
    
            PathChildrenCache pathChildrenCache = new PathChildrenCache(client,parentPath,true);
    
        /* * StartMode：初始化方式 
         * POST_INITIALIZED_EVENT：异步初始化。初始化后会触发事件 
         * NORMAL：异步初始化 
         * BUILD_INITIAL_CACHE：同步初始化 
        * */
    
            pathChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
            pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                    System.out.println("事件类型："  + event.getType() + "；操作节点：" + event.getData().getPath());
                }
            });
    
            String path = "/p1/c1";
            client.create().withMode(CreateMode.PERSISTENT).forPath(path);
            Thread.sleep(1000); // 此处需留意，如果没有现成睡眠则无法触发监听事件
            client.delete().forPath(path);
    
            Thread.sleep(15000);
    
        }
    
        private static CuratorFramework getClient(){
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
            CuratorFramework client = CuratorFrameworkFactory.builder()
                    .connectString("127.0.0.1:2181")
                    .retryPolicy(retryPolicy)
                    .sessionTimeoutMs(6000)
                    .connectionTimeoutMs(3000)
                    .namespace("demo")
                    .build();
            client.start();
            return client;
        }
    }

打印结果为：

事件类型：CHILD_ADDED；操作节点：/p1/c1
    事件类型：CHILD_REMOVED；操作节点：/p1/c1

PathChildrenCache不会对二级子节点进行监听，只会对子节点进行监听。看上面的实例会发现在创建子节点和删除子节点两个操作中间使用了线程睡眠，否则无法接收到监听事件，这也是在使用过程中需要留意的一点。

### TreeCache ###

**既可以监听节点的状态，又可以监听子节点的状态**。类似于上面两种Cache的组合。

public class CuratorWatcher3 {
        private static final String CONNECT_ADDR = "192.168.3.58:2181,192.168.3.59:2181,192.168.3.66:2181";
        private static final int SESSION_TIMEOUT = 5000;
     
        public static void main(String[] args) throws Exception {
            RetryPolicy policy = new ExponentialBackoffRetry(1000, 10);
            CuratorFramework curator = CuratorFrameworkFactory.builder().connectString(CONNECT_ADDR).sessionTimeoutMs(SESSION_TIMEOUT)
                    .retryPolicy(policy).build();
            curator.start();
            TreeCache treeCache = new TreeCache(curator, "/treeCache");
            treeCache.start();
            treeCache.getListenable().addListener((curatorFramework, treeCacheEvent) -> {
                switch (treeCacheEvent.getType()) {
                    case NODE_ADDED:
                        System.out.println("NODE_ADDED：路径：" + treeCacheEvent.getData().getPath() + "，数据：" + new String(treeCacheEvent.getData().getData())
                        + "，状态：" + treeCacheEvent.getData().getStat());
                        break;
                    case NODE_UPDATED:
                        System.out.println("NODE_UPDATED：路径：" + treeCacheEvent.getData().getPath() + "，数据：" + new String(treeCacheEvent.getData().getData())
                                + "，状态：" + treeCacheEvent.getData().getStat());
                        break;
                    case NODE_REMOVED:
                        System.out.println("NODE_REMOVED：路径：" + treeCacheEvent.getData().getPath() + "，数据：" + new String(treeCacheEvent.getData().getData())
                                + "，状态：" + treeCacheEvent.getData().getStat());
                        break;
                    default:
                        break;
                }
            });
     
            curator.create().forPath("/treeCache", "123".getBytes());
            curator.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/treeCache/c1", "456".getBytes());
            curator.setData().forPath("/treeCache", "789".getBytes());
            curator.setData().forPath("/treeCache/c1", "910".getBytes());
            curator.delete().forPath("/treeCache/c1");
            curator.delete().forPath("/treeCache");
            Thread.sleep(5000);
            curator.close();
        }
    }

运行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70]: /images/20220216/2a6ece633fb64799a9c62100036ee520.png