[大数据]基于Java API实现访问ZooKeeper

我们可以使用Java API访问ZooKeeper，对ZooKeeper的节点进行一些必要的操作，下面我们一起来看下该如何通过代码实现。

环境准备

1）首先我们要保证已经开启的ZooKeeper的端口2181，如果你还没开启，请在防火墙启动的情况下执行如下指令：

firewall-cmd --zone=public --add-port=2181/tcp --permanent

然后重新加载防火墙

firewall-cmd --reload

2）启动ZooKeeper

知识准备

1）Java 连接 ZK，客户端需要通过创建一个 ZooKeeper 实例来连接 ZK 服务器，ZooKeeper 构造方法有如下几个：

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)
 
public ZooKeeper(String connectString, int sessionTimeout,Watcher watcher,ZKClientConfig conf)
 
public ZooKeeper(String connectString, int sessionTimeout,Watcher watcher, boolean canBeReadOnly, HostProvider aHostProvider) 
 
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly, HostProvider aHostProvider, ZKClientConfig clientConfig) 

Zookeeper(Arguments) 方法

• connectString：连接服务器列表，用“，”分隔
• sessionTimeout：心跳检测时间周期（毫秒）
• watcher：事件处理通知器
• canBeReadOnly：标识当前会话是否支持只读
• SessionId?和?SessionPasswd：提供连接 Zookeeper 的sessionId 和 密码，通过这俩个确定唯一一台客户端，目的是可以提供重复会话

ZK 客户端和服务器会话的建立是一个异步的过程，我们程序方法在处理完客户端初始化后立即返回，也就是程序往下执行代码，这样，大多数情况下我们并没有真正构建好一个可用会话，在会话生命周期处于“CONNECTING”时，才算建立完毕。

2）创建节点（znode）方法：create
提供了两套创建节点的方法，同步和异步创建节点方法

参数1：

节点路径：/nodeName （不允许递归创建节点，也就是在父节点不存在的情况下，不允许创建子节点）

参数2：

节点内容：要求类型是字节数组（不支持序列化方式，如果需要实现程序化，可使用 Java 相关序列化框架，如 Hession、Kryo 框架）

参数3：

节点权限：使用 Ids.OPEN_ACL_UNSAFE 开放权限

参数4：

节点类型：创建节点的类型： CreateMode.*

• ?persistent（持久节点）
• ?persistent_sequential（持久顺序节点）
• ?ephemeral（临时节点）
• ?ephemeral_sequential（临时顺序节点）

参数5：

注册一个异步回调函数，要实现 AsynCallBack.StringCallBack 接口，重写 processResult(int rc, String path, Object ctx, String name) 方法，当节点创建完毕后执行此方法。

• rc：为服务端相应码 0 表示调用成功、-4 表示端口连接、-110 表示指定节点存在、-112 表示会话已经过期
• ?path：接口调用时传入 API 的数据节点的路径参数
• ?ctx：为调用接口传入 API 的 ctx 值
• ?name：实际在服务器端创建节点的名称

参数6：

传递给回调函数的参数，一般为上下文（Context）信息

代码实现

1）首先我们需要创建一个Maven项目，其pom.xml中导入ZooKeeper操作需要的jar包（版本自行根据自己的zk版本决定），具体如下：

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

2）新建一个测试类（我们这里采用的是单机版），具体代码如下：

package com.panziye;
 
import org.apache.zookeeper.*;
 
import java.io.IOException;
 
/**
 * <h3>zk</h3>
 *
 * @author panziye
 * @description <p></p>
 * @date 2021-04-24 22:15
 **/
public class ZKDemo implements Watcher {
    private static final int SESSION_TIMEOUT = 30000;
    public static ZooKeeper zooKeeper;
 
    public static void main(String[] args) {
        String path = "/zknode1";
        try {
            zooKeeper = new ZooKeeper("192.168.55.128:2181",SESSION_TIMEOUT,new ZKDemo());
            zooKeeper.exists(path,true);
            //创建节点
            zooKeeper.create(path,"zkcontent1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
            Thread.sleep(3000);
 
            //得到节点内容
            byte[] bytes1 = zooKeeper.getData(path,null,null);
            String result1 = new String(bytes1);
            System.out.println("result1==="+result1);
 
            //设置节点内容
            zooKeeper.setData(path,"testSetData111".getBytes(),-1);
 
            //再次的带节点内容
            byte[] bytes2 = zooKeeper.getData(path,null,null);
            String result2 = new String(bytes2);
            System.out.println("result2==="+result2);
            Thread.sleep(3000);
 
            //删除节点
            zooKeeper.delete(path,-1);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }finally {
            if(zooKeeper != null){
                try {
                    zooKeeper.close();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
 
        }
 
    }
 
    @Override
    public void process(WatchedEvent watchedEvent) {
        if(watchedEvent.getState() == Event.KeeperState.SyncConnected){
            if(watchedEvent.getType() == Event.EventType.NodeCreated){
                //当节点创建成功时进行回调，此处进行提示打印
                System.out.println("Node created success....");
                try {
                    zooKeeper.exists(watchedEvent.getPath(),true);
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }else if(watchedEvent.getType() == Event.EventType.NodeDeleted){
                try {
                    zooKeeper.exists(watchedEvent.getPath(),true);
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Node deleted success....");
            }else if(watchedEvent.getType() == Event.EventType.NodeDataChanged){
                try {
                    zooKeeper.exists(watchedEvent.getPath(),true);
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Node changed success....");
            }
        }
    }
}
 

3）运行测试，结果如下：

本文首发于潘子夜个人博客：https://www.panziye.com/bigdata/3063.html，转载请注明出处，万分感谢!