java操作zookeeper，包括增删改节点及其事件通知

java操作zookeeper，包括增删改节点及其事件通知

今天写了一个java操作zookeeper的demo（为方便使用windows版zookeeper）。后续会在此基础上实现服务注册与发现，负载均衡，还会用zookeeper实现分布式锁，以及后面的leader选举也会用到。这一系列的目标是结合前面实现的RPC（后面应该会优化）来实现一个简单的微服务框架，一方面加深自己的理解，一方面希望能帮助到初学者明白一些微服务原理性的东西。

• zookeeper客户端demo

package testZookeeper.testZKBsae;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.concurrent.CountDownLatch;

public class ZKClient implements Watcher {
    private ZooKeeper zookeeper;
    // 集群连接地址
    private static final String ZOOKEEPER_ADDRES = "127.0.0.1:2181";
    private CountDownLatch countDownLatch = new CountDownLatch(1);

    /**
     * 创建zookeeper连接
     * @param connectAddres 连接地址
     * @param sessionTimeOut 超时时间
     */
    private void createConnection(String connectAddres, int sessionTimeOut) {
        try {
            zookeeper = new ZooKeeper(connectAddres, sessionTimeOut, this);
            System.out.println("开始连接zokeeper服务....");
            //等待直到连接创建成功
            countDownLatch.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建zookeeper节点
     * @param nodePath 节点路径
     * @param data 节点数据
     * @return 是否创建成功
     */
    private boolean createNode(String nodePath, String data) {
        try {
            exists(nodePath, true);
            zookeeper.create(nodePath, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println("节点创建成功, Path:" + nodePath + ",data:" + data);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    /**
     * 修改zookeeper节点数据
     * @param nodePath 节点路径
     * @param data 节点数据
     * @return 是否修改成功
     */
    private boolean updateNode(String nodePath, String data){
        exists(nodePath, true);
        try {
            zookeeper.setData(nodePath, data.getBytes(), -1);
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 列出指定path下的孩子
     */
    private void list(String path){
        System.out.println(path);
        List<String> list = null;
        try {
            list = zookeeper.getChildren(path,this);
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
        if(list == null || list.isEmpty()){
            return;
        }
        for(String pa : list){
            //迭代输出path下的所有节点
            if(path.equals("/")){
                list(path + pa);
            }
            else{
                list(path + "/" + pa);
            }
        }
    }

    /**
     * 判断指定节点是否存在
     * @param node 节点路径
     */
    private Stat exists(String node, boolean needWatch) {
        try {
            return zookeeper.exists(node, needWatch);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * zookeeper事件监听
     * @param watchedEvent watchedEvent对象
     */
    @Override
    public void process(WatchedEvent watchedEvent) {
        Event.KeeperState zkState = watchedEvent.getState();
        String path = watchedEvent.getPath();
        Event.EventType eventType = watchedEvent.getType();

        if(zkState == Event.KeeperState.SyncConnected){
            if(eventType == Event.EventType.None){
                //连接建立事件
                System.out.println("建立zookeeper连接成功!");
                countDownLatch.countDown();
            }else if(eventType == Event.EventType.NodeCreated){
                //节点创建事件
                System.out.println("节点" + path + "被创建");
            }else if(eventType == Event.EventType.NodeDataChanged){
                //节点修改事件
                System.out.println("节点" + path + "被修改");

            }else if(eventType == Event.EventType.NodeDeleted){
                //节点删除事件
                System.out.println("节点" + path + "被删除");
            }
        }

    }

    public static void main(String[] args) {
        ZKClient zkClient = new ZKClient();
        zkClient.createConnection(ZOOKEEPER_ADDRES, 5000);
        zkClient.createNode("/zk_test_parent", "main_node");
        zkClient.createNode("/zk_test_parent/zk_test_child", "child_node");
        zkClient.updateNode("/zk_test_parent","farther_node");
        zkClient.createNode("/zk_test_parent/zk_test_child/zk_test_cc", "cc_node");
        zkClient.list("/");
    }
}

• 代码执行前
  

• 代码执行结果
  

• 刷新后发现节点都已创建完成