zookeeper API demo - 简书
zookeeper API demo
zk server启动
例子是client创建一个session需要确保zkServer开启了，即
Paste_Image.png
demo实现1.客户端创建zk链接2.创建，修改，删除znode3.获取子节点列表，节点内容4.watcher的使用
import org.apache.zookeeper.AsyncC
import org.apache.zookeeper.CreateM
import org.apache.zookeeper.WatchedE
import org.apache.zookeeper.W
import org.apache.zookeeper.ZooD
import org.apache.zookeeper.ZooK
import org.apache.zookeeper.data.S
import java.util.L
import java.util.concurrent.CountDownL
public class Main {
private static CountDownLatch countDownLatch = new CountDownLatch(1);
private static ZooK
private static final String ZOOKEEPER_ZNODE_NAME = "zookeeper";
public static void startZK() throws Exception {
System.out.println("startZK----------------------");
//确保server确实已经开启了，这里是创建client到server的session
zk = new ZooKeeper("127.0.0.1:2181", 20000,
new Watcher() {
public void process(WatchedEvent watchedEvent) {
System.out.println("process " + watchedEvent);
if (watchedEvent.getState() ==
Event.KeeperState.SyncConnected) {
countDownLatch.countDown();
countDownLatch.await();
} catch (Exception e) {
e.printStackTrace();
System.out.println("state is " + zk.getState());
System.out.println("zk session begin");
//同步创建临时节点
public static void syncCreateNode() {
System.out.println("syncCreateNode----------------------");
String path1 = zk.create("/test1", "znode1".getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
String path2 = zk.create("/test2", "znode2".getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
System.out.println("path1 = " + path1);
System.out.println("path2 = " + path2);
} catch (Exception e) {
e.printStackTrace();
//异步创建临时节点，这里等待它执行完看processResult
public static void asnycCreateNode() throws Exception {
System.out.println("asnycCreateNode----------------------");
zk.create("/test3", "znode3".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL, new IStringCallBack(), "context");
Thread.sleep(2000);
static class IStringCallBack implements AsyncCallback.StringCallback {
public void processResult(int rc, String path, Object ctx,
String name) {
System.out.println(
"Create path result " + rc + " " + path + " " + ctx + " " +
public static void clear() throws Exception {
List&String& childList = zk.getChildren("/", false);
for(String s : childList) {
if(s.equals(ZOOKEEPER_ZNODE_NAME))
zk.delete("/" + s, -1);
//获取子节点列表，在子节点列表变更时触发
public static void getChildren() throws Exception {
System.out.println("getChildren----------------------");
List&String& childList = zk.getChildren("/", new Watcher() {
//这个znode的子节点变化的时候会收到通知
public void process(WatchedEvent watchedEvent) {
System.out.println("getChildren " + watchedEvent);
System.out.println("childList " + childList);
//获取数据，注册监听器，在znode内容被改变时触发
public static void getData() throws Exception {
System.out.println("getData----------------------");
String ans1 = new String(zk.getData("/test1", false, null));
String ans2 = new String(zk.getData("/test2", new Watcher() {
public void process(WatchedEvent watchedEvent) {
System.out.println("getData " + watchedEvent);
}, null));
System.out.println("znode /test1 content is " + ans1);
System.out.println("znode /test2 content is " + ans2);
//更新内容，会触发对应znode的watch事件
public static void setData() throws Exception {
System.out.println("setData----------------------");
String data = "zNode22";
zk.setData("/test2", data.getBytes(), -1);
String ans2 = new String(zk.getData("/test2", false, null));
System.out.println("setData to " + ans2);
//节点是否存在，watch监听节点的创建，删除以及更新
public static void exists() throws Exception {
System.out.println("exists----------------------");
Stat stat = zk.exists("/test2", new Watcher() {
public void process(WatchedEvent watchedEvent) {
System.out.println("exists " + watchedEvent);
System.out.println("stat is " + stat);
public static void delete() throws Exception {
System.out.println("delete----------------------");
zk.delete("/test2", -1);
zk.delete("/test1", -1);
public static void main(String[] args) throws Exception {
startZK();
syncCreateNode();
asnycCreateNode();
getChildren();
getData();
setData();
startZK----------------------
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
process WatchedEvent state:SyncConnected type:None path:null
state is CONNECTED
zk session begin
syncCreateNode----------------------
path1 = /test1
path2 = /test2
asnycCreateNode----------------------
Create path result 0 /test3 context /test3
getChildren----------------------
childList [test2, test3, zookeeper, test1]
exists----------------------
stat is 166,166,3,3,0,0,0,,166
getData----------------------
znode /test1 content is znode1
znode /test2 content is znode2
setData----------------------
getData WatchedEvent state:SyncConnected type:NodeDataChanged path:/test2
exists WatchedEvent state:SyncConnected type:NodeDataChanged path:/test2
setData to zNode22
delete----------------------
getChildren WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/
Process finished with exit code 0
watcher是一次性的，只触发一次,参见refer，并不是一直监听。不是很理解这样做的原因watcher用于监听各种事件，包括子节点的变动，节点的删除，创建，以及内容(实质是版本)的改变"zookeeper"是默认存在的一个znode
网易杭州，后台开发，工作经验1年。27922人阅读
Hadoop（12）
org.apache.zookeeper.ZooKeeper类 主要方法列表
String create(final String path, byte data[], List acl, CreateMode createMode)
创建一个znode节点，
参数： 路径、 znode内容，ACL(访问控制列表)、 znode创建类型
void delete(final String path, int version)
删除一个znode节点，
参数： 路径、版本号；如果版本号与znode的版本号不一致，将无法删除，是一种乐观加锁机制；如果将版本号设置为-1，不会去检测版本，直接删除；
Stat exists(final String path, Watcher watcher)
判断某个znode节点是否存在
参数： 路径、Watcher(监视器)；当这个znode节点被改变时，将会触发当前Watcher
Stat exists(String path, boolean watch)
判断某个znode节点是否存在
参数： 路径、并设置是否监控这个目录节点，这里的 watcher 是在创建 ZooKeeper 实例时指定的 watcher
Stat setData(final String path, byte data[], int version)
设置某个znode上的数据
参数： 路径、数据、版本号；如果为-1，跳过版本检查
byte[] getData(final String path, Watcher watcher, Stat stat)
获取某个znode上的数据
参数： 路径、监视器、数据版本等信息
List getChildren(final String path, Watcher watcher)
获取某个节点下的所有子节点
参数： 路径、监视器;该方法有多个重载
znode创建类型(CreateMode)：
PERSISTENT & & & & & & &&持久化节点
PERSISTENT_SEQUENTIAL &&& 顺序自动编号持久化节点，这种节点会根据当前已存在的节点数自动加 1
EPHEMERAL &&&临时节点， 客户端session超时这类节点就会被自动删除
EPHEMERAL_SEQUENTIAL &&临时自动编号节点
* ZookeeperTest.java
package com.x.
import java.io.IOE
import java.util.L
import org.apache.zookeeper.CreateM
import org.apache.zookeeper.WatchedE
import org.apache.zookeeper.W
import org.apache.zookeeper.ZooK
import org.apache.zookeeper.ZooDefs.I
import org.apache.zookeeper.data.S
import org.junit.A
import org.junit.A
import org.junit.B
import org.junit.T
import org.slf4j.L
import org.slf4j.LoggerF
* @author http://blog.csdn.net/java2000_wl
* @version &b&1.0&/b&
public class ZookeeperTest {
private static final int SESSION_TIMEOUT = 30000;
public static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperTest.class);
private Watcher watcher =
new Watcher() {
public void process(WatchedEvent event) {
(&process : & + event.getType());
private ZooKeeper zooK
连接zookeeper
* &br&------------------------------&br&
* @throws IOException
public void connect() throws IOException {
= new ZooKeeper(&localhost:2181,localhost:2182,localhost:2183&, SESSION_TIMEOUT, watcher);
* &br&------------------------------&br&
public void close() {
zooKeeper.close();
} catch (InterruptedException e) {
e.printStackTrace();
* 创建一个znode
1.CreateMode 取值
PERSISTENT：持久化，这个目录节点存储的数据不会丢失
PERSISTENT_SEQUENTIAL：顺序自动编号的目录节点，这种目录节点会根据当前已近存在的节点数自动加 1，然后返回给客户端已经成功创建的目录节点名；
EPHEMERAL：临时目录节点，一旦创建这个节点的客户端与服务器端口也就是 session过期超时，这种节点会被自动删除
EPHEMERAL_SEQUENTIAL：临时自动编号节点
* &br&------------------------------&br&
public void testCreate() {
String result =
result = zooKeeper.create(&/zk001&, &zk001data&.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
} catch (Exception e) {
LOGGER.error(e.getMessage());
Assert.fail();
(&create result : {}&, result);
* 删除节点
* &br&------------------------------&br&
public void testDelete() {
zooKeeper.delete(&/zk001&, -1);
} catch (Exception e) {
LOGGER.error(e.getMessage());
Assert.fail();
* &br&------------------------------&br&
public void testGetData() {
String result =
byte[] bytes = zooKeeper.getData(&/zk001&, null, null);
result = new String(bytes);
} catch (Exception e) {
LOGGER.error(e.getMessage());
Assert.fail();
(&getdata result : {}&, result);
* &br&------------------------------&br&
public void testGetDataWatch() {
String result =
byte[] bytes = zooKeeper.getData(&/zk001&, new Watcher() {
public void process(WatchedEvent event) {
(&testGetDataWatch
watch : {}&, event.getType());
result = new String(bytes);
} catch (Exception e) {
LOGGER.error(e.getMessage());
Assert.fail();
(&getdata result : {}&, result);
// 触发wacth
NodeDataChanged
zooKeeper.setData(&/zk001&, &testSetData&.getBytes(), -1);
} catch (Exception e) {
LOGGER.error(e.getMessage());
Assert.fail();
判断节点是否存在
设置是否监控这个目录节点，这里的 watcher 是在创建 ZooKeeper实例时指定的 watcher
* &br&------------------------------&br&
public void testExists() {
Stat stat =
stat = zooKeeper.exists(&/zk001&, false);
} catch (Exception e) {
LOGGER.error(e.getMessage());
Assert.fail();
Assert.assertNotNull(stat);
(&exists result : {}&, stat.getCzxid());
设置对应znode下的数据
-1表示匹配所有版本
* &br&------------------------------&br&
public void testSetData() {
Stat stat =
stat = zooKeeper.setData(&/zk001&, &testSetData&.getBytes(), -1);
} catch (Exception e) {
LOGGER.error(e.getMessage());
Assert.fail();
Assert.assertNotNull(stat);
(&exists result : {}&, stat.getVersion());
判断节点是否存在,
设置是否监控这个目录节点，这里的 watcher 是在创建 ZooKeeper实例时指定的 watcher
* &br&------------------------------&br&
public void testExistsWatch1() {
Stat stat =
stat = zooKeeper.exists(&/zk001&, true);
} catch (Exception e) {
LOGGER.error(e.getMessage());
Assert.fail();
Assert.assertNotNull(stat);
zooKeeper.delete(&/zk001&, -1);
} catch (Exception e) {
e.printStackTrace();
判断节点是否存在,
设置监控这个目录节点的 Watcher
* &br&------------------------------&br&
public void testExistsWatch2() {
Stat stat =
stat = zooKeeper.exists(&/zk002&, new Watcher() {
public void process(WatchedEvent event) {
(&testExistsWatch2
watch : {}&, event.getType());
} catch (Exception e) {
LOGGER.error(e.getMessage());
Assert.fail();
Assert.assertNotNull(stat);
// 触发watch 中的process方法
NodeDataChanged
zooKeeper.setData(&/zk002&, &testExistsWatch2&.getBytes(), -1);
} catch (Exception e) {
e.printStackTrace();
// 不会触发watch 只会触发一次
zooKeeper.delete(&/zk002&, -1);
} catch (Exception e) {
e.printStackTrace();
获取指定节点下的子节点
* &br&------------------------------&br&
public void testGetChild() {
zooKeeper.create(&/zk/001&, &001&.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
zooKeeper.create(&/zk/002&, &002&.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
List&String& list = zooKeeper.getChildren(&/zk&, true);
for (String node : list) {
(&node {}&, node);
} catch (Exception e) {
LOGGER.error(e.getMessage());
Assert.fail();
&&相关文章推荐
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：946131次
积分：8290
积分：8290
排名：第2363名
原创：83篇
转载：10篇
评论：329条
阅读：85225
阅读：10982
文章：11篇
阅读：181827
(2)(2)(1)(1)(1)(9)(4)(2)(7)(10)(4)(13)(6)(19)(4)(8)(1)(1)(4)Zookeeper API_百度文库
两大类热门资源免费畅读
续费一年阅读会员，立省24元！
Zookeeper API
W3Cschool（）最大的技术知...|
总评分0.0|
阅读已结束，下载文档到电脑
想免费下载更多文档？
定制HR最喜欢的简历
下载文档到电脑，方便使用
还剩10页未读，继续阅读
定制HR最喜欢的简历
你可能喜欢4334人阅读
ZooKeeper（3）
在ZooKeeper中，节点也称为znode。由于对于程序员来说，对zk的操作主要是对znode的操作，因此，有必要对znode进行深入的了解。
ZooKeeper采用了类似文件系统的的数据模型，其节点构成了一个具有层级关系的树状结构。例如，图1展示了zk节点的层级树状结构。
图1：ZooKeeper的层级树状结构
图1中，根节点 / 包含了两个字节点 /module1，/module2，而节点
/module1 又包含了三个字节点 /module1/app1，/module1/app2，/module1/app3。在zk中，节点以绝对路径表示，不存在相对路径，且路径最后不能以 / 结尾（根节点除外）。
根据节点的存活时间，可以对节点划分为持久节点和临时节点。节点的类型在创建时就被确定下来，并且不能改变。
持久节点的存活时间不依赖于客户端会话，只有客户端在显式执行删除节点操作时，节点才消失。
临时节点的存活时间依赖于客户端会话，当会话结束，临时节点将会被自动删除（当然也可以手动删除临时节点）。利用临时节点的这一特性，我们可以使用临时节点来进行集群管理，包括发现服务的上下线等。
ZooKeeper规定，临时节点不能拥有子节点。
使用命令create可以创建一个持久节点。
create /module1 module1
这样，便创建了一个持久节点/module1，且其数据为”module1”。
使用create命令，并加上-e参数，可以创建一个临时节点。
create -e /module1/app1 app1
这样，便创建了一个临时节点 /module1/app1，数据为”app1”。关闭会话，然后输入命令：
get /module1/app1
可以看到有以下提示，说明临时节点已经被删除。
Node does not exist: /module1/app1
ZooKeeper中还提供了一种顺序节点的节点类型。每次创建顺序节点时，zk都会在路径后面自动添加上10位的数字（计数器），例如 & path &，& path &，……这个计数器可以保证在同一个父节点下是唯一的。在zk内部使用了4个字节的有符号整形来表示这个计数器，也就是说当计数器的大小超过时，将会发生溢出。
顺序节点为节点的一种特性，也就是，持久节点和临时节点都可以设置为顺序节点。这样一来，znode一共有4种类型：持久的、临时的，持久顺序的，临时顺序的。
使用命令create加上-s参数，可以创建顺序节点，例如，
create -s /module1/app app
Created /module1/app
便创建了一个持久顺序节点 /module1/app。如果再执行此命令，则会生成节点 /module1/app。
如果在create -s再添加-e参数，则可以创建一个临时顺序节点。
节点的数据
在创建节点时，可以指定节点中存储的数据。ZooKeeper保证读和写都是原子操作，且每次读写操作都是对数据的完整读取或完整写入，并不提供对数据进行部分读取或者写入的操作。
以下命令创建一个节点/module1/app2，且其存储的数据为app2。
create /module1/app2 app2
ZooKeeper虽然提供了在节点存储数据的功能，但它并不将自己定位为一个通用的数据库，也就是说，你不应该在节点存储过多的数据。Zk规定节点的数据大小不能超过1M，但实际上我们在znode的数据量应该尽可能小，因为数据过大会导致zk的性能明显下降。如果确实需要存储大量的数据，一般解决方法是在另外的分布式数据库（例如redis）中保存这部分数据，然后在znode中我们只保留这个数据库中保存位置的索引即可。
节点的属性
每个znode都包含了一系列的属性，通过命令get，我们可以获得节点的属性。
get /module1/app2
cZxid = 0xe
ctime = Thu Jun 30 20:41:55 HKT 2016
mZxid = 0xe
mtime = Thu Jun 30 20:41:55 HKT 2016
pZxid = 0xe
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 0
对于每个znode来说，均存在三个版本号：
dataVersion
数据版本号，每次对节点进行set操作，dataVersion的值都会增加1（即使设置的是相同的数据）。
子节点的版本号。当znode的子节点有变化时，cversion 的值就会增加1。
aclVersion
ACL的版本号，关于znode的ACL（Access Control List，访问控制），可以参考 参考资料1 有关ACL的描述。
以数据版本号来说明zk中版本号的作用。每一个znode都有一个数据版本号，它随着每次数据变化而自增。ZooKeeper提供的一些API例如setData和delete根据版本号有条件地执行。多个客户端对同一个znode进行操作时，版本号的使用就会显得尤为重要。例如，假设客户端C1对znode /config写入一些配置信息，如果另一个客户端C2同时更新了这个znode，此时C1的版本号已经过期，C1调用setData一定不会成功。这正是版本机制有效避免了数据更新时出现的先后顺序问题。在这个例子中，C1在写入数据时使用的版本号无法匹配，使得操作失败。图2描述了这个情况。
图2：使用版本号来阻止并行操作的不一致性
对于zk来说，每次的变化都会产生一个唯一的事务id，zxid（ZooKeeper Transaction Id）。通过zxid，可以确定更新操作的先后顺序。例如，如果zxid1小于zxid2，说明zxid1操作先于zxid2发生。
需要指出的是，zxid对于整个zk都是唯一的，即使操作的是不同的znode。
Znode创建的事务id。
Znode被修改的事务id，即每次对znode的修改都会更新mZxid。
图3：Zxid在客户端重连中的作用
在集群模式下，客户端有多个服务器可以连接，当尝试连接到一个不同的服务器时，这个服务器的状态要与最后连接的服务器的状态要保持一致。Zk正是使用zxid来标识这个状态，图3描述了客户端在重连情况下zxid的作用。当客户端因超时与S1断开连接后，客户端开始尝试连接S2，但S2延迟于客户端所识别的状态。然而，S3的状态与客户端所识别的状态一致，所以客户端可以安全连接上S3。
包括znode的创建时间和修改时间，创建时间是znode创建时的时间，创建后就不会改变；修改时间在每次更新znode时都会发生变化。
以下命令创建了一个 /module2 节点。
create /module2 module2
Created /module2
通过 get 命令，可以看到 /module2的 ctime和mtime均为Sat Jul 02 11:18:32 CST 2016。
get /module2
cZxid = 0x2
ctime = Sat Jul 02 11:18:32 CST 2016
mZxid = 0x2
mtime = Sat Jul 02 11:18:32 CST 2016
pZxid = 0x2
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 7
numChildren = 0
修改 /module2，可以看到 ctime 没有发生变化，mtime已更新为最新的时间。
set /module2 module2_1
cZxid = 0x2
ctime = Sat Jul 02 11:18:32 CST 2016
mZxid = 0x3
mtime = Sat Jul 02 11:18:50 CST 2016
pZxid = 0x2
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 9
numChildren = 0
《ZooKeeper分布式过程协同技术详解》，Flavio Junqueira等著，谢超等译
&&相关文章推荐
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：487678次
积分：5357
积分：5357
排名：第5004名
原创：154篇
评论：105条
阅读：4891
阅读：5115
文章：53篇
阅读：58499
文章：17篇
阅读：159595
(1)(2)(4)(3)(7)(10)(2)(7)(1)(1)(3)(3)(2)(4)(3)(1)(1)(2)(2)(3)(7)(3)(2)(1)(1)(1)(2)(1)(2)(1)(1)(1)(1)(1)(1)(1)(17)(6)(19)(16)(1)(2)(1)(1)(1)(1)(1)(1)(1)(1)(1)(2)(1)(1)9.1 基本使用
org.apache.zookeeper.Zookeeper是客户端入口主类，负责建立与server的会话
它提供以下几类主要方法
在本地目录树中创建一个节点
删除一个节点
测试本地是否存在目标节点
get/set data
从目标节点上读取 / 写数据
get/set ACL
9.1 基本使用
org.apache.zookeeper.Zookeeper是客户端入口主类，负责建立与server的会话
它提供以下几类主要方法
在本地目录树中创建一个节点
删除一个节点
测试本地是否存在目标节点
get/set data
从目标节点上读取 / 写数据
get/set ACL
获取 / 设置目标节点访问控制列表信息
get children
检索一个子节点上的列表
等待要被传送的数据
表 1 ： ZooKeeper API 描述
9.2 增删改查znode数据
cn.com.toto.
java.io.IOE
org.apache.zookeeper.CreateM
org.apache.zookeeper.KeeperE
org.apache.zookeeper.WatchedE
org.apache.zookeeper.W
org.apache.zookeeper.ZooDefs.I
org.apache.zookeeper.ZooK
public class SimpleDemo {
//回话超时时间，设置为与系统默认时间一致
private static final int SESSION_TIMEOUT = 30000;
//创建ZooKeeper实例
ZooKeeper zk;
//创建Watcher实例
Watcher wh = new Watcher() {
public void process(WatchedEvent event) {
System.out.println(event.toString());
//初始化ZooKeeper实例
private void createZKInstance() throws IOException {
zk = new ZooKeeper("hadoop:2181,hadoop2:2181,hadoop3:2181",SimpleDemo.SESSION_TIMEOUT,this.wh);
private void ZKOperations() throws KeeperException, InterruptedException {
System.out.println("/n1. 创建 ZooKeeper 节点 (znode ： zoo2, 数据： myData2 ，权限： OPEN_ACL_UNSAFE ，节点类型： Persistent");
zk.create("/zoo2", "myData2".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
System.out.println("/n2.查看是否创建成功：");
System.out.println(new String(zk.getData("/zoo2", false, null)));
System.out.println("/n3.修改节点数据");
zk.setData("/zoo2", "toto".getBytes(), -1);
System.out.println("/n4.查看是否修改成功:");
System.out.println(new String(zk.getData("/zoo2", false, null)));
System.out.println("/n5.删除节点");
zk.delete("/zoo2", -1);
System.out.println("/n6.查看节点是否被删除：");
System.out.println("节点状态:[" + zk.exists("/zoo2", false) + "]");
private void ZKClose() throws InterruptedException {
zk.close();
public static void main(String[] args) throws KeeperException,
InterruptedException, IOException {
SimpleDemo dm = new SimpleDemo();
dm.createZKInstance();
dm.ZKOperations();
dm.ZKClose();
运行结果：
/n1. 创建 ZooKeeper 节点 (znode ： zoo2, 数据： myData2 ，权限： OPEN_ACL_UNSAFE ，节点类型： Persistent
一月 10, :26 上午
org.apache.zookeeper.ClientCnxn$SendThread primeConnection
信息: Socket connection established to
hadoop3/192.168.106.82:2181, initiating session
一月 10, :26 上午
org.apache.zookeeper.ClientCnxn$SendThread onConnected
信息: Session establishment complete on server
hadoop3/192.168.106.82:2181, sessionid = 0x007, negotiated
timeout = 30000
WatchedEvent state:SyncConnected type:None path:null
/n2.查看是否创建成功：
/n3.修改节点数据
/n4.查看是否修改成功:
/n5.删除节点
/n6.查看节点是否被删除：
节点状态:[null]
9.3 监听znode
Zookeeper的监听器工作机制
监听器是一个接口，我们的代码中可以实现Wather这个接口，实现其中的process方法，方法中即我们自己的业务逻辑
监听器的注册是在获取数据的操作中实现：
getData(path,watch?)监听的事件是：节点数据变化事件
getChildren(path,watch?)监听的事件是：节点下的子节点增减变化事件
9.4其它案例
所需jar包：
图1 项目包结构
cn.com.toto.
java.util.L
java.util.concurrent.CountDownL
org.apache.zookeeper.CreateM
org.apache.zookeeper.KeeperE
org.apache.zookeeper.WatchedE
org.apache.zookeeper.W
org.apache.zookeeper.Watcher.Event.KeeperS
org.apache.zookeeper.ZooDefs.I
org.apache.zookeeper.ZooK
org.apache.zookeeper.data.S
org.junit.B
org.junit.T
public class SimpleZkClient {
private static final String connectString = "192.168.106.80:.106.81:.106.82:2181";
private static final int sessionTimeout = 2000;
// latch就相当于一个对象锁，当latch.await()方法执行时，方法所在的线程会等待
//当latch的count减为0时，将会唤醒等待的线程
CountDownLatch latch = new CountDownLatch(1);
ZooKeeper zkClient = null;
public void init() throws Exception {
zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
//事件监听回调方法
public void process(WatchedEvent event) {
if (latch.getCount() & 0 && event.getState() == KeeperState.SyncConnected) {
System.out.println("countdown");
latch.countDown();
//收到事件通知后的回调函数(应该是我们自己的事件处理逻辑)
System.out.println(event.getType() + "---" + event.getPath());
System.out.println(event.getState());
latch.await();
//创建数据节点到zk中
public void testCreate() throws KeeperException, InterruptedException {
//参数1：要创建的节点的路径
参数2:节点大数据参数3:节点的权限
参数4:节点的类型
String nodeCreated = zkClient.create("/eclipse", "hellozk".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//上传的数据可以是任何类型，但都要转成byte
zkClient.close();
//判断znode是否存在
public void testExist() throws KeeperException, InterruptedException {
Stat stat = zkClient.exists("/eclipse", false);
System.out.println(stat == null ? "not exist" : "exist");
//获取znode下的孩子节点
public void getChildren() throws KeeperException, InterruptedException {
List&String& children = zkClient.getChildren("/", true);
for(String child : children) {
System.out.println(child);
Thread.sleep(Long.MAX_VALUE);
//获取参数
public void getData() throws KeeperException, InterruptedException {
byte[] data = zkClient.getData("/eclipse", true, null);
System.out.println(new String(data));
Thread.sleep(Long.MAX_VALUE);
//删除znode
public void deleteZnode() throws InterruptedException, KeeperException {
//参数2:指定要删除的版本，-1表示删除所有版本
zkClient.delete("/eclipse", -1);
//设置参数
public void setData() throws Exception {
//要注意，这里的/zookeeper 要在zookeeper中的节点中有
zkClient.setData("/zookeeper", "imissyou angelababy".getBytes(), -1);
byte[] data = zkClient.getData("/zookeeper", false, null);
System.out.println(new String(data));
cn.com.toto.
import java.util.List;
java.util.concurrent.CountDownL
import org.apache.zookeeper.CreateMode;
org.apache.zookeeper.WatchedE
org.apache.zookeeper.W
org.apache.zookeeper.Watcher.Event.KeeperS
import org.apache.zookeeper.ZooDefs.Ids;
org.apache.zookeeper.ZooK
import org.apache.zookeeper.data.Stat;
import com.sun.org.apache.bcel.internal.generic.NEW;
public class TestZKclient {
static ZooKeeper zk = null;
public static void main(String[] args) throws Exception {
final CountDownLatch countDownLatch = new CountDownLatch(1);
zk = new ZooKeeper("hadoop:2181",2000,new Watcher() {
public void process(WatchedEvent event) {
if (event.getState() == KeeperState.SyncConnected) {
countDownLatch.countDown();
System.out.println(event.getPath());
System.out.println(event.getType());
zk.getChildren("/zookeeper", true);
} catch (Exception e) {
e.printStackTrace();
countDownLatch.await();
zk.create("/myboys", "丑陋型".getBytes("UTF-8"), Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
zk.close();
byte[] data = zk.getData("/myboys", true,
System.out.println(new String(data,"UTF-8"));
Thread.sleep(Long.MAX_VALUE);
zk.create("/myboys/wangkai",
"测试型".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
zk.close();
List&String& children =
zk.getChildren("/myboys", true);
for(String child : children) {
System.out.println(child);
/**zk.delete("/myboys/wangkai",
/**zk.setData("/myboys",
"fasdfasdf".getBytes(), -1);**/
byte[] data = zk.getData("/myboys", true,
System.out.println(new String(data,"UTF-8"));
stat = zk.exists("/mywives", true);
System.out.println(stat
== null ? "确实不存在" : "存在");
zk.close();
9.5 其它网络参考资料
拷贝ZooKeeper安装目录下的zookeeper.x.x.x.jar文件到项目的classpath路径下.
创建连接和回调接口
首先需要创建ZooKeeper对象, 后续的一切操作都是基于该对象进行的.
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException
以下为各个参数的详细说明:
connectString. zookeeper
server列表, 以逗号隔开. ZooKeeper对象初始化后, 将从server列表中选择一个server,
并尝试与其建立连接. 如果连接建立失败, 则会从列表的剩余项中选择一个server, 并再次尝试建立连接.
sessionTimeout. 指定连接的超时时间.
watcher. 事件回调接口.
注意, 创建ZooKeeper对象时, 只要对象完成初始化便立刻返回. 建立连接是以异步的形式进行的, 当连接成功建立后, 会回调watcher的process方法. 如果想要同步建立与server的连接, 需要自己进一步封装.
public class ZKConnection {
* server列表, 以逗号分割
protected String hosts = "localhost:4180,localhost:4181,localhost:4182";
* 连接的超时时间, 毫秒
private static final int SESSION_TIMEOUT = 5000;
private CountDownLatch connectedSignal = new CountDownLatch(1);
protected ZooK
* 连接zookeeper server
public void connect() throws Exception {
zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new ConnWatcher());
// 等待连接完成
connectedSignal.await();
public class ConnWatcher implements Watcher {
public void process(WatchedEvent event) {
// 连接建立, 回调process接口时, 其event.getState()为KeeperState.SyncConnected
if (event.getState() == KeeperState.SyncConnected) {
// 放开闸门, wait在connect方法上的线程将被唤醒
connectedSignal.countDown();
ZooKeeper对象的create方法用于创建znode.
String create(String path, byte[] data, List acl, CreateMode createMode);
以下为各个参数的详细说明:
path. znode的路径.
data. 与znode关联的数据.
acl. 指定权限信息, 如果不想指定权限, 可以传入Ids.OPEN_ACL_UNSAFE.
指定znode类型.
CreateMode是一个枚举类, 从中选择一个成员传入即可. 关于znode类型的详细说明, 可参考本人的上一篇博文.
* 创建临时节点
public void create(String nodePath, byte[] data) throws Exception {
zk.create(nodePath, data, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
获取子node列表
ZooKeeper对象的getChildren方法用于获取子node列表.
List getChildren(String path, boolean watch);
watch参数用于指定是否监听path node的子node的增加和删除事件, 以及path node本身的删除事件.
判断znode是否存在
ZooKeeper对象的exists方法用于判断指定znode是否存在.
Stat exists(String path, boolean watch);
watch参数用于指定是否监听path node的创建, 删除事件, 以及数据更新事件. 如果该node存在, 则返回该node的状态信息, 否则返回null.
获取node中关联的数据
ZooKeeper对象的getData方法用于获取node关联的数据.
byte[] getData(String path, boolean watch, Stat stat);
watch参数用于指定是否监听path node的删除事件, 以及数据更新事件, 注意, 不监听path node的创建事件, 因为如果path node不存在, 该方法将抛出KeeperException.NoNodeException异常.
stat参数是个传出参数, getData方法会将path node的状态信息设置到该参数中.
更新node中关联的数据
ZooKeeper对象的setData方法用于更新node关联的数据.
Stat setData(final String path, byte data[], int version);
data为待更新的数据.
version参数指定要更新的数据的版本, 如果version和真实的版本不同, 更新操作将失败. 指定version为-1则忽略版本检查.返回path node的状态信息.
ZooKeeper对象的delete方法用于删除znode.
void delete(final String path, int version);
version参数的作用同setData方法.
请查看ZooKeeper对象的API文档.
需要注意的几个地方
znode中关联的数据不能超过1M. zookeeper的使命是分布式协作, 而不是数据存储.
getChildren, getData,
exists方法可指定是否监听相应的事件. 而create, delete, setData方法则会触发相应的事件的发生.
以上介绍的几个方法大多存在其异步的重载方法, 具体请查看API说明.
版权声明：本文内容由互联网用户自发贡献，本社区不拥有所有权，也不承担相关法律责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件至： 进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容。
用云栖社区APP，舒服~
【云栖快讯】数据库技术天团集体亮相，分享一线生产实践经验，告诉你踩过的坑、走过的路，都是老司机，靠谱！干货分享，不可错过！&&
通过机器学习和数据建模发现潜在的入侵和攻击威胁，帮助客户建设自己的安全监控和防御体系，从而解决因网络攻击导致企业...
API 网关（API Gateway），提供高性能、高可用的 API 托管服务，帮助用户对外开放其部署在 ECS...
在云上签发Symantec、WoSign、CFCA证书，实现网站HTTPS化，使网站可信，防劫持、防篡改、防监听...
为您提供简单高效、处理能力可弹性伸缩的计算服务，帮助您快速构建更稳定、安全的应用，提升运维效率，降低 IT 成本...
2017杭州云栖大会火热抢票
Loading...