亲宝软件园·资讯

展开

Java调用Zookeeper Java调用Zookeeper的实现步骤

blue星空 人气:0
想了解Java调用Zookeeper的实现步骤的相关内容吗,blue星空在本文为您仔细讲解Java调用Zookeeper的相关知识和一些Code实例,欢迎阅读和指正,我们先划重点:Java调用Zookeeper,下面大家一起来学习吧。

watch机制

Zookeeper watch是一种监听通知机制,可以随时监听一些数据的变化,从而实现数据的及时性。

Zookeeper所有的读操作getData(), getChildren()和 exists()都可以设置监听(watch)。【写操作则是不能设置监视点的。】

Watch的三个关键点:

常用API

/**
* 构造器
* @param connectString 集群的IP:端口号;多个服务器时,中间用逗号分割
* @param sessionTimeout 超时时间,单位:毫秒
* @param watcher  监听器,监听节点变化
* @throws IOException
*/
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException

/**
*
* @param path 节点路径
* @param data 数据
* @param acl 访问控制列表
* @param createMode 节点类型
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public String create(String path, byte[] data, List<ACL> acl, CreateMode createMode) throws KeeperException, InterruptedException

/**
*
* @param path 节点路径
* @param watch 监听器
* @return 所有的子节点的名称
* @throws KeeperException
* @throws InterruptedException
*/
public List<String> getChildren(String path, boolean watch) throws KeeperException, InterruptedException

/**
*
* @param path 节点路径
* @param watcher 监听器
* @param stat  状态信息【可以为null】
* @return 节点数据的二进制数组【可以通过new String()转换成字符串信息】
* @throws KeeperException
* @throws InterruptedException
*/
public byte[] getData(String path, Watcher watcher, Stat stat) throws KeeperException, InterruptedException

/**
*
* @param path 节点路径
* @param watch 监听器
* @param cb 回调函数
* @param ctx  上下文参数 ?【该参数不太理解,望知道的留言讲解,谢谢】
*/
public void getData(String path, boolean watch, AsyncCallback.DataCallback cb, Object ctx)

/**
*
* @param path 节点路径
* @param data 数据
* @param version 版本号【初始通常赋值为-1,每次更新会自动+1】
* @return  状态信息
* @throws KeeperException
* @throws InterruptedException
*/
public Stat setData(String path, byte[] data, int version) throws KeeperException, InterruptedException

/**
*如果Stat为null,则节点不存在
* @param path 节点路径
* @param watch 监听器
* @return 状态信息
* @throws KeeperException
* @throws InterruptedException
*/
public Stat exists(String path, boolean watch) throws KeeperException, InterruptedException

/**
* 如果要删除的节点有子节点,会报错:KeeperException$NotEmptyException: KeeperErrorCode = Directory not empty for
* 如果节点不存在,会报错:KeeperException$NoNodeException: KeeperErrorCode = NoNode for
* @param path 节点路径
* @param version 版本号[version = -1 : 匹配所有的版本]
* @throws InterruptedException
* @throws KeeperException
*/
public void delete(String path, int version) throws InterruptedException, KeeperException

JAVA调用

初始化

try {
  ZooKeeper zooKeeper = new ZooKeeper("172.23.34.13:2181", 15000, event -> {
        if (event.getType() == Watcher.Event.EventType.None && event.getState() == Watcher.Event.KeeperState.SyncConnected) {
            System.out.println("Connectted successful.");
        }
    });
} catch (IOException e) {
    e.printStackTrace();
}

创建节点: create

@Test
public void create() throws KeeperException, InterruptedException {
    //参数:1,节点路径; 2,要存储的数据; 3,节点的权限; 4,节点的类型
    String nodePath = zooKeeper.create("/java/2183", "This is Java Node 2183.".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    System.out.println(nodePath);
}

获取子节点: ls

public void getChildren() throws KeeperException, InterruptedException {
    List<String> children = zooKeeper.getChildren("/", true);
    for (String child : children) {
        System.out.println("child: "+child);
    }
}

同步获取节点内容: get

@Test
public void getData() throws KeeperException, InterruptedException {
    String path = "/java";
    byte[] bytes = zooKeeper.getData(path, event -> {
        if (event.getType() == Watcher.Event.EventType.NodeDataChanged && path.equals(event.getPath())) {
            System.out.println("Date changed.");
        }
    }, null);
    System.out.printf("The data of %s is : %s \n",path, new String(bytes));
}

异步获取节点内容: get

@Test
public void getDataAsync() {
    String path = "/java";
    zooKeeper.getData(path, false, new AsyncCallback.DataCallback() {
        @Override
        public void processResult(int i, String s, Object o, byte[] bytes, Stat stat) {
            System.out.printf("The data of %s is : %s \n",path, new String(bytes));
        }
    },"1000");

    //休眠20秒,查看响应结果
    try {
        Thread.sleep(20000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

指定版本号更新数据:set

@Test
public void setData() throws KeeperException, InterruptedException {
    Stat stat = zooKeeper.setData("/java", "This is from java.".getBytes(), -1);
    //更新节点后,version会自动+1。故,返回值为0
    System.out.println(stat.getAversion());
}

多线程下更新数据:set

@Test
public void setDataThread() throws KeeperException, InterruptedException {
    String path = "/java";
    Stat stat = new Stat();
    //1,先获取节点的当前版本
    zooKeeper.getData(path,false,stat);
    //2,在当前版本的基础上修改节点内容
    zooKeeper.setData(path, "This is from java.".getBytes(), stat.getVersion());
}

判断节点是否存在

@Test
public void exists() throws KeeperException, InterruptedException {
    Stat stat = zooKeeper.exists("/java", false);
    if (stat == null) {
        System.out.println("Not Exists.");
    }else {
        System.out.println("Exists.");
    }
}

删除节点

@Test
public void delete() throws KeeperException, InterruptedException {
    //version = -1 : 匹配所有的版本
    zooKeeper.delete("/java/2182", -1);
}

加载全部内容

相关教程
猜你喜欢
用户评论