Apache ZooKeeper - 使用Apache Curator操作ZK

小小工匠

发布于 2021-08-17 16:05:40

1.9K0

文章被收录于专栏：小工匠聊架构小工匠聊架构

原生ZK API VS Curator

Apache ZooKeeper - 使用原生的API操作ZK

ZooKeeper原生Java API的不足之处：

连接zk超时时，不支持自动重连，需要手动操作
Watch注册一次就会失效，需手工反复注册
不支持递归创建节点
异步支持，没有线程池
…

Apache curator：

解决Watch注册一次就会失效的问题
API 更加简单易用、封装了常用的ZooKeeper工具类
使用Curator实现比如分布式锁等需求更简单
异步执行，支持自定义线程池
…

Curator是netflix公司开源的一套zookeeper客户端，Apache的顶级项目

与Zookeeper提供的原生客户端相比，Curator的抽象层次更高，简化了Zookeeper客户端的开发量

Curator解决了很多zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册wathcer和NodeExistsException 异常等

Curator 概述

Apache Curator : https://curator.apache.org/

看看模块

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

Maven依赖

     <dependency>
            <groupId>org.apache.curatorgroupId>
            <artifactId>curator-recipesartifactId>
            <version>5.0.0version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.zookeepergroupId>
                    <artifactId>zookeeperartifactId>
                exclusion>
            exclusions>
        dependency>
        <dependency>
            <groupId>org.apache.curatorgroupId>
            <artifactId>curator-x-discoveryartifactId>
            <version>5.0.0version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.zookeepergroupId>
                    <artifactId>zookeeperartifactId>
                exclusion>
            exclusions>
        dependency>

会话创建

和客户端/ 服务器交互，第一步就要创建会话

Curator 提供了多种方式创建会话

静态工厂方式创建会话

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(5000, 30);
        CuratorFramework client = CuratorFrameworkFactory.newClient(getConnectStr(), retryPolicy);
        client .start();

使用 fluent 风格创建会话

   RetryPolicy retryPolicy = new ExponentialBackoffRetry(5000, 30);
   curatorFramework = CuratorFrameworkFactory.builder().connectString(getConnectStr())
           .retryPolicy(retryPolicy)
           .sessionTimeoutMs(sessionTimeoutMs)
           .connectionTimeoutMs(connectionTimeoutMs)
           .canBeReadOnly(true)
           .build();
 curatorFramework.start();

上述代码采用了流式方式，最核心的类是 CuratorFramework 类，该类的作用是定义一个 ZooKeeper 客户端对象，并在之后的上下文中使用。

在定义 CuratorFramework 对象实例的时候，使用了 CuratorFrameworkFactory 工厂方法，并指定了 connectionString服务器地址列表、retryPolicy 重试策略、sessionTimeoutMs 会话超时时间、connectionTimeoutMs 会话创建超时时间。

connectionString：服务器地址列表，在指定服务器地址列表的时候可以是一个地址，也可以是多个地址。如果是多个地址，那么每个服务器地址列表用逗号分隔, 如 host1:port1,host2:port2,host3:port3
retryPolicy：重试策略，当客户端异常退出或者与服务端失去连接的时候，可以通过设置客户端重新连接 ZooKeeper 服务端。而 Curator 提供了一次重试、多次重试等不同种类的实现方式。在 Curator 内部，可以通过判断服务器返回的 keeperException 的状态代码来判断是否进行重试处理，如果返回的是 OK 表示一切操作都没有问题，而 SYSTEMERROR 表示系统或服务端错误

策略名称	描述
ExponentialBackoffRetry	重试一组次数，重试之间的睡眠时间增加
RetryNTimes	重试最大次数
RetryOneTime	只重试一次
RetryUntilElapsed	在给定的时间结束之前重试

sessionTimeoutMs 超时时间：Curator 客户端创建过程中，有两个超时时间的设置。一个是 sessionTimeoutMs会话超时时间，用来设置该条会话在 ZooKeeper 服务端的失效时间。
另一个是 connectionTimeoutMs 客户端创建会话的超时时间，用来限制客户端发起一个会话连接到接收ZooKeeper 服务端应答的时间。sessionTimeoutMs 作用在服务端，而connectionTimeoutMs 作用在客户端。

创建节点

    /**
     * 递归创建子节点
     */
    @SneakyThrows
    @Test
    public void testCreateWithParent()  {
        CuratorFramework curatorFramework = getCuratorFramework();

        String pathWithParent = "/artisan-node/artisan-node-sub1/artisan-node-sub1-1";
        String path = curatorFramework.create().creatingParentsIfNeeded().forPath(pathWithParent);
        log.info("curator create node :{}  successfully.", path);
    }

使用 create 函数创建数据节点，并通过 withMode 函数指定节点类型持久化节点，临时节点，顺序节点，临时顺序节点，持久化顺序节点等），默认是持久化节点，之后调用 forPath 函数来指定节点的路径和数据信息

protection 模式 ,规避僵尸节点

  /**
     * protection 模式，防止由于异常原因，导致僵尸节点
     * @throws Exception
     */
    @SneakyThrows
    @Test
    public void testCreate()  {

        CuratorFramework curatorFramework = getCuratorFramework();
        String forPath = curatorFramework
                .create()
                .withProtection()  // 防止僵尸节点
                .withMode(CreateMode.EPHEMERAL_SEQUENTIAL).
                        forPath("/curator-node", "data".getBytes());
        log.info("curator create node :{}  successfully.", forPath);
    }

看下zk中的数据

实现原理后面单独开篇解读，总体思想就是随机生成一个UUID，再创建之前客户端根据这个缓存的UUID去看ZK Server是否存在，存在则认为是成功的，否则就继续创建。

获取数据

 @Test
    public void testGetData() throws Exception {
        CuratorFramework curatorFramework = getCuratorFramework();

        byte[] bytes = curatorFramework.getData().forPath("/curator-node");
        log.info("get data from  node :{}  successfully.", new String(bytes));
    }

通过客户端实例的getData() 方法更新 ZooKeeper 服务上的数据节点，在getData 方法的后边，通过 forPath 函数来指定查询的节点名称

修改数据

@Test
    public void testSetData() throws Exception {
        CuratorFramework curatorFramework = getCuratorFramework();

        curatorFramework.setData().forPath("/curator-node", "changed!".getBytes());
        byte[] bytes = curatorFramework.getData().forPath("/curator-node");
        log.info("get data from  node /curator-node :{}  successfully.", new String(bytes));
    }

通过客户端实例的 setData() 方法更新 ZooKeeper 服务上的数据节点，在setData 方法的后边，通过 forPath 函数来指定更新的数据节点路径以及要更新的数据。

删除数据 guaranteed()

  @Test
    public void testDelete() throws Exception {
        CuratorFramework curatorFramework = getCuratorFramework();

        String pathWithParent = "/node-parent";
        curatorFramework.delete().guaranteed().deletingChildrenIfNeeded().forPath(pathWithParent);
    }

guaranteed：主要起到一个保障删除成功的作用，只要该客户端的会话有效，就会在后台持续发起删除请求，直到该数据节点在ZooKeeper 服务端被删除。

deletingChildrenIfNeeded：指定了该函数后，系统在删除该数据节点的时候会以递归的方式直接删除其子节点，以及子节点的子节点。

获取子节点

     @Test
    public void testListChildren() throws Exception {
        CuratorFramework curatorFramework = getCuratorFramework();

        String pathWithParent = "/artisan-node";
        List<String> list = curatorFramework.getChildren().forPath(pathWithParent);
        list.forEach(System.out::println); 
    }

通过客户端实例的 getChildren() 方法更新 ZooKeeper 服务上的数据节点，在getChildren方法的后边，通过 forPath 函数来指定节点下的一级子节点的名称

异步线程池

Curator 使用BackgroundCallback 接口处理服务器端返回来的信息。

如果在异步线程中调用，默认在 EventThread 线程中调用，支持自定义线程池

 /**
     * 使用默认的 EventThread异步线程处理
     * @throws Exception
     */
    @Test
    public void testThreadPoolByDefaultEventThread() throws Exception {

        CuratorFramework curatorFramework = getCuratorFramework();
        String ZK_NODE="/artisan-node";
        curatorFramework.getData().inBackground((client, event) -> {
            log.info(" background: {}", new String(event.getData()));
        }).forPath(ZK_NODE);;

    }

    /**
     * 使用自定义线程池
     * @throws Exception
     */
    @Test
    public void testThreadPoolByCustomThreadPool() throws Exception {

        CuratorFramework curatorFramework = getCuratorFramework();
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        String ZK_NODE="/artisan-node";
        curatorFramework.getData().inBackground((client, event) -> {
            log.info(" background: {}", new String(event.getData()));
        },executorService).forPath(ZK_NODE);

     }