目录

前置知识：

前提：启动集群

创建项目基础

pom.xml

log4j.properties

创建Zookeeper客户端

主要API

连接到ZooKeeper集合

创建Znode

Exists - 检查Znode的存在

getData方法

setData方法

getChildren方法

删除Znode

监听器

监听器触发有如下几种情况：

关心的监听器有四个地方可以触发：

持久化监听

getData()

getChildren()

exists()

摆烂

来源：

前置知识：

监控原理：

（1）首先要有一个main()线程

（2）在main线程中创建Zookeeper客户端，这时就会创建两个线程，一个负责网络连接通信(connet)，一个负责监听(listener) 。

（3）通过connect线程将注册的监听事件发送给Zookeeper。

（4）在Zookeeper的注册监听器列表中将注册的监听事件添加到列表中。

（5）Zookeeper监听到有数据或路径变化，就会将这个消息发送给listener线程。

（6） listener线程内部调用了process()方法。

前提：启动集群

创建项目基础

pom.xml

junitjunitRELEASEorg.apache.logging.log4jlog4j-core2.8.2org.apache.zookeeperzookeeper3.5.7

log4j.properties

log4j.rootLogger=INFO, stdout  
log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n  
log4j.appender.logfile=org.apache.log4j.FileAppender  
log4j.appender.logfile.File=target/spring.log  
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n 

创建Zookeeper客户端

//connectString 中,前后不能有空格
private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
//sessionTimeout 连接超时时长 毫秒
private int sessionTimeout = 2000;
//watch 监听器
private ZooKeeper zkClient = null;

ctrl+t 显示函数参数

主要API

连接到ZooKeeper集合

ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)

• connectionString - ZooKeeper集合主机。

• sessionTimeout - 会话超时（以毫秒为单位）。

• watcher - 实现“监视器”界面的对象。ZooKeeper集合通过监视器对象返回连接状态。

例子：

创建Znode

create(String path, byte[] data, List acl, CreateMode createMode)

• path - Znode路径。例如，/myapp1，/myapp2，/myapp1/mydata1，myapp2/mydata1/myanothersubdata

• data - 要存储在指定znode路径中的数据

• acl - 要创建的节点的访问控制列表。ZooKeeper API提供了一个静态接口 ZooDefs.Ids 来获取一些基本的acl列表。例如，ZooDefs.Ids.OPEN_ACL_UNSAFE返回打开znode的acl列表。

OPEN_ACL_UNSAFE  : 完全开放的ACL，任何连接的客户端都可以操作该属性znode

CREATOR_ALL_ACL : 只有创建者才有ACL权限

READ_ACL_UNSAFE：只能读取ACL

• createMode - 节点的类型，即临时，顺序或两者。这是一个枚举。

例子

import java.io.IOException;import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;public class ZKCreate {// create static instance for zookeeper class.private static ZooKeeper zk;// create static instance for ZooKeeperConnection class.private static ZooKeeperConnection conn;// Method to create znode in zookeeper ensemblepublic static void create(String path, byte[] data) throws KeeperException,InterruptedException {zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);}public static void main(String[] args) {// znode pathString path = "/MyFirstZnode"; // Assign path to znode// data in byte arraybyte[] data = "My first zookeeper app".getBytes(); // Declare datatry {conn = new ZooKeeperConnection();zk = conn.connect("localhost");create(path, data); // Create the data to the specified pathconn.close();} catch (Exception e) {System.out.println(e.getMessage()); //Catch error message}}
}

Exists - 检查Znode的存在

exists(String path, boolean watcher)

• path- Znode路径

• watcher - 布尔值，用于指定是否监视指定的znode

/*
创建一个新的Java应用程序来检查ZooKeeper API的“exists”功能。创建文件“ZKExists.java”。在main方法中，使用“ZooKeeperConnection”对象创建ZooKeeper对象“zk”。然后，使用自定义“path”调用“zk”对象的“exists”方法。
*/
import java.io.IOException;import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.data.Stat;public class ZKExists {private static ZooKeeper zk;private static ZooKeeperConnection conn;// Method to check existence of znode and its status, if znode is available.public static Stat znode_exists(String path) throwsKeeperException,InterruptedException {return zk.exists(path, true);}public static void main(String[] args) throws InterruptedException,KeeperException {String path = "/MyFirstZnode"; // Assign znode to the specified pathtry {conn = new ZooKeeperConnection();zk = conn.connect("localhost");Stat stat = znode_exists(path); // Stat checks the path of the znodeif(stat != null) {System.out.println("Node exists and the node version is " +stat.getVersion());} else {System.out.println("Node does not exists");}} catch(Exception e) {System.out.println(e.getMessage()); // Catches error messages}}
}

getData方法

获取附加在指定znode中的数据及其状态。

getData(String path, Watcher watcher, Stat stat)

• path - Znode路径。

• watcher - 监视器类型的回调函数。当指定的znode的数据改变时，ZooKeeper集合将通过监视器回调进行通知。这是一次性通知。

• stat - 返回znode的元数据。

/*
创建一个新的Java应用程序来了解ZooKeeper API的 getData 功能。创建文件 ZKGetData.java 。在main方法中，使用 ZooKeeperConnection 对象创建一个ZooKeeper对象 zk 。然后，使用自定义路径调用zk对象的 getData 方法。
*/
import java.io.IOException;
import java.util.concurrent.CountDownLatch;import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.data.Stat;public class ZKGetData {private static ZooKeeper zk;private static ZooKeeperConnection conn;public static Stat znode_exists(String path) throws KeeperException,InterruptedException {return zk.exists(path,true);}public static void main(String[] args) throws InterruptedException, KeeperException {String path = "/MyFirstZnode";final CountDownLatch connectedSignal = new CountDownLatch(1);try {conn = new ZooKeeperConnection();zk = conn.connect("localhost");Stat stat = znode_exists(path);if(stat != null) {byte[] b = zk.getData(path, new Watcher() {public void process(WatchedEvent we) {if (we.getType() == Event.EventType.None) {switch(we.getState()) {case Expired:connectedSignal.countDown();break;}} else {String path = "/MyFirstZnode";try {byte[] bn = zk.getData(path,false, null);String data = new String(bn,"UTF-8");System.out.println(data);connectedSignal.countDown();} catch(Exception ex) {System.out.println(ex.getMessage());}}}}, null);String data = new String(b, "UTF-8");System.out.println(data);connectedSignal.await();} else {System.out.println("Node does not exists");}} catch(Exception e) {System.out.println(e.getMessage());}}
}

setData方法

setData(String path, byte[] data, int version)

• path- Znode路径

• data - 要存储在指定znode路径中的数据。

• version- znode的当前版本。每当数据更改时，ZooKeeper会更新znode的版本号。

/*
创建一个新的Java应用程序来了解ZooKeeper API的 setData 功能。创建文件 ZKSetData.java 。在main方法中，使用 ZooKeeperConnection 对象创建一个ZooKeeper对象 zk 。然后，使用指定的路径，新数据和节点版本调用 zk 对象的 setData 方法。
*/
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;import java.io.IOException;public class ZKSetData {private static ZooKeeper zk;private static ZooKeeperConnection conn;// Method to update the data in a znode. Similar to getData but without watcher.public static void update(String path, byte[] data) throwsKeeperException,InterruptedException {zk.setData(path, data, zk.exists(path,true).getVersion());}public static void main(String[] args) throws InterruptedException,KeeperException {String path= "/MyFirstZnode";byte[] data = "Success".getBytes(); //Assign data which is to be updated.try {conn = new ZooKeeperConnection();zk = conn.connect("localhost");update(path, data); // Update znode data to the specified path} catch(Exception e) {System.out.println(e.getMessage());}}
}

getChildren方法

获取特定znode的所有子节点。

getChildren(String path, Watcher watcher)

• path - Znode路径。

• watcher - 监视器类型的回调函数。当指定的znode被删除或znode下的子节点被创建/删除时，ZooKeeper集合将进行通知。这是一次性通知。

import java.io.IOException;
import java.util.*;import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.data.Stat;public class ZKGetChildren {private static ZooKeeper zk;private static ZooKeeperConnection conn;// Method to check existence of znode and its status, if znode is available.public static Stat znode_exists(String path) throws KeeperException,InterruptedException {return zk.exists(path,true);}public static void main(String[] args) throws InterruptedException,KeeperException {String path = "/MyFirstZnode"; // Assign path to the znodetry {conn = new ZooKeeperConnection();zk = conn.connect("localhost");Stat stat = znode_exists(path); // Stat checks the pathif(stat!= null) {//“getChildren" method- get all the children of znode.It has twoargs, path and watchList  children = zk.getChildren(path, false);for(int i = 0; i < children.size(); i++)System.out.println(children.get(i)); //Print children's} else {System.out.println("Node does not exists");}} catch(Exception e) {System.out.println(e.getMessage());}}}

删除Znode

delete(String path, int version)

• path - Znode路径。

• version - znode的当前版本。

/*
创建文件 ZKDelete.java 。在main方法中，使用 ZooKeeperConnection 对象创建一个ZooKeeper对象 zk 。然后，使用指定的路径和版本号调用 zk 对象的 delete 方法。
*/
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException;public class ZKDelete {private static ZooKeeper zk;private static ZooKeeperConnection conn;// Method to check existence of znode and its status, if znode is available.public static void delete(String path) throws KeeperException,InterruptedException {zk.delete(path,zk.exists(path,true).getVersion());}public static void main(String[] args) throws InterruptedException,KeeperException {String path = "/MyFirstZnode"; //Assign path to the znodetry {conn = new ZooKeeperConnection();zk = conn.connect("localhost");delete(path); //delete the node with the specified path} catch(Exception e) {System.out.println(e.getMessage()); // catches error messages}}
}

监听器

监听器触发有如下几种情况：

        1、通过客户端去查询或者创建、修改zookeeper上的节点都会去触发一次监听器
        2、再去查询zookeeper的节点数据、子节点、节点存在与否的时候也会注册监听器，那么此时这个监听器也可以让他们三个取用

关心的监听器有四个地方可以触发：

        ZooKeeper zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent event) {//        1、构建zookeeper对象之后 调用zookeeper的方法时候监听器会被触发---不常用System.out.println("zkClient的监听器被触发了");}});}//    2、getData()---获取节点数据的  里面注册监听器
//      被触发--当节点的数据发生变化的时候byte[] data = zkClient.getData("/java", true, new Stat());//true代表使用zookeeper监听器
//    自定义监视器
byte[] data = zkClient.getData("/java", new Watcher() {@Overridepublic void process(WatchedEvent event) {System.out.println("getData自定义的监听器被触发了");}
}, new Stat());// 3、getChildren()----获取节点的子节点信息
//    被触发---当前节点的子节点列表发生变化会触发List children = zkClient.getChildren("/java", true);//     4、exists() ----判断节点存不存在
//      被触发---当节点被创建或者删除会触发
//      或者当节点的数据变化也会去触发Stat stat = zkClient.exists("/atguigu", false);

如果watcher =true,就走new zookeeper里面的watcher ；如果watcher = new watchar 自己定义了一个watcher ，那么就走自己定义的。

不监听watcher =false

持久化监听

getData()

public class MoreTimesWatcher {static  ZooKeeper zkClient;public static void main(String[] args) throws Exception {zkClient= new ZooKeeper("node1:2181,node2:2181,node3:2181", 2000, new Watcher() {@Overridepublic void process(WatchedEvent event) {/*** 如果想持久监听 我们需要再监听器被触发之后再次去注册一下这个监听器即可*/try {byte[]  data = zkClient.getData("/java", true, new Stat());System.out.println(new String(data));} catch (Exception e) {e.printStackTrace();}}});byte[] data = zkClient.getData("/java", true, new Stat());//监听一次以后再去执行process()方法重新调用此监听器进行监听System.out.println(new String(data));Thread.sleep(Long.MAX_VALUE);}
}

getChildren()

public class NodeListWatcher {static ZooKeeper zkClient;public static void main(String[] args) throws Exception {zkClient= new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent event) {try {List children = zkClient.getChildren("/java", true);System.out.println(children);} catch (KeeperException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}});List children = zkClient.getChildren("/java", true);System.out.println(children);Thread.sleep(Long.MAX_VALUE);}
}

exists()

/*** 判断节点是否存在的一个监听器：*    两种情况：*      1、节点创建或删除*      2、节点数据发生变化也会触发*/
public class ExistsWatcher {static ZooKeeper zkClient;public static void main(String[] args) throws Exception {zkClient= new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent event) {try {Stat stat  = zkClient.exists("/java", true);System.out.println(stat==null?"节点不存在":"节点存在");} catch (KeeperException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}});Stat stat = zkClient.exists("/java", true);System.out.println(stat==null?"节点不存在":"节点存在");Thread.sleep(Long.MAX_VALUE);}
}

摆烂

不学了，分布式tm太难了.

package com.atguigu.zk;import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;public class ZookeeperWatch implements Watcher {/** 定义原子变量* 线程安全，使用于高并发* */AtomicInteger seq = new AtomicInteger();/** 定义session失效时间 */private static final int SESSION_TIMEOUT = 10000;/** zookeeper服务器地址 */private static final String CONNECTION_ADDR = "192.168.145.134:2181,192.168.145.135:2181,192.168.145.136:2181";/** zk父路径设置 */private static final String PARENT_PATH = "/m";/** zk子路径设置 */private static final String CHILDREN_PATH = "/m/c";/** 进入标识 */private static final String LOG_PREFIX_OF_MAIN = "【Main】";/** zk变量 */private ZooKeeper zk = null;/**用于等待zookeeper连接建立之后 通知阻塞程序继续向下执行 */private CountDownLatch connectedSemaphore = new CountDownLatch(1);//创建zk的连接/*** 创建ZK连接* @param connectAddr ZK服务器地址列表* @param sessionTimeout Session超时时间*/public void createConnection(String connectAddr, int sessionTimeout) {this.releaseConnection();try {//this表示把当前对象进行传递到其中去（也就是在主函数里实例化的new ZooKeeperWatcher()实例对象）zk = new ZooKeeper(connectAddr, sessionTimeout, this);System.out.println(LOG_PREFIX_OF_MAIN + "开始连接ZK服务器");connectedSemaphore.await();} catch (Exception e) {e.printStackTrace();}}/*** 关闭ZK连接*/public void releaseConnection() {if (this.zk != null) {try {this.zk.close();} catch (InterruptedException e) {e.printStackTrace();}}}/*** 创建节点* @param path 节点路径* @param data 数据内容* @return*/public boolean createPath(String path, String data, boolean needWatch) {try {//设置监控(由于zookeeper的监控都是一次性的所以 每次必须设置监控,若needWatch为false,则不会走process)this.zk.exists(path, needWatch);System.out.println(LOG_PREFIX_OF_MAIN + "节点创建成功, Path: " +this.zk.create(	/**路径*/path,/**数据*/data.getBytes(),/**所有可见*/ZooDefs.Ids.OPEN_ACL_UNSAFE,/**永久存储*/CreateMode.PERSISTENT ) +", content: " + data);} catch (Exception e) {e.printStackTrace();return false;}return true;}/*** 读取指定节点数据内容* @param path 节点路径* @return*/public String readData(String path, boolean needWatch) {try {System.out.println("读取数据操作...");return new String(this.zk.getData(path, needWatch, null));} catch (Exception e) {e.printStackTrace();return "";}}/*** 更新指定节点数据内容* @param path 节点路径* @param data 数据内容* @return*/public boolean writeData(String path, String data) {try {this.zk.exists(path, true);System.out.println(LOG_PREFIX_OF_MAIN + "更新数据成功，path：" + path + ", stat: " +this.zk.setData(path, data.getBytes(), -1));} catch (Exception e) {e.printStackTrace();return false;}return true;}/*** 删除指定节点** @param path*            节点path*/public void deleteNode(String path) {try {this.zk.delete(path, -1);System.out.println(LOG_PREFIX_OF_MAIN + "删除节点成功，path：" + path);} catch (Exception e) {e.printStackTrace();}}/*** 判断指定节点是否存在* @param path 节点路径*/public Stat exists(String path, boolean needWatch) {try {return this.zk.exists(path, needWatch);} catch (Exception e) {e.printStackTrace();return null;}}/*** 获取子节点* @param path 节点路径*/private List getChildren(String path, boolean needWatch) {try {System.out.println("读取子节点操作...");return this.zk.getChildren(path, needWatch);} catch (Exception e) {e.printStackTrace();return null;}}/*** 删除所有节点*/public void deleteAllTestPath(boolean needWatch) {if(this.exists(CHILDREN_PATH, needWatch) != null){this.deleteNode(CHILDREN_PATH);}if(this.exists(PARENT_PATH, needWatch) != null){this.deleteNode(PARENT_PATH);}}/*** 收到来自Server的Watcher通知后的处理。*/@Overridepublic void process(WatchedEvent event) {System.out.println("进入 process 。。。。。event = " + event);try {Thread.sleep(200);} catch (InterruptedException e) {e.printStackTrace();}if (event == null) {return;}// 连接状态Event.KeeperState keeperState = event.getState();// 事件类型Event.EventType eventType = event.getType();// 受影响的pathString path = event.getPath();//原子对象seq 记录进入process的次数String logPrefix = "【Watcher-" + this.seq.incrementAndGet() + "】";System.out.println(logPrefix + "收到Watcher通知");System.out.println(logPrefix + "连接状态:\t" + keeperState.toString());System.out.println(logPrefix + "事件类型:\t" + eventType.toString());if (Event.KeeperState.SyncConnected == keeperState) {// 成功连接上ZK服务器if (Event.EventType.None == eventType) {System.out.println(logPrefix + "成功连接上ZK服务器");connectedSemaphore.countDown();}//创建节点else if (Event.EventType.NodeCreated == eventType) {System.out.println(logPrefix + "节点创建");try {Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}}//更新节点else if (Event.EventType.NodeDataChanged == eventType) {System.out.println(logPrefix + "节点数据更新");try {Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}}//更新子节点else if (Event.EventType.NodeChildrenChanged == eventType) {System.out.println(logPrefix + "子节点变更");try {Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}}//删除节点else if (Event.EventType.NodeDeleted == eventType) {System.out.println(logPrefix + "节点 " + path + " 被删除");}else ;}else if (Event.KeeperState.Disconnected == keeperState) {System.out.println(logPrefix + "与ZK服务器断开连接");}else if (Event.KeeperState.AuthFailed == keeperState) {System.out.println(logPrefix + "权限检查失败");}else if (Event.KeeperState.Expired == keeperState) {System.out.println(logPrefix + "会话失效");}else ;System.out.println("--------------------------------------------");}/*** 方法名称：测试zookeeper监控
* 概要说明：主要测试watch功能
* @param args* @throws Exception*/public static void main(String[] args) throws Exception {//建立watcher //当前客户端可以称为一个watcher 观察者角色ZookeeperWatch zkWatch = new ZookeeperWatch();//创建连接zkWatch.createConnection(CONNECTION_ADDR, SESSION_TIMEOUT);//System.out.println(zkWatch.zk.toString());Thread.sleep(1000);// 清理节点//   zkWatch.deleteAllTestPath(false);//-----------------第一步: 创建父节点 /p ------------------------//if (zkWatch.createPath(PARENT_PATH, System.currentTimeMillis() + "", false)) {Thread.sleep(1000);zkWatch.writeData(PARENT_PATH, System.currentTimeMillis() + "");zkWatch.createPath(CHILDREN_PATH, System.currentTimeMillis() + "", true);zkWatch.getChildren(PARENT_PATH, true);/*  //-----------------第二步: 读取节点 /p 和    读取/p节点下的子节点(getChildren)的区别 --------------//// 读取数据zkWatch.readData(PARENT_PATH, true);// 读取子节点(监控childNodeChange事件)// 更新数据Thread.sleep(1000);// 创建子节点zkWatch.createPath(CHILDREN_PATH, System.currentTimeMillis() + "", true);//-----------------第三步: 建立子节点的触发 --------------//
//			zkWatch.createPath(CHILDREN_PATH + "/c1", System.currentTimeMillis() + "", true);
//			zkWatch.createPath(CHILDREN_PATH + "/c1/c2", System.currentTimeMillis() + "", true);//-----------------第四步: 更新子节点数据的触发 --------------////在进行修改之前，我们需要watch一下这个节点：Thread.sleep(1000);zkWatch.readData(CHILDREN_PATH, true);zkWatch.writeData(CHILDREN_PATH, System.currentTimeMillis() + "");*/}//  Thread.sleep(10000);// 清理节点// zkWatch.deleteAllTestPath(false);Thread.sleep(10000);System.out.println("===========关闭zk");zkWatch.releaseConnection();}}

(16条消息) Zookeeper Watcher详解_zookeeperwatcher_athena_lee_的博客-CSDN博客

Zookeeper全解析——Paxos作为灵魂 (douban.com)

(17条消息) 史上最全的Zookeeper原理详解(万字长文)_雷恩Layne的博客-CSDN博客

 (16条消息) 什么是分布式锁？几种分布式锁分别是怎么实现的？_Java后端架构猛猛的博客-CSDN博客

来源：

Zookeeper 应用程序_w3cschool

尚硅谷

(17条消息) zookeeper的监听器原理_zookeeper监听机制原理_阳哥赚钱很牛的博客-CSDN博客