首页 > > 程序设计 > Java技术 >

Watch机制

2018-06-18 03:24:10来源：未知阅读 ()

一、整体流程

1、客户端注册Watcher

2、服务端处理Watcher

3、客户端回调Watcher

二、Watcher 通知状态与事件类型 state type path

WatcheEvent 只有三个属性 state type path

不会告诉原始数据内容和更新内容，需要client再次主动获取最新信息

三、客户端：ZKWatchManager

创建一个Zookeeper实例时，构造方法中传入一个默认的Watcher，会一直保存在ZKWatchManager的defaultWatcher中，ZKWatchManager实现了ClientWatchManager接口，实现了materialize方法

四、服务端：WatchManager

watchTable和watch2Paths

两个存储结构，从两个维度对Watcher进行存储

五、具体流程

1、客户端注册Watcher getData()

客户端发起一个getData方法，注册一个Watcher

1、封装一个Watcher的注册信息WatchRegistration，保存数据节点的路径和Watcher的对应关系

2、将WatchRegistration封装到Packet中，放入发送队列中等待客户端发送(outgoingQueue)

3、客户端向服务端发送这个请求，同时等待请求的返回，由SendThread线程的readResponse方法负责接收来自服务端的相应，finishPacket方法从Packet中取出Watcher，并注册到ZKWatchManager中，保存到dataWatches中，是一个Map<String,Set<Watcher>>的数据结构

Packet在网络传输中只对requestHeader和request两个属性进行序列化，WatchRegistration没有序列化到底层数组中，不进行网络传输

2、服务端处理Watcher getData() setData()

1、服务端收到客户端的请求后，判断当前是否需要注册Watcher（简单的Boolean类型），需要的话，将当前ServerCnxn对象和数据节点路径传入getData方法中

2、ServerCnxn对象和数据节点路径最终会被存储在WatchManager的watchTable和watch2Paths中

在setData方法中，对应的数据节点的内容发生变更，调用WatchManager的triggerWatch方法触发相关事件，封装WatchedEvent，查询Watcher后从存储结构中剔除，调用ServerCnxn的对应方法process方法，将WatchedEvent包装成WatcherEvent，向客户端发送通知 sendResponse

借助当前客户端连接的ServerCnxn对象来实现对客户端的WatchedEvent传递，真正的Watcher回调与义务逻辑都在客户端执行

3、客户端回调Watcher

1、SendThread接受到事件通知，调用readResponse方法，EventThread.queueEvent(watchedEvent)将对象交给eventThread线程来进行处理

2、eventThread.queueEvent方法，根据通知事件类型状态path，从ZKWatchManager中取出Watcher，同时剔除（materialize方法）

3、取到相关所有的Watcher后，将其放入WaitingEvents这个队列中，WaitingEvents是一个待处理Watcher的队列，通过EventThread的run方法对该队列进行处理，核心processEvent 方法中 watcher.process方法实现watcher方法的调用

Watcher：Watcher是用于监听节点，session 状态的，比如getData对数据节点a设置了watcher，那么当a的数据内容发生改变时，客户端会收到NodeDataChanged通知，然后进行watcher的回调。
AsyncCallback:AsyncCallback是在以异步方式使用 ZooKeeper API 时，用于处理返回结果的。例如：getData同步调用的版本是：byte[] getData(String path, boolean watch,Stat stat)，异步调用的版本是：void getData(String path,Watcher watcher,AsyncCallback.DataCallback cb,Object ctx)，可以看到，前者是直接返回获取的结果，后者是通过AsyncCallback回调处理结果的。