Zookeeper-04-ZK的API基本使用

叁歲伎倆 2023-09-29 13:32 7阅读 0赞

### 文章目录 ###

*  1：环境准备
 *   *  1：准备好Zookeeper或集群
     *  2：创建项目，导入pom依赖
     *  3:拷贝 log4j.properties 文件到项目根目录
 *  2：基本api
 *   *  1:创建zookeeper客户端
     *  2:创建节点
     *  3：获取节点
     *  4：永久监听
     *  5：修改数据
     *  7：并发修改节点
     *  8:删除节点
     *  9:异步获取数据

# 1：环境准备 #

## 1：准备好Zookeeper或集群 ##

[Zookeeper安装教程][Zookeeper]  
注意：Zookeeper使用的jdk建议为oracleJDK，不建议使用openjdk

## 2：创建项目，导入pom依赖 ##

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

## 3:拷贝 log4j.properties 文件到项目根目录 ##

需要在项目的 src/main/resources 目录下，新建一个文件，命名为“log4j.properties”，在  
文件中填入。

log4j.rootLogger=INFO, stdout 
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
    log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] 
    - %m%n 
    log4j.appender.logfile=org.apache.log4j.FileAppender 
    log4j.appender.logfile.File=target/spring.log 
    log4j.appender.logfile.layout=org.apache.log4j.PatternLayout 
    log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

# 2：基本api #

## 1:创建zookeeper客户端 ##

public class AppTest {
        
    
        String connectString = "192.168.138.128:2181";
        int timeout = 4000;
    
        @Test
        public void NewZK() throws IOException, InterruptedException {
        
            System.out.println("start-------------");
            //zk是有session概念的，没有连接池的概念
            //watch:观察，回调
            //watch的注册值发生在 读类型调用，get，exites。。。
            //第一类：new zk 时候，传入的watch，这个watch，session级别的，跟path 、node没有关系。
            ZooKeeper zooKeeper = new ZooKeeper(connectString, timeout, new Watcher() {
        
                @Override
                public void process(WatchedEvent watchedEvent) {
        
                    Event.KeeperState state = watchedEvent.getState();
                    Event.EventType type = watchedEvent.getType();
                    String path = watchedEvent.getPath();
                    System.out.println("new zk watch: "+ watchedEvent.toString());
    
                    switch (state) {
        
                        case Unknown:
                            break;
                        case Disconnected:
                            break;
                        case NoSyncConnected:
                            break;
                        case SyncConnected:
                            System.out.println("connected");
                            //cd.countDown();
                            break;
                        case AuthFailed:
                            break;
                        case ConnectedReadOnly:
                            break;
                        case SaslAuthenticated:
                            break;
                        case Expired:
                            break;
                    }
    
                    switch (type) {
        
                        case None:
                            break;
                        case NodeCreated:
                            break;
                        case NodeDeleted:
                            break;
                        case NodeDataChanged:
                            break;
                        case NodeChildrenChanged:
                            break;
                    }
                }
            });
    
            
        }
    }

输出：  
![在这里插入图片描述][476c1d1305a9425abc4082b80568c70c.png]  
说明：

① 创建zk客户端连接过程是异步的；

② 底层是使用守护线程，当没有业务线程执行的话，那么守护线程也就结束了。这里也就是main线程结束守护线程也就结束了。

③ 如何避免main线程不结束呢？使用线程的sleep，休眠一下。或者使用CountDownLatch 来阻塞线程

public class AppTest {
        
    
        String connectString = "192.168.138.128:2181";
        int timeout = 4000;
    
        @Test
        public void NewZK() throws IOException, InterruptedException {
        
            CountDownLatch cd = new CountDownLatch(1);
            System.out.println("start-------------");
            //zk是有session概念的，没有连接池的概念
            //watch:观察，回调
            //watch的注册值发生在 读类型调用，get，exites。。。
            //第一类：new zk 时候，传入的watch，这个watch，session级别的，跟path 、node没有关系。
            ZooKeeper zooKeeper = new ZooKeeper(connectString, timeout, new Watcher() {
        
                @Override
                public void process(WatchedEvent watchedEvent) {
        
                    Event.KeeperState state = watchedEvent.getState();
                    Event.EventType type = watchedEvent.getType();
                    String path = watchedEvent.getPath();
                    System.out.println("new zk watch: "+ watchedEvent.toString());
    
                    switch (state) {
        
                        case Unknown:
                            break;
                        case Disconnected:
                            break;
                        case NoSyncConnected:
                            break;
                        case SyncConnected:
                            System.out.println("connected");
                            cd.countDown();
                            break;
                        case AuthFailed:
                            break;
                        case ConnectedReadOnly:
                            break;
                        case SaslAuthenticated:
                            break;
                        case Expired:
                            break;
                    }
    
                    switch (type) {
        
                        case None:
                            break;
                        case NodeCreated:
                            break;
                        case NodeDeleted:
                            break;
                        case NodeDataChanged:
                            break;
                        case NodeChildrenChanged:
                            break;
                    }
                }
            });
            Thread.sleep(10);
            cd.await();
    
        }
    }

注： 集群的情况下，多个地址使用逗号分隔：

String connectString = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183,127.0.0.1:2184";

## 2:创建节点 ##

String connectString = "192.168.138.128:2181";
        int timeout = 4000;
    
        @Test
        public void NewZK() throws IOException, InterruptedException, KeeperException {
        
            CountDownLatch cd = new CountDownLatch(1);
            System.out.println("start-------------");
            //zk是有session概念的，没有连接池的概念
            //watch:观察，回调
            //watch的注册值发生在 读类型调用，get，exites。。。
            //第一类：new zk 时候，传入的watch，这个watch，session级别的，跟path 、node没有关系。
            ZooKeeper zooKeeper = new ZooKeeper(connectString, timeout, new Watcher() {
        
                @Override
                public void process(WatchedEvent watchedEvent) {
        
                    Event.KeeperState state = watchedEvent.getState();
                    Event.EventType type = watchedEvent.getType();
                    String path = watchedEvent.getPath();
                    System.out.println("new zk watch: "+ watchedEvent.toString());
    
                    switch (state) {
        
                        case Unknown:
                            break;
                        case Disconnected:
                            break;
                        case NoSyncConnected:
                            break;
                        case SyncConnected:
                            System.out.println("connected");
                            cd.countDown();
                            break;
                        case AuthFailed:
                            break;
                        case ConnectedReadOnly:
                            break;
                        case SaslAuthenticated:
                            break;
                        case Expired:
                            break;
                    }
    
                    switch (type) {
        
                        case None:
                            break;
                        case NodeCreated:
                            break;
                        case NodeDeleted:
                            break;
                        case NodeDataChanged:
                            break;
                        case NodeChildrenChanged:
                            break;
                    }
                }
            });
            Thread.sleep(10);
            cd.await();
    
            //创建节点
            String pathName = zooKeeper.create("/ooxx", "olddata".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            System.out.println("pathName:"+pathName);
        }

输出：  
![在这里插入图片描述][aef77ab027a647e99b900ed19c384ac4.png]

注：  
创建节点调用的方法是：zooKeeper.create()，第一个参数是节点的path；第二个参数是byte\[\] data；第三个参数是：ACl权限信息；第四个参数是：节点的类型；

## 3：获取节点 ##

//通过create创建数据，通过get获取数据
            //这种方式只能监听一次
            byte[] dataBytes = zooKeeper.getData("/ooxx",new Watcher(){
        
                public void process(WatchedEvent event) {
        
                    if(event.getType() == Event.EventType.NodeDataChanged && event.getPath() != null && event.getPath().equals("/ooxx")){
        
                        System.out.println("数据改变了："+event.getPath());
    
                    }
                }
            },null);
            System.out.println("获取到的数据是："+ new String(dataBytes));

输出：  
![在这里插入图片描述][e5058bfd997b4e57a7bcfc934188acde.png]  
此时已经通过获取节点数据为此节点创建了一个监听器，我们通过zkcli.sh 来改变该节点数据，（注意此时主线程设置位长久休眠，必须保证主线程还获取，否则守护线程timeout时间后就会自动消亡）  
![在这里插入图片描述][6196e669fc4b4340b9b913b92b1cc7cb.png]  
![在这里插入图片描述][3729c27fa31740608afc14256c85c059.png]  
注：  
![在这里插入图片描述][2fa120aee1d0425cba76a0ae1e0806a3.png]  
第二次改变时，发现并没有输出，说明监控功能，只能监控一次

## 4：永久监听 ##

我们发现现在我们的编码只能监听一次，如何实现永久监听呢？很简单，只要在监听的代码里再次监听就可以了

//通获过create创建数据，通过get取数据
            //这种方式只能监听一次
            byte[] dataBytes = zooKeeper.getData("/ooxx",new Watcher(){
        
                public void process(WatchedEvent event) {
        
                    if(event.getType() == Event.EventType.NodeDataChanged && event.getPath() != null && event.getPath().equals("/ooxx")){
        
                        System.out.println("数据改变了："+event.getPath());
    
                        try {
        
                            byte[] dataBytes = zooKeeper.getData("/ooxx",this,null);
                            System.out.println("再次监听获取到的数据是："+ new String(dataBytes));
                        } catch (KeeperException e) {
        
                            e.printStackTrace();
                        } catch (InterruptedException e) {
        
                            e.printStackTrace();
                        }
                    }
                }
            },null);
            System.out.println("获取到的数据是："+ new String(dataBytes));

## 5：修改数据 ##

修改节点使用的是set的操作，我们看下简单的例子：

Stat stat1 = zooKeeper.setData("/ooxx", "newdata".getBytes(), 0);

![在这里插入图片描述][fe18ec815d4a480cb052db45e4178870.png]  
说明：

（1）执行set的时候，如果node不存在，会抛出异常：KeeperErrorCode = NoNode for /myconfig。在执行zooKeeper.setData()就会抛出异常，就不会有返回值Stat了

（2）这里的第三个参数是version：

① 如果不考虑并发修改的问题的话，那么version=-1;
    
    ② 如果填写的version和我们节点的version对不上的话，那么执行setData会报错：KeeperErrorCode = BadVersion for /myconfig。

## 7：并发修改节点 ##

对于节点的修改，会出现多个线程进行并发修改的问题，那么我们控制并发节点的修改问题呐，很简单，在前面的例子中的第三个参数version就是用来解决节点的并发修改问题的。具体的一个修改思路：

（1）、通过getData获取到节点的版本信息；

（2）、在执行setData的时候，传递当前获取到的版本号；

Stat nodeStat = new Stat();
    zooKeeper.getData(path, false,nodeStat);
     
    byte[] dataBytes = new String("hello-update").getBytes();
    zooKeeper.setData(path,dataBytes,nodeStat.getVersion());

## 8:删除节点 ##

zooKeeper.delete(path, -1);

版本号的说明：-1 代表匹配所有版本号，直接删除。任意大于-1的代表可以指定数据版本删除。

## 9:异步获取数据 ##

因为上边的获取数据是同步进行的，我们可以利用DataCallback来异步获取，即获取到数据后就直接执行processResult方法，同时不影响主线程继续执行；

zk.getData("/ooxx", false, new AsyncCallback.DataCallback() {
        
                @Override
                public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
        
                    System.out.println("-------async call back----------");
                    System.out.println(ctx.toString());
                    System.out.println(new String(data));
    
                }
    
            },"abc");

[Zookeeper]: https://blog.csdn.net/qq_41694906/article/details/124613743
[476c1d1305a9425abc4082b80568c70c.png]: https://img-blog.csdnimg.cn/476c1d1305a9425abc4082b80568c70c.png
[aef77ab027a647e99b900ed19c384ac4.png]: https://img-blog.csdnimg.cn/aef77ab027a647e99b900ed19c384ac4.png
[e5058bfd997b4e57a7bcfc934188acde.png]: https://img-blog.csdnimg.cn/e5058bfd997b4e57a7bcfc934188acde.png
[6196e669fc4b4340b9b913b92b1cc7cb.png]: https://img-blog.csdnimg.cn/6196e669fc4b4340b9b913b92b1cc7cb.png
[3729c27fa31740608afc14256c85c059.png]: https://img-blog.csdnimg.cn/3729c27fa31740608afc14256c85c059.png
[2fa120aee1d0425cba76a0ae1e0806a3.png]: https://img-blog.csdnimg.cn/2fa120aee1d0425cba76a0ae1e0806a3.png
[fe18ec815d4a480cb052db45e4178870.png]: https://img-blog.csdnimg.cn/fe18ec815d4a480cb052db45e4178870.png