ZooKeeper【基础 03】Java 客户端 Apache Curator 基础 API 使用举例(含源代码)

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介: ZooKeeper【基础 03】Java 客户端 Apache Curator 基础 API 使用举例(含源代码)

what

Curator(监护人;管理者) 是 Netflix 公司开源的一个 Zookeeper 客户端,目前由 Apache 进行维护。与 Zookeeper 原生客户端相比,Curator 的抽象层次更高,功能也更加丰富,是目前 Zookeeper 使用范围最广的 Java客户端。

use

  1. 依赖
<dependencies>
  <!--Curator 相关依赖-->
  <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.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.13</version>
  </dependency>
  <!--单元测试相关依赖-->
  <dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.12</version>
  </dependency>
</dependencies>
  1. 基础API测试类 BasicOperationTest.java
public class BasicOperationTest {
    private CuratorFramework client = null;
    /**
     * zookeeper服务器地址
     */
    private static final String ZK_SERVER_PATH = "xxx.xx.xxx.xxx:2181";
    private static final String NODE_PATH = "/hadoop/yarn";
    @Before
    public void prepare() {
        // 重试策略
        RetryPolicy retryPolicy = new RetryNTimes(3, 5000);
        client = CuratorFrameworkFactory.builder()
                .connectString(ZK_SERVER_PATH)
                .sessionTimeoutMs(10000).retryPolicy(retryPolicy)
                .namespace("workspace").build();
        //指定命名空间后,client 的所有路径操作都会以 / workspace 开头
        client.start();
    }
    /**
     * 判断服务状态
     */
    @Test
    public void getStatus() {
        CuratorFrameworkState state = client.getState();
        System.out.println("服务是否已经启动:" + (state == CuratorFrameworkState.STARTED));
    }
    /**
     * 创建节点
     *
     * @throws Exception 可能出现异常
     */
    @Test
    public void createNodes() throws Exception {
        byte[] data = "abc".getBytes();
        client.create().creatingParentsIfNeeded()
                //节点类型
                .withMode(CreateMode.PERSISTENT)
                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                .forPath(NODE_PATH, data);
    }
    /**
     * 获取节点信息
     *
     * @throws Exception 可能出现异常
     */
    @Test
    public void getNode() throws Exception {
        Stat stat = new Stat();
        byte[] data = client.getData().storingStatIn(stat).forPath(NODE_PATH);
        System.out.println("节点数据:" + new String(data));
        System.out.println("节点信息:" + stat.toString());
    }
    /**
     * 获取子节点列表
     *
     * @throws Exception 可能出现异常
     */
    @Test
    public void getChildrenNodes() throws Exception {
        List<String> childNodes = client.getChildren().forPath("/hadoop");
        for (String s : childNodes) {
            System.out.println(s);
        }
    }
    /**
     * 更新节点
     *
     * @throws Exception 可能出现异常
     */
    @Test
    public void updateNode() throws Exception {
        byte[] newData = "defg".getBytes();
        // 传入版本号,如果版本号错误则拒绝更新操作,并抛出 BadVersion 异常
        client.setData().withVersion(0)
                .forPath(NODE_PATH, newData);
    }
    /**
     * 删除节点
     *
     * @throws Exception 可能出现异常
     */
    @Test
    public void deleteNodes() throws Exception {
        client.delete()
                // 如果删除失败,那么在会继续执行,直到成功
                .guaranteed()
                // 如果有子节点,则递归删除
                .deletingChildrenIfNeeded()
                // 传入版本号,如果版本号错误则拒绝删除操作,并抛出 BadVersion 异常
                .withVersion(0)
                .forPath(NODE_PATH);
    }
    /**
     * 判断节点是否存在
     *
     * @throws Exception 可能出现异常
     */
    @Test
    public void existNode() throws Exception {
        // 如果节点存在则返回其状态信息如果不存在则为 null
        Stat stat = client.checkExists().forPath(NODE_PATH + "aa/bb/cc");
        System.out.println("节点是否存在:" + (stat != null));
    }
    /**
     * 创建一次监听
     *
     * @throws Exception 可能出现异常
     */
    @Test
    public void DisposableWatch() throws Exception {
        client.getData().usingWatcher(new CuratorWatcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println("节点" + event.getPath() + "发生了事件:" + event.getType());
            }
        }).forPath(NODE_PATH);
        //休眠以观察测试效果
        Thread.sleep(1000 * 1000);
    }
    /**
     * 创建永久监听
     *
     * @throws Exception 可能出现异常
     */
    @Test
    public void permanentWatch() throws Exception {
        // 使用 NodeCache 包装节点,对其注册的监听作用于节点,且是永久性的
        final NodeCache nodeCache = new NodeCache(client, NODE_PATH);
        // 通常设置为 true, 代表创建 nodeCache 时,就去获取对应节点的值并缓存
        nodeCache.start(true);
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() {
                ChildData currentData = nodeCache.getCurrentData();
                if (currentData != null) {
                    System.out.println("节点路径:" + currentData.getPath() +
                            "数据:" + new String(currentData.getData()));
                }
            }
        });
        // 休眠以观察测试效果
        Thread.sleep(1000 * 1000);
    }
    /**
     * 监听字节点
     *
     * @throws Exception 可能出现异常
     */
    @Test
    public void permanentChildrenNodesWatch() throws Exception {
        // 第三个参数代表除了节点状态外,是否还缓存节点内容
        PathChildrenCache childrenCache = new PathChildrenCache(client, "/hadoop",
                true);
        /*
         * StartMode 代表初始化方式:
         * NORMAL: 异步初始化
         * BUILD_INITIAL_CACHE: 同步初始化
         * POST_INITIALIZED_EVENT: 异步并通知,初始化之后会触发 INITIALIZED 事件
         */
        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        List<ChildData> childDataList = childrenCache.getCurrentData();
        System.out.println("当前数据节点的子节点列表:");
        childDataList.forEach(x -> System.out.println(x.getPath()));
        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent
                    event) {
                switch (event.getType()) {
                    case INITIALIZED:
                        System.out.println("childrenCache 初始化完成");
                        break;
                    case CHILD_ADDED:
                        // 需要注意的是: 即使是之前已经存在的子节点,也会触发该监听,因为会把该子节点加入 childrenCache 缓存中
                        System.out.println("增加子节点:" + event.getData().getPath());
                        break;
                    case CHILD_REMOVED:
                        System.out.println("删除子节点:" + event.getData().getPath());
                        break;
                    case CHILD_UPDATED:
                        System.out.println("被修改的子节点的路径:" +
                                event.getData().getPath());
                        System.out.println("修改后的数据:" + new
                                String(event.getData().getData()));
                        break;
                    default:
                        System.out.println("无匹配!");
                }
            }
        });
        //休眠以观察测试效果
        Thread.sleep(1000 * 1000);
    }
    @After
    public void destroy() {
        if (client != null) {
            client.close();
        }
    }
}
相关实践学习
基于MSE实现微服务的全链路灰度
通过本场景的实验操作,您将了解并实现在线业务的微服务全链路灰度能力。
目录
相关文章
|
14小时前
|
安全 Java API
Java进阶-Java Stream API详解与使用
效、更易于维护的代码,同时享受到函数式编程带来的好处。
9 2
|
1天前
|
Java 大数据 API
利用Java Stream API实现高效数据处理
在大数据和云计算时代,数据处理效率成为了软件开发者必须面对的重要挑战。Java 8及以后版本引入的Stream API为开发者提供了一种声明式、函数式的数据处理方式,极大提升了数据处理的效率和可读性。本文将详细介绍Java Stream API的基本概念、使用方法和性能优势,并通过实际案例展示如何在实际项目中应用Stream API实现高效数据处理。
|
1天前
|
自然语言处理 Java API
Java 8的Stream API和Optional类:概念与实战应用
【5月更文挑战第17天】Java 8引入了许多重要的新特性,其中Stream API和Optional类是最引人注目的两个。这些特性不仅简化了集合操作,还提供了更好的方式来处理可能为空的情况,从而提高了代码的健壮性和可读性。
24 7
|
1天前
|
Java API
Java 8新特性之Lambda表达式与Stream API
【5月更文挑战第17天】本文将介绍Java 8中的两个重要特性:Lambda表达式和Stream API。Lambda表达式是一种新的编程语法,它允许我们将函数作为参数传递给其他方法,从而使代码更加简洁。Stream API是一种用于处理集合的新工具,它提供了一种高效且易于使用的方式来处理数据。通过结合使用这两个特性,我们可以编写出更加简洁、高效的Java代码。
7 0
|
3天前
|
Java API
Java 8新特性之Lambda表达式与Stream API实践指南
【5月更文挑战第15天】 随着Java语言的不断发展,Java 8作为一个重要的版本,引入了许多令人兴奋的新特性。其中,Lambda表达式和Stream API是Java 8最受关注的两个特性。本文将深入探讨Lambda表达式的基本概念、语法和使用场景,以及如何结合Stream API实现更加简洁、高效的代码编写。通过实例演示,帮助读者快速掌握这两个新特性,提高Java编程能力。
|
3天前
|
SQL Java API
Java一分钟之-JPA:Java持久化API简介
【5月更文挑战第14天】Java Persistence API (JPA) 是Java的ORM规范,用于简化数据库操作。常见问题包括实体映射、事务管理和性能问题。避免错误的关键在于明确主键策略、妥善使用事务、优化查询及理解实体生命周期。示例展示了如何定义实体和使用`EntityManager`保存数据。JPA通过标准化API让开发者更专注于业务逻辑,提升开发效率和代码维护性。
14 0
|
3天前
|
Java API
Java一分钟之-Java日期与时间API:LocalDate, LocalDateTime
【5月更文挑战第13天】Java 8引入`java.time`包,改进日期时间API。`LocalDate`代表日期,`LocalDateTime`包含日期和时间。本文概述两者的基本用法、常见问题及解决策略。创建日期时间使用`of()`和`parse()`,操作日期时间有`plusDays()`、`minusMonths()`等。注意点包括:设置正确的`DateTimeFormatter`,考虑闰年影响,以及在需要时区信息时使用`ZonedDateTime`。正确使用这些类能提升代码质量。
12 3
|
3天前
|
Java API 数据处理
Java一分钟之-Stream API:数据处理新方式
【5月更文挑战第13天】Java 8的Stream API为集合操作提供了声明式编程,简化数据处理。本文介绍了Stream的基本概念、常见问题和易错点。问题包括并行流与顺序流的区别,状态改变操作的影响,以及忘记调用终止操作和误用`peek()`。理解并合理使用Stream API能提升代码效率和可维护性。实践中不断探索,将发掘更多Stream API的潜力。
12 3
|
3天前
|
安全 Java API
Java Stream API详解与使用
Java Stream API是Java 8引入的特性,提供函数式操作处理集合,支持链式操作和并行处理,提升代码可读性和性能。关键点包括:延迟执行的中间操作(如filter, map)和触发计算的终端操作(如collect, forEach)。示例展示了如何从Person列表过滤出年龄大于20的姓名并排序。使用Stream时注意避免中间操作的副作用,终端操作后Stream不能复用,以及并行操作的线程安全性。
|
3天前
|
Java 程序员 API
Java 8新特性之Lambda表达式与Stream API的深度解析
【5月更文挑战第12天】本文将深入探讨Java 8中的两个重要新特性:Lambda表达式和Stream API。我们将从基本概念入手,逐步深入到实际应用场景,帮助读者更好地理解和掌握这两个新特性,提高Java编程效率。
42 2

推荐镜像

更多