2018-07-13期 Zookeeper客户端基本操作本人亲自反复验证通过分享

Posted 2020-12-04

说明：本文档主要给大家介绍ZK客户端API基本操作

一、新建Java工程

1、New->Project->Java Project

填写JAVA工程名字，选择JDK版本等

2、为新建的Java工程添加外部jar包

工程名称->Java Build Path->Libraries->用户Library->新建用户Library

为用户Library添加ZK jar包，比如路径如下：

E:depslibzookeeper-3.4.5lib 将此路径下的所有lib添加到新建用户库

为此Java工程准备完成，详细类似操作方法，请参考往期Java工程新建部分，这里不再赘述。

二、ZK客户端API操作

1、ZK连接通用工具类

package cn.itcast.zk.test;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.WatchedEvent;

import org.apache.zookeeper.Watcher;

import org.apache.zookeeper.ZooKeeper;

import org.apache.zookeeper.Watcher.Event.KeeperState;

import org.apache.zookeeper.ZooKeeper.States;

/**

 * 本类是一个ZK客户端端连接工具类

 * @author songjq

 *

 */

public class ZkConnUtils {

/**

* ZK一直处于连接中，则进入等待，直到连接成功为止

* @param zooKeeper

* @param connectedLatch

*/

public static void waitUntilConnected(ZooKeeper zooKeeper, CountDownLatch connectedLatch) {

if (States.CONNECTING == zooKeeper.getState()) {

try {

connectedLatch.await();

} catch (InterruptedException e) {

throw new IllegalStateException(e);

}

}

}

/**

* ZK连接监听器

* @author songjq

*

*/

static class ConnectedWatcher implements Watcher {

private CountDownLatch connectedLatch;

private ZooKeeper zk_;

ConnectedWatcher(CountDownLatch connectedLatch,ZooKeeper zk_) {

this.connectedLatch = connectedLatch;

this.zk_ = zk_;

}

/**

* 监听器回调方法

* 如果需要对某个znode进行持续监听，需要重新在回调方法中注册监听器

*/

@Override

public void process(WatchedEvent event) {

/**

* 例如在exists上注册监听器

*/

System.out.println("节点:" + event.getPath() + " 发生了事件：" + event.getType());

if (event.getState() == KeeperState.SyncConnected) {

connectedLatch.countDown();

}

}

}

/**

* 获取ZK连接实例，调用该方法即可获取可用zk连接

* @param hostports

* @param times

* @return

*/

public ZooKeeper getConnection(String hostports, int times) {

ZooKeeper zk_ = null;

try {

CountDownLatch connectedLatch = new CountDownLatch(1);

Watcher watcher = new ConnectedWatcher(connectedLatch,zk_);

zk_ = new ZooKeeper(hostports, times, watcher);

waitUntilConnected(zk_, connectedLatch);

} catch (Exception e) {

System.out.println(e);

return null;

}

return zk_;

}

static private ZkConnUtils static_;

static public ZkConnUtils Instance() {

if (static_ == null) {

static_ = new ZkConnUtils();

}

return static_;

}

}

2、ZK客户端基本操作

package cn.itcast.zk.test;

import java.io.IOException;

import java.io.UnsupportedEncodingException;

import org.apache.zookeeper.CreateMode;

import org.apache.zookeeper.KeeperException;

import org.apache.zookeeper.ZooDefs.Ids;

import org.apache.zookeeper.ZooKeeper;

import org.junit.Before;

import org.junit.Test;

/**

 * 本类主要实现ZK Znode节点创建、获取、修改、删除、简单监听器注册

 * @author songjq

 *

 */

public class ZkOperation {

private ZooKeeper zkCli = null;

@Before

public void init() throws Exception {

String hostports = "192.168.1.201:2181,192.168.1.202:2181,192.168.1.203:2181";

ZkConnUtils instance = ZkConnUtils.Instance();

zkCli = instance.getConnection(hostports, 2000);

}

/**

* 创建znode节点

* @throws IOException

* @throws KeeperException

* @throws InterruptedException

*/

@Test

public void testCreateZnode() throws IOException, KeeperException, InterruptedException {

// PERSISTENT：永久数据，客户端断开连接，数据依旧存在

String path_PERSISTENT = zkCli.create("/jackmablog", "jackma的博客".getBytes("utf-8"), Ids.OPEN_ACL_UNSAFE,

CreateMode.PERSISTENT);

// EPHEMERAL：临时数据，客户端断开连接，数据自动删除

String path_EPHEMERAL = zkCli.create("/jackmablog/cloudNumber", "90".getBytes(), Ids.OPEN_ACL_UNSAFE,

CreateMode.EPHEMERAL);

System.out.println("path_PERSISTENT:" + path_PERSISTENT);

System.out.println("path_EPHEMERAL:" + path_EPHEMERAL);

//模拟客户端不断开连接，查看临时数据

//Thread.sleep(Long.MAX_VALUE);

}

/**

* 获取数据节点

* @throws KeeperException

* @throws InterruptedException

* @throws UnsupportedEncodingException

*/

@Test

public void getZnodeData() throws KeeperException, InterruptedException, UnsupportedEncodingException {

byte[] data = zkCli.getData("/jackmablog", true, null);

System.out.println("data:"+new String(data, "utf-8"));

}

/**

* 修改数据节点

* @throws Exception

*/

@Test

public void setZnodeData() throws  Exception {

/*

* -1：匹配所有版本

*/

byte[] data1 = zkCli.getData("/jackmablog", true, null);

System.out.println("修改前数据:"+new String(data1, "utf-8"));

   zkCli.setData("/itheima", "伟大的云计算".getBytes("utf-8"), -1);

   byte[] data2 = zkCli.getData("/jackmablog", true, null);

System.out.println("修改后数据:"+new String(data2, "utf-8"));

//模拟zk客户端不断开连接，监视数据变化

//Thread.sleep(Long.MAX_VALUE);

}

/**

* 删除Znode

* @throws InterruptedException

* @throws KeeperException

*/

@Test

public void deleteZnode() throws InterruptedException, KeeperException {

System.out.println((zkCli.exists("/jackmablog", true)==null)?"/jackmablog不存在":"/jackmablog存在");

zkCli.delete("/itheima", -1);

System.out.println((zkCli.exists("/jackmablog", true)==null)?"/jackmablog不存在":"/jackmablog存在");

}

/**

* 持续注册监听器

* @throws InterruptedException 

* @throws KeeperException 

*/

@Test

public void listenerRedister() throws KeeperException, InterruptedException {

//在znode /java上注册监听器

//zkCli.create("/java", "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

zkCli.exists("/java", true);

Thread.sleep(Long.MAX_VALUE);

}

}

3、ZK持续注册监听器

package cn.itcast.zk.test;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;

import org.apache.zookeeper.KeeperException;

import org.apache.zookeeper.WatchedEvent;

import org.apache.zookeeper.Watcher;

import org.apache.zookeeper.ZooKeeper;

import org.apache.zookeeper.Watcher.Event.KeeperState;

import org.apache.zookeeper.ZooDefs.Ids;

import org.apache.zookeeper.ZooKeeper.States;

/**

 * 本类主要实现客户端持续向ZK注册监听器，保障监听器永久有效

 * @author songjq

 *

 */

public class ZkListenerRegister {

private static ZooKeeper zk1_;

public static void waitUntilConnected(ZooKeeper zooKeeper, CountDownLatch connectedLatch) {

if (States.CONNECTING == zooKeeper.getState()) {

try {

connectedLatch.await();

} catch (InterruptedException e) {

throw new IllegalStateException(e);

}

}

}

static class ConnectedWatcher implements Watcher {

private CountDownLatch connectedLatch;

ConnectedWatcher(CountDownLatch connectedLatch) {

this.connectedLatch = connectedLatch;

}

/**

* 监听器回调方法

* 如果需要对某个znode进行持续监听，需要重新在回调方法中注册监听器

*/

@Override

public void process(WatchedEvent event) {

System.out.println("节点:" + event.getPath() + " 发生了事件：" + event.getType());

if (event.getState() == KeeperState.SyncConnected) {

connectedLatch.countDown();

}

/**

* 例如在exists上注册监听器

* 备注：exists上注册的监听器只能对注册的znode进行监听，不能监听子znode

* 若要监听子znode可以在getChildren上注册监听

*/

try {

zk1_.exists("/sjq", true);

zk1_.getChildren("/sjq", true);

} catch (KeeperException | InterruptedException e) {

e.printStackTrace();

}

}

}

public ZooKeeper getConnection(String hostports, int times) {

ZooKeeper zktmp = null;

try {

CountDownLatch connectedLatch = new CountDownLatch(1);

Watcher watcher = new ConnectedWatcher(connectedLatch);

zktmp = new ZooKeeper(hostports, times, watcher);

waitUntilConnected(zktmp, connectedLatch);

} catch (Exception e) {

System.out.println(e);

return null;

}

return zktmp;

}

static private ZkListenerRegister static_;

static public ZkListenerRegister Instance() {

if (static_ == null) {

static_ = new ZkListenerRegister();

}

return static_;

}

public static void main(String[] args) throws Exception {

String hostports = "192.168.1.201:2181,192.168.1.202:2181,192.168.1.203:2181";

ZkListenerRegister instance = ZkListenerRegister.Instance();

zk1_ = instance.getConnection(hostports, 2000);

zk1_.create("/sjq", "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

zk1_.exists("/sjq", true);

Thread.sleep(Long.MAX_VALUE);

}

}