https://zookeeper.apache.org/doc/r3.4.14/zookeeperProgrammers.html#sc_WatchRememberThese
zookeeper
提供了数据的发布/订阅
功能,多个订阅者可同时监听某一特定主题对象,当该主题对象的自身状态发生变化时例如节点内容改变、节点下的子节点列表改变等,会实时、主动通知所有订阅者
zookeeper
采用了 Watcher
机制实现数据的发布订阅功能。该机制在被订阅对象发生变化时会异步通知客户端,因此客户端不必在 Watcher
注册后轮询阻塞,从而减轻了客户端压力
watcher
机制事件上与观察者模式类似,也可看作是一种观察者模式在分布式场景下的实现方式
watcher
实现由三个部分组成
zookeeper
服务端
zookeeper
客户端
客户端的ZKWatchManager对象
客户端首先将 Watcher
注册到服务端,同时将 Watcher
对象保存到客户端的watch
管理器中。当Zookeeper
服务端监听的数据状态发生变化时,服务端会主动通知客户端,接着客户端的 Watch
管理器会
特性 | 说明 |
---|---|
一次性 | watcher 是一次性的,一旦被触发就会移除,再次使用时需要重新注册 |
客户端顺序回调 | watcher 回调是顺序串行执行的,只有回调后客户端才能看到最新的数据状态。一个watcher 回调逻辑不应该太多,以免影响别的watcher 执行 |
轻量级 | WatchEvent 是最小的通信单位,结构上只包含通知状态、事件类型和节点路径,并不会告诉数据节点变化前后的具体内容 |
时效性 | watcher 只有在当前session 彻底失效时才会无效,若在session 有效期内快速重连成功,则watcher 依然存在,仍可接收到通知; |
watcher接口设计
Watcher
是一个接口,任何实现了Watcher
接口的类就算一个新的Watcher
。Watcher
内部包含了两个枚举类:KeeperState
、EventType
KeeperState
是客户端与服务端连接状态发生变化时对应的通知类型。路径为org.apache.zookeeper.Watcher.EventKeeperState
,是一个枚举类,其枚举属性如下:
枚举属性 | 说明 |
---|---|
SyncConnected |
客户端与服务器正常连接时 |
Disconnected |
客户端与服务器断开连接时 |
Expired |
会话session 失效时 |
AuthFailed |
身份认证失败时 |
EventType
是数据节点znode
发生变化时对应的通知类型。EventType
变化时KeeperState
永远处于SyncConnected
通知状态下;当keeperState
发生变化时,EventType
永远为None
。其路径为org.apache.zookeeper.Watcher.Event.EventType
,是一个枚举类,枚举属性如下:
枚举属性 | 说明 |
---|---|
None |
无 |
NodeCreated |
Watcher 监听的数据节点被创建时 |
NodeDeleted |
Watcher 监听的数据节点被删除时 |
NodeDataChanged |
Watcher 监听的数据节点内容发生更改时(无论数据是否真的变化) |
NodeChildrenChanged |
Watcher 监听的数据节点的子节点列表发生变更时 |
注意:客户端接收到的相关事件通知中只包含状态以及类型等信息,不包含节点变化前后的具体内容,变化前的数据需业务自身存储,变化后的数据需要调用get
等方法重新获取
上面讲到zookeeper
客户端连接的状态和zookeeper
对znode
节点监听的事件类型,下面我们来讲解如何建立zookeeper
的watcher
监听。在zookeeper
中采用zk.getChildren(path,watch)、zk.exists(path,watch)、zk.getData(path,watcher,stat)
这样的方式来为某个znode
注册监听 。
下表以node-x
节点为例,说明调用的注册方法和可用监听事件间的关系:
注册方式 | created | childrenChanged | Changed | Deleted |
---|---|---|---|---|
zk.exists("/node-x",watcher) |
可监控 | 可监控 | 可监控 | |
zk.getData("/node-x",watcher) |
可监控 | 可监控 | ||
zk.getChildren("/node-x",watcher) |
可监控 | 可监控 |
注册watcher的方法
KeeperState
:通知状态
SyncConnected
:客户端与服务器正常连接时
Disconnected
:客户端与服务器断开连接时
Expired
:会话session
失效时
AuthFailed
:身份认证失败时
事件类型为:None
public class ZkConnectionWatcher implements Watcher { @Override public void process(WatchedEvent watchedEvent) { Event.KeeperState state = watchedEvent.getState(); if(state == Event.KeeperState.SyncConnected){ // 正常 System.out.println("正常连接"); }else if (state == Event.KeeperState.Disconnected){ // 可以用Windows断开虚拟机网卡的方式模拟 // 当会话断开会出现,断开连接不代表不能重连,在会话超时时间内重连可以恢复正常 System.out.println("断开连接"); }else if (state == Event.KeeperState.Expired){ // 没有在会话超时时间内重新连接,而是当会话超时被移除的时候重连会走进这里 System.out.println("连接过期"); }else if (state == Event.KeeperState.AuthFailed){ // 在操作的时候权限不够会出现 System.out.println("授权失败"); } countDownLatch.countDown(); } private static final String IP = "192.168.133.133:2181" ; private static CountDownLatch countDownLatch = new CountDownLatch(1); public static void main(String[] args) throws Exception { // 5000为会话超时时间 ZooKeeper zooKeeper = new ZooKeeper(IP, 5000, new ZkConnectionWatcher()); countDownLatch.await(); // 模拟授权失败 zooKeeper.addAuthInfo("digest1","itcast1:123451".getBytes()); byte[] data = zooKeeper.getData("/hadoop", false, null); System.out.println(new String(data)); TimeUnit.SECONDS.sleep(50); } }
exists
exists(String path, boolean b)
exists(String path, Watcher w)
NodeCreated
:节点创建
NodeDeleted
:节点删除
NodeDataChanged
:节点内容
public class EventTypeTest { private static final String IP = "192.168.133.133:2181"; private static CountDownLatch countDownLatch = new CountDownLatch(1); private static ZooKeeper zooKeeper; // 采用zookeeper连接创建时的监听器 public static void exists1() throws Exception{ zooKeeper.exists("/watcher1",true); } // 自定义监听器 public static void exists2() throws Exception{ zooKeeper.exists("/watcher1",(WatchedEvent w) -> { System.out.println("自定义" + w.getType()); }); } // 演示使用多次的监听器 public static void exists3() throws Exception{ zooKeeper.exists("/watcher1", new Watcher() { @Override public void process(WatchedEvent watchedEvent) { try { System.out.println("自定义的" + watchedEvent.getType()); } finally { try { zooKeeper.exists("/watcher1",this); } catch (Exception e) { e.printStackTrace(); } } } }); } // 演示一节点注册多个监听器 public static void exists4() throws Exception{ zooKeeper.exists("/watcher1",(WatchedEvent w) -> { System.out.println("自定义1" + w.getType()); }); zooKeeper.exists("/watcher1", new Watcher() { @Override public void process(WatchedEvent watchedEvent) { try { System.out.println("自定义2" + watchedEvent.getType()); } finally { try { zooKeeper.exists("/watcher1",this); } catch (Exception e) { e.printStackTrace(); } } } }); } // 测试 public static void main(String[] args) throws Exception { zooKeeper = new ZooKeeper(IP, 5000, new ZKWatcher()); countDownLatch.await(); exists4(); TimeUnit.SECONDS.sleep(50); } static class ZKWatcher implements Watcher{ @Override public void process(WatchedEvent watchedEvent) { countDownLatch.countDown(); System.out.println("zk的监听器" + watchedEvent.getType()); } } }
getData(String path, boolean b, Stat stat)
getData(String path, Watcher w, Stat stat)
NodeDeleted
:节点删除
NodeDataChange
:节点内容发生变化
getChildren
getChildren(String path, boolean b)
getChildren(String path, Watcher w)
NodeChildrenChanged
:子节点发生变化
NodeDeleted
:节点删除
原文:https://www.cnblogs.com/lemon-flm/p/14605237.html