Zookeeper--开源客户端curator常用API

Dear 丶 2021-08-28 18:24 471阅读 0赞

### Curator ###

Curator是 Netflix公司开源的一套 ZooKeeper客户端框架,作者是 Jordan Zimmerman和ZkClient一样, Curator解决了很多 ZooKeeper客户端非常底层的细节开发工作,包括连接重连、反复注册 Watcher和 NodeExistsException异常等,目前已经成为了Apache的顶级项目,是全世界范围内使用最广泛的 ZooKeeper客户端之一。

除了封装- - 些开发人员不需要特别关注的底层细节之外，Curator 还在ZooKeeper 原生API的基础上进行了包装，提供了一套易用性和可读性更强的Fluent风格的客户端API框架。

除此之外，Curator中还提供了ZooKeeper 各种应用场景(Recipe,如共享锁服务、Master选举机制和分布式计数器等)的抽象封装。

Maven依赖：

<dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-framework</artifactId>
                <version>2.4.2</version>
            </dependency>

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

工厂方法创建：

1.  使用CuratorFrameworkFactory这个工厂类的两个静态方法来创建一个客户端：
    
     *  `static CuratorFramework newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy);`
     *  `static CuratorFramework newClient(String connectString, RetryPolicy retryPolicy);`
2.  通过调用CuratorFramework中的start()方法来启动会话

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]

在重试策略上，Curator通过一个借口RetryPolicy来让用户实现自定义的重试策略。

public interface RetryPolicy { 
        boolean allowRetry(int retryCount, long elapsedTimeMs, RetrySleeper sleeper);
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

//使用Curator来创建一个Zookeeper客户端
    public class Create_Session_Sample { 
        public static void main(String[] args) throws InterruptedException { 
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
            CuratorFramework client =
                    CuratorFrameworkFactory.newClient("127.0.0.1:2181",
                            5000,
                            3000,
                            retryPolicy);
            client.start();
            Thread.sleep(Integer.MAX_VALUE);
        }
    }

首先创建了一个名为ExponentialBackoffRetry的重试策略，该重试策略是Curator默认提供的几种重试策略之一一， 其构造方法如下:

1.  `ExponentialBackoffRetry(int baseSleepTimeMs, int maxRetries);`
2.  `ExponentialBackoffRetry(int baseSleepTimeMs, int maxRetries, int maxSleepMs);`

![在这里插入图片描述][20200512104048769.png]

ExponentialBackoffRetry的重试策略如下：

给定一个初始sleep事件baseSleepTimeMs，在这个基础上结合重试次数，通过以下公式计算出当前需要sleep的时间：

当前sleep时间 = baseSleepTimeMs * Math.max(1, random.nextInt(1 << (retryCount + 1)))

可以看出，随着重试次数的增加，计算出的sleep时间会越来越大。如果该sleep时间在maxSleepMs的范围之内，那么就使用该sleep时间，否则使用maxSleepMs。另外，maxRetries参数控制了最大重试次数，以避免无限制的重试。

使用Fluent风格的API接口创建会话：

//使用Fluent风格API创建一个Zookeeper客户度
    public class Create_Session_Sample_Fluent { 
        public static void main(String[] args) throws InterruptedException { 
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
            CuratorFramework client = CuratorFrameworkFactory.builder()
                    .connectString("127.0.0.1:2181")
                    .sessionTimeoutMs(5000)
                    .retryPolicy(retryPolicy)
                    .build();
            client.start();
            Thread.sleep(Integer.MAX_VALUE);
        }
    }

使用Curator创建含隔离命名空间的会话：  
为了实现不同的ZooKeeper业务之间的隔离，往往会为每个业务分配–个独立的命名空间，即指定一个ZooKeeper根路径。例如，下面所示的代码片段中定义了某一个客户端的独立命名空间为/base,那么该客户端对ZooKeeper.上数据节点的任何操作，都是基于该相对目录进行的:

CuratorFrameworkFactoy.builder()
    	.connectString("127.0.0.1:2181")
    	.sessionTimeoutMs(5000)
    	.retryPolicy(retryPolicy)
    	.namespace("base")
    	.build();

#### 创建结点 ####

CuratorFramework
    	--public CreateBuilder create();
    CreateBuilder
    	--public ProtectACLCreateModePathAndBytesable<String>creatingParentsIfNeeded();
    CreateModable
    	--public T withMode(createMode mode);
    PathAndBytesable<T>
    	--public T forPath(String path, byte[] data) throws Exception;
    	--public T forPath(String path) thorws Exception;

创建一个节点，初始内容为空：

//注意，如果没有设置节点属性，那么Curator默认创建的是持久节点，内容默认是空。
    client.create().forPath("/zk-test");

![在这里插入图片描述][20200512111847226.png]

创建一个结点，附带初始内容：

//Curator按照Zookeeper原生API的风格，使用byte{]作为方法参数
    client.create().forPath("/zk-test1", "init".getBytes());

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]

创建一个临时结点，初始内容为空：

//创建一个临时结点，初始内容为空
           client.create().withMode(CreateMode.EPHEMERAL).forPath("/zk-ephemeral");

创建一个临时结点，并自动递归创建父节点：

//创建一个临时结点，并自动递归创建父节点
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
                    .forPath("/zk-parent/zk-son");

同时要注意的一点是，由于在ZooKeeper中规定了所有非叶子节点必须为持久节点，调用上面这个API之后，只有path参数对应的数据节点是临时节点，其父节点均为持久节点。  
![在这里插入图片描述][20200512112621782.png]

##### 删除结点 #####

CuratorFramework
    	--public DeleteBuilder delete();
    Versionable<T>
    	--public T withVersion(int version);
    DeleteBuilder
    	--public DeleteBuilderBase guaranteed();
    PathAndBytesable<T>
    	--public T forPath(String path, byte[] data) throws Exception;
    	--public T forPath(String path) thorws Exception;

删除一个结点：

//删除一个结点，只能删除叶子结点
    client.delete().forPath("/zk-parent");

删除一个结点，并递归删除其所有子节点：

//删除一个结点，并递归删除其所有子节点
            client.delete().deletingChildrenIfNeeded().forPath("/zk-test");

删除一个结点，强制指定版本进行删除：

//删除一个结点，强制指定版本进行删除
            client.delete().withVersion(0).forPath("/zk-test1");

删除一个结点，强制保证删除：

//删除一个结点，强制保证删除
            client.delete().guaranteed().forPath("/zk-delete");

注意，guaranteed()接口是一个保障措施，只要客户端会话有效，那么Curator会在后台持续进行删除操作，直到节点删除成功。

#### 读取数据 ####

CuratorFramework
    	--public GetDataBuilder getData();
    Statable<T>
    	--public T storingStatIn(Stat stat);
    Pathable<T>
    	--public T forPath(String path) throws Exception

读取一个结点的数据内容：

//返回值为byte[]
    client.getData().forPath(path);

读取一个结点的数据内容，同时获取到该结点的stat：

//通过传入一个旧的stat变量的方式来存储服务端返回的最新的结点状态信息
    client.getData().storingStatIn(stat).forPath(path);

示例：

public class Get_Data_Sample { 
    
        static String path = "/zk-book";
        static CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
    
        public static void main(String[] args) throws Exception { 
            client.start();
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(path, "init".getBytes());
            Stat stat = new Stat();
            System.out.println(new String(client.getData().storingStatIn(stat).forPath(path)));
        }
    }

#### 更新数据 ####

CuratorFramework
    	--public SetDataBuilder setData();
    Versionable<T>
    	--public T withVersion(int version);
    PathAndBytesable<T>
    	--public T forPath(String path, byte[] data) throws Exception;
    	-- public T forPath(String path) throws Exception;

更新一个结点的数据内容：

//返回stat对象
    client.setData().forPath(path);

更新一个结点的数据内容，强制指定版本进行更新：

//注意，withVersion接口就是用来实现CAS (Compare and Swap)的，version(版本信息)通常是从一个旧的stat对象中获取到的。
    client.setData().withVersion(version).forPath(path);

public class Set_Data_Sample { 
        static String path = "/zk-book";
        static CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
    
        public static void main(String[] args) throws Exception { 
            client.start();
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(path, "init".getBytes());
            Stat stat = new Stat();
            client.getData().storingStatIn(stat).forPath(path);
    
            System.out.println("Success set node for: " + path + ", new version:"
                + client.setData().withVersion(stat.getVersion()).forPath(path).getVersion());
            try { 
                client.setData().withVersion(stat.getVersion()).forPath(path);
            } catch (Exception e) { 
                System.out.println("Fail set node due to " + e.getMessage());
            }
        }
    }

Success set node for: /zk-book, new version:1
    Fail set node due to KeeperErrorCode = BadVersion for /zk-book

该程序前后进行了两次更新操作，第一次使用最新的stat变量进行更新操作，更新成功;第二次使用了过期的stat变量进行更新操作，抛出异常: KeeperErrorCode =BadVersion。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/0e0a2da81bbf4e1ca5dca26e3734bd05.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/8dc01645bfc844bc8c0bb79ac01c3b7b.png
[20200512104048769.png]: /images/20210813/599ded873753428d83355b48c8227fe3.png
[20200512111847226.png]: /images/20210813/6f5bc14630f74f0cafae29ebba325d8f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/2bbd3dc4cc4e4ce8abbc811a3d7f4b31.png
[20200512112621782.png]: /images/20210813/36f50a5fa66d42f29a50e57111062440.png