ZooKeeper Java API操作实战深度解析

张

张建站

2026/7/9 12:02:37

10分钟阅读

一、环境搭建与前置准备1.1 创建Maven工程在IDEA中创建Maven工程命名为zookeeper-client。1.2 导入核心依赖dependenciesdependencygroupIdjunit/groupIdartifactIdjunit/artifactIdversion4.13.2/version/dependencydependencygroupIdorg.apache.logging.log4j/groupIdartifactIdlog4j-core/artifactIdversion2.18.0/version/dependencydependencygroupIdorg.apache.zookeeper/groupIdartifactIdzookeeper/artifactIdversion3.5.7/version/dependency/dependencies依赖说明junit单元测试框架用于验证API操作log4j-core日志输出便于观察ZooKeeper内部交互细节zookeeperZooKeeper官方Java客户端版本3.5.7与集群版本保持一致1.3 配置log4j日志在src/main/resources目录下创建log4j.propertieslog4j.rootLoggerINFO, stdout log4j.appender.stdoutorg.apache.log4j.ConsoleAppender log4j.appender.stdout.layoutorg.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern%d %p [%c]- %m%n log4j.appender.logfileorg.apache.log4j.FileAppender log4j.appender.logfile.Filetarget/spring.log log4j.appender.logfile.layoutorg.apache.log4j.PatternLayout log4j.appender.logfile.layout.ConversionPattern%d %p [%c]- %m%n日志级别设为INFO既能观察关键交互又不会因DEBUG信息过多干扰阅读。二、客户端初始化与连接建立2.1 核心成员变量publicclassZkClientTest{// 集群连接地址逗号分隔注意逗号前后不能有空格privatestaticfinalStringCONNECT_STRINGhadoop102:2181,hadoop103:2181,hadoop104:2181;// 会话超时时间单位毫秒建议设置为tickTime的2-20倍privatestaticfinalintSESSION_TIMEOUT2000;// ZooKeeper客户端实例privateZooKeeperzkClient;}2.2 初始化连接BeforeBeforepublicvoidinit()throwsIOException{zkClientnewZooKeeper(CONNECT_STRING,SESSION_TIMEOUT,newWatcher(){Overridepublicvoidprocess(WatchedEventevent){System.out.println(---------- 监听器触发 ------------);System.out.println(事件类型event.getType());System.out.println(节点路径event.getPath());System.out.println(连接状态event.getState());// 持续监听每次触发后重新注册监听try{ListStringchildrenzkClient.getChildren(/sanguo,true);System.out.println(当前子节点列表children);}catch(KeeperException|InterruptedExceptione){e.printStackTrace();}}});}关键设计解析Watcher回调机制ZooKeeper的所有监听都通过Watcher接口实现当事件发生时客户端会回调process方法持续监听 vs 一次性监听在init()中设置的Watcher会持续生效因为每次触发后我们在process中重新调用了getChildren(path, true)再次注册异步连接new ZooKeeper()是异步操作构造函数立即返回真正的连接建立通过Watcher的SyncConnected事件通知三、核心API操作实战3.1 创建节点create方法签名与参数解读Stringcreate(Stringpath,byte[]data,ListACLacl,CreateModecreateMode)参数类型说明pathString要创建的节点路径必须以/开头databyte[]节点存储的数据最大1MB可为nullaclList访问控制列表ZooDefs.Ids.OPEN_ACL_UNSAFE表示完全开放createModeCreateMode节点类型持久/临时/顺序创建持久节点示例TestpublicvoidcreateNode()throwsInterruptedException,KeeperException{// 创建持久节点 /sanguo/wuguo数据为 liubeiStringnodePathzkClient.create(/sanguo/wuguo,liubei.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);System.out.println(创建成功节点路径nodePath);// 输出创建成功节点路径/sanguo/wuguo}四种节点类型创建对比TestpublicvoidcreateAllTypes()throwsException{// 1. 持久无序号节点zkClient.create(/demo/persistent,data.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);// 2. 持久有序号节点zkClient.create(/demo/persistent_seq,data.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT_SEQUENTIAL);// 实际路径/demo/persistent_seq0000000000// 3. 临时无序号节点zkClient.create(/demo/ephemeral,data.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL);// 4. 临时有序号节点分布式锁核心zkClient.create(/demo/ephemeral_seq,data.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);// 实际路径/demo/ephemeral_seq0000000000}3.2 获取子节点并监听getChildren核心代码TestpublicvoidgetChildrenAndWatch()throwsException{System.out.println(-------------------- 开始监听 --------------------);// 参数2true表示使用默认Watcher即init中创建的WatcherListStringchildrenzkClient.getChildren(/sanguo,true);// 初次获取的子节点列表System.out.println(当前子节点children);// 阻塞线程保持客户端运行等待监听事件Thread.sleep(Long.MAX_VALUE);}监听效果演示测试流程运行getChildrenAndWatch()测试方法在另一终端通过zkCli.sh操作# 创建 /sanguo/weiguo 节点[zk:localhost:2181(CONNECTED)2]create /sanguo/weiguocaocao# 创建 /sanguo/weiguo/simayi 孙节点不会触发监听因为监听只针对子节点[zk:localhost:2181(CONNECTED)3]create /sanguo/weiguo/simayisimayi# 创建 /sanguo/sima 新子节点触发监听[zk:localhost:2181(CONNECTED)6]create /sanguo/simasimaIDEA控制台输出-------------------- 开始监听 -------------------- 当前子节点[wuguo, shuguo] ---------- 监听器触发 ------------ 事件类型NodeChildrenChanged 节点路径/sanguo 连接状态SyncConnected 当前子节点列表[wuguo, shuguo, weiguo] ---------- 监听器触发 ------------ 事件类型NodeChildrenChanged 节点路径/sanguo 连接状态SyncConnected 当前子节点列表[wuguo, shuguo, weiguo, sima]重要结论getChildren只能监听直接子节点的变化孙节点的变化不会触发监听每次事件触发后需要重新注册监听才能继续监听后续变化在init()中通过循环调用getChildren(path, true)实现了持续监听3.3 判断节点是否存在existsTestpublicvoidcheckNodeExists()throwsException{// 参数1节点路径// 参数2是否开启监听false表示不监听StatstatzkClient.exists(/sanguo,false);if(statnull){System.out.println(节点不存在);}else{System.out.println(节点存在);System.out.println(数据版本stat.getVersion());System.out.println(数据长度stat.getDataLength());System.out.println(创建时间stat.getCtime());System.out.println(最后修改时间stat.getMtime());}}Stat对象包含了节点的所有元数据即使节点数据为空Stat也能提供丰富的状态信息。四、写数据流程深度剖析ZooKeeper的写操作是理解其分布式一致性的核心。集群中的每台服务器都保存相同的数据副本只要一台服务器修改了Znode信息集群中的每个节点都会立即更新。4.1 场景一直接发送给Leader节点执行步骤客户端向Leader发送写数据请求Leader写入本地将数据写入Leader节点的内存数据库Leader向Follower同步数据通过ZAB协议ZooKeeper Atomic Broadcast将数据同步到所有FollowerFollower写入并返回ACK每个Follower写入成功后向Leader发送确认过半数确认即成功当Leader收到超过半数节点包括自己的ACK后写操作完成响应客户端Leader通知客户端写入成功继续同步剩余节点未同步完成的节点也会立即同步数据但不影响客户端响应假设集群有3台节点1 Leader 2 Follower只要Leader 任意1台Follower写入成功即满足过半数客户端即可收到成功响应。4.2 场景二直接发送给Follower节点执行步骤客户端向Follower1发送写数据请求Follower1转发给LeaderFollower没有写入权限将请求转发给LeaderLeader写入本地Leader将数据写入自己的内存数据库Leader向所有Follower同步数据包括最初接收请求的Follower1各Follower写入并返回ACK所有Follower写入成功后向Leader确认过半数确认即成功Leader收到超过半数节点的ACK后写操作完成Leader通知Follower1由Follower1原路返回成功响应给客户端继续同步剩余节点确保最终所有节点数据一致4.3 两种写流程对比对比项写给Leader写给Follower网络跳数2跳Client→Leader→Client4跳Client→Follower→Leader→Follower→Client写入延迟较低较高负载均衡Leader压力大可分散读请求但写请求最终都到Leader数据一致性强一致性强一致性过半数机制是是五、完整代码示例packagecom.example.zookeeper;importorg.apache.zookeeper.*;importorg.apache.zookeeper.data.Stat;importorg.junit.Before;importorg.junit.Test;importjava.io.IOException;importjava.util.List;publicclassZooKeeperApiDemo{privatestaticfinalStringCONNECT_STRINGhadoop102:2181,hadoop103:2181,hadoop104:2181;privatestaticfinalintSESSION_TIMEOUT2000;privateZooKeeperzkClient;Beforepublicvoidinit()throwsIOException{zkClientnewZooKeeper(CONNECT_STRING,SESSION_TIMEOUT,event-{System.out.println(事件触发event.getType() - event.getPath());if(event.getType()Event.EventType.NodeChildrenChanged){try{ListStringchildrenzkClient.getChildren(event.getPath(),true);System.out.println(更新后的子节点children);}catch(Exceptione){e.printStackTrace();}}});}TestpublicvoidtestCreate()throwsException{StringpathzkClient.create(/test,hello.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);System.out.println(创建节点path);}TestpublicvoidtestGetChildren()throwsException{ListStringchildrenzkClient.getChildren(/,true);System.out.println(根节点子节点children);Thread.sleep(Long.MAX_VALUE);}TestpublicvoidtestExists()throwsException{StatstatzkClient.exists(/test,false);System.out.println(statnull?不存在:存在版本stat.getVersion());}TestpublicvoidtestDelete()throwsException{zkClient.delete(/test,-1);System.out.println(删除成功);}}六、核心要点总结知识点关键内容连接初始化异步建立通过Watcher监听连接状态create参数path, data, acl, createMode4种节点类型Watch机制一次性触发需重新注册只监听直接子节点exists作用判断节点存在性返回Stat元数据写流程核心过半数确认机制Leader协调ZAB协议同步写给Leader2跳完成延迟低写给Follower4跳完成Follower转发给Leader一致性保证无论写到哪里最终都通过Leader过半数确认