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

本文涉及的产品
注册配置 MSE Nacos/ZooKeeper,118元/月
MSE Nacos 企业版免费试用,1600元额度,限量50份
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介: 【4月更文挑战第11天】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>
    
    AI 代码解读
  2. 基础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();
        }
    }
}
AI 代码解读
相关实践学习
基于MSE实现微服务的全链路灰度
通过本场景的实验操作,您将了解并实现在线业务的微服务全链路灰度能力。
目录
打赏
0
10
11
1
283
分享
相关文章
尼恩一键开发环境: vagrant+java+springcloud+redis+zookeeper镜像下载(&制作详解)
尼恩提供了一系列文章,旨在帮助开发者轻松搭建一键开发环境,涵盖Java分布式、高并发场景下的多种技术组件安装与配置。内容包括但不限于Windows和CentOS虚拟机的安装与排坑指南、MySQL、Kafka、Redis、Zookeeper等关键组件在Linux环境下的部署教程,并附带详细的视频指导。此外,还特别介绍了Vagrant这一虚拟环境部署工具,
尼恩一键开发环境: vagrant+java+springcloud+redis+zookeeper镜像下载(&制作详解)
Hadoop-32 ZooKeeper 分布式锁问题 分布式锁Java实现 附带案例和实现思路代码
Hadoop-32 ZooKeeper 分布式锁问题 分布式锁Java实现 附带案例和实现思路代码
143 2
Hadoop-30 ZooKeeper集群 JavaAPI 客户端 POM Java操作ZK 监听节点 监听数据变化 创建节点 删除节点
Hadoop-30 ZooKeeper集群 JavaAPI 客户端 POM Java操作ZK 监听节点 监听数据变化 创建节点 删除节点
169 1
|
6天前
|
Java 多线程:线程安全与同步控制的深度解析
本文介绍了 Java 多线程开发的关键技术,涵盖线程的创建与启动、线程安全问题及其解决方案,包括 synchronized 关键字、原子类和线程间通信机制。通过示例代码讲解了多线程编程中的常见问题与优化方法,帮助开发者提升程序性能与稳定性。
43 0
从阻塞到畅通:Java虚拟线程开启并发新纪元
从阻塞到畅通:Java虚拟线程开启并发新纪元
137 83
Java 无锁方式实现高性能线程实战操作指南
本文深入探讨了现代高并发Java应用中单例模式的实现方式,分析了传统单例(如DCL)的局限性,并提出了多种无锁实现方案。包括基于ThreadLocal的延迟初始化、VarHandle原子操作、Record不可变对象、响应式编程(Reactor)以及CDI依赖注入等实现方式。每种方案均附有代码示例及适用场景,同时通过JMH性能测试对比各实现的优劣。最后,结合实际案例设计了一个高性能配置中心,展示了无锁单例在实际开发中的应用。总结中提出根据场景选择合适的实现方式,并遵循现代单例设计原则以优化性能和安全性。文中还提供了代码获取链接,便于读者实践与学习。
43 0
Java虚拟线程:轻量级并发的革命性突破
Java虚拟线程:轻量级并发的革命性突破
151 83
说一说 JAVA 内存模型与线程
我是小假 期待与你的下一次相遇 ~

热门文章

最新文章

AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等

登录插画

登录以查看您的控制台资源

管理云资源
状态一览
快捷访问