首页 > 其他 > 详细

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

时间:2018-07-13 10:55:05      阅读:225      评论:0      收藏:0      [点我收藏+]
说明:本文档主要给大家介绍ZK客户端API基本操作


一、新建Java工程

1、New->Project->Java Project

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


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

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

为用户Library添加ZK jar包,比如路径如下:

E:\depslib\zookeeper-3.4.5\lib 将此路径下的所有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);

}


}


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

原文:http://blog.51cto.com/2951890/2141268

(0)
(0)
   
举报
评论 一句话评论(0
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣,分享程序人生!