ZooKeeper Java API学习笔记（一）

青旅半醒 2022-05-15 07:14 244阅读 0赞

ZooKeeper 官方提供的API，相比较Curator或者zkclient，比较晦涩，不易使用。不过它是官方API，提供使用ZK的基本操作。下面给出使用API的简单例子（刚开始学习，可能有不对之处，欢迎指出）。

1， pom文件引入依赖

核心依赖是

<dependency>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
                <version>3.4.9</version>
            </dependency>

下面是完整pom文件

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.yq</groupId>
        <artifactId>ZKClientDemo</artifactId>
        <version>2.0</version>
        <packaging>jar</packaging>
    
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
            <java.version>1.8</java.version>
    
            <lombok.version>1.16.14</lombok.version>
            <fastjson.version>1.1.33</fastjson.version>
            <commons.io.version>2.5</commons.io.version>
            <common.lang.version>3.5</common.lang.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
                <version>3.4.9</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>${lombok.version}</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>${fastjson.version}</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-lang3</artifactId>
                <version>${common.lang.version}</version>
            </dependency>
        </dependencies>
    
        <build>
            <finalName>ZKClientDemo</finalName>
            <resources>
                <resource>
                    <directory>src/main/resources</directory>
                    <filtering>true</filtering>
                </resource>
            </resources>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    </project>

2，主要代码

MyZooKeeper 负责连接到zk上，创建path，获取数据，删除path，以及watch zk上的额变化。  
完整的代码在[这里][Link 1]， 欢迎加星，fork。

package com.yq.service;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.ZooDefs;
    import org.apache.zookeeper.ZooKeeper;
    
    import java.io.IOException;
    import java.util.concurrent.CountDownLatch;
    
    
    @Slf4j
    public class MyZooKeeper implements Watcher { 
        protected CountDownLatch countDownLatch = new CountDownLatch(1);
    
        private static final int SESSION_TIME = 2000;
        public static ZooKeeper zooKeeper = null;
    
        private static final String PATH_NAME = "/yqPath";
    
        @Override
        public void process(WatchedEvent event) {
            log.info("event notification={}", event.getState() );
            String path = event.getPath();
            log.info("type={}. path={}", event.getType(), path);
            if(event.getState()==KeeperState.SyncConnected){
                countDownLatch.countDown();
    
            }
            else if (event.getType() == Event.EventType.NodeCreated) {
                log.info("node create={}", event.getPath() );
            }
            else if (event.getType() == Event.EventType.NodeDeleted) {
                log.info("node delete={}", event.getPath() );
            }if (event.getType() == Event.EventType.NodeDataChanged) {
                String data = readData(event.getPath());
                log.info("data change. path={}, data={}", event.getPath(), data);
            }
    
            //必须重新注册，要不然就无法获得变化通知
            try {
                zooKeeper.exists(PATH_NAME, this);
            }catch (Exception ex) {
                log.error( "Failed to connect, Exception , ex={}", ex.getMessage(), ex );
            }
        }
    
        public void connect(String hosts){
            try {
                if(zooKeeper == null){
                    zooKeeper = new ZooKeeper(hosts,SESSION_TIME,this);
                    zooKeeper.exists(PATH_NAME, this);
                    countDownLatch.await();
                }
            } catch (IOException ex) {
                log.error("Failed to connect, IOException , ex={} ", ex.getMessage(), ex);
            } catch (InterruptedException ex) {
                log.error( "Failed to connect, InterruptedException , ex={}", ex.getMessage(), ex );
            } catch (Exception ex) {
                log.error( "Failed to connect, Exception , ex={}", ex.getMessage(), ex );
            }
        }
    
    
        public void close(){
            try {
                if(zooKeeper != null){
                    zooKeeper.close();
                }
            } catch (InterruptedException ex) {
                log.error("release connection error={}", ex.getMessage() ,ex);
            }
        }
    
        public boolean createPath(String path, String data){
            try {
                String zkPath = MyZooKeeper.zooKeeper.create(path, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                log.info("ZooKeeper. create path={}", zkPath);
                return true;
            } catch (KeeperException e) {
                log.error("Failed to create path. errMsg：" + e.getMessage() + "，path:" + path  ,e);
            } catch (InterruptedException e) {
                log.error("Failed to create path. errMsg:" + e.getMessage() + "，path:" + path  ,e);
            }
            return false;
        }
        public boolean detelePath(String path){
            try {
                MyZooKeeper.zooKeeper.delete(path, -1);
                log.info("ZooKeeper delete path，path={}", path);
                return true;
            } catch (InterruptedException e) {
                log.error("Failed to delete path, errMsg：" + e.getMessage() + "，path:" + path ,e);
            } catch (KeeperException e) {
                log.error("Failed to delete path, errMsg：" + e.getMessage() + "，path:" + path  ,e);
            }
            return false;
        }
    
        public String readData(String path ){
            String data = null;
            try {
                data = new String( MyZooKeeper.zooKeeper.getData( path, false, null ) );
                log.info( "read data. path={}, content={}", path, data);
            } catch (KeeperException e) {
                log.error("Failed to read data. zKException!.", path, e.getMessage());
            } catch (InterruptedException e) {
                log.error("Failed to read data. 发生InterruptedException!. path={}, exMsg={}", path, e.getMessage());
            }
            return  data;
        }
    }

main函数。 这是个简单的示例程序，因此我在主函数使用 Thread.sleep(1000 \* 120); 让用户有机会通过zkCli命令行工具，对zk的数据进行变更。

package com.yq;
    
    import com.yq.service.MyZooKeeper;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /** * Simple to Introduction * className: ZKDemoOneApplication * * @author EricYang * @version 2018/8/24 23:43 */
    
    public class ZKClientApp { 
        private static final Logger log = LoggerFactory.getLogger(ZKClientApp.class);
        private static final String ZK_SERVERS = "127.0.0.1:2181";
        private static final String PATH_NAME_TEST = "/test001";
    
        public static void main(String[] args) {
    
            MyZooKeeper zk = new MyZooKeeper();
            zk.connect(ZK_SERVERS);
            zk.createPath(PATH_NAME_TEST, "hello world!");
            String value = zk.readData(PATH_NAME_TEST);
            log.info("znode={}, data={}", PATH_NAME_TEST, value);
            zk.detelePath(PATH_NAME_TEST);
            try {
                Thread.sleep(1000 * 120);
            }
            catch (Exception ex){
                log.info("start ex={}.", ex);
            }
    
            log.info("start done.");
        }
    }

3，截图

通过截图可以的看到，当我们对/yqPath的数据进行变更时，程序得到了通知  
![这里写图片描述][70]

[Link 1]: https://github.com/yqbjtu/SpringCloudTutorials/tree/master/ZKClientDemo
[70]: /images/20220515/765322e78635490aa6cd6d4755ede882.png