Watcher机制(一)

简介: 本文深入分析ZooKeeper的Watcher机制核心类与源码,涵盖Watcher接口、Event枚举(KeeperState、EventType)、WatchedEvent事件通知模型,以及客户端管理器ZKWatchManager的实现原理,揭示数据监听的注册、触发与清理全过程。

一、前言
  前面已经分析了Zookeeper持久话相关的类,下面接着分析Zookeeper中的Watcher机制所涉及到的类。
二、总体框图
  对于Watcher机制而言,主要涉及的类主要如下。   

说明:
Watcher 接口类型,其定义了process方法,需子类实现
Event 接口类型,Watcher的内部类,无任何方法
KeeperState 枚举类型,Event的内部类,表示Zookeeper所处的状态
EventType 枚举类型,Event的内部类,表示Zookeeper中发生的事件类型
WatchedEvent 表示对ZooKeeper上发生变化后的反馈,包含了KeeperState和EventType
ClientWatchManager 接口类型,表示客户端的Watcher管理者,其定义了materialized方法,需子类实现
ZKWatchManager Zookeeper的内部类,继承ClientWatchManager
MyWatcher ZooKeeperMain的内部类,继承Watcher
ServerCnxn 接口类型,继承Watcher,表示客户端与服务端的一个连接
WatchManager 管理Watcher
三、Watcher源码分析
3.1 内部类
  Event,接口类型,表示事件代表的状态,除去其内部类,其源码结构如下
public interface Watcher {

public interface Event {
    /**
     * Enumeration of states the ZooKeeper may be at the event
     */
    public enum KeeperState {

        @Deprecated
        Unknown (-1),

        Disconnected (0),

        @Deprecated
        NoSyncConnected (1),

        SyncConnected (3),

        AuthFailed (4),

        ConnectedReadOnly (5),

        SaslAuthenticated(6),

        Expired (-112);

        private final int intValue;     

        KeeperState(int intValue) {
            this.intValue = intValue;
        }

        public int getIntValue() {
            return intValue;
        }

        public static KeeperState fromInt(int intValue) {
            switch(intValue) {
                case   -1: return KeeperState.Unknown;
                case    0: return KeeperState.Disconnected;
                case    1: return KeeperState.NoSyncConnected;
                case    3: return KeeperState.SyncConnected;
                case    4: return KeeperState.AuthFailed;
                case    5: return KeeperState.ConnectedReadOnly;
                case    6: return KeeperState.SaslAuthenticated;
                case -112: return KeeperState.Expired;

                default:
                    throw new RuntimeException("Invalid integer value for conversion to KeeperState");
            }
        }
    }

    /**
     * Enumeration of types of events that may occur on the ZooKeeper
     */
    public enum EventType {
        None (-1),
        NodeCreated (1),
        NodeDeleted (2),
        NodeDataChanged (3),
        NodeChildrenChanged (4);

        private final int intValue;     

        EventType(int intValue) {
            this.intValue = intValue;
        }

        public int getIntValue() {
            return intValue;
        }

        public static EventType fromInt(int intValue) {
            switch(intValue) {
                case -1: return EventType.None;
                case  1: return EventType.NodeCreated;
                case  2: return EventType.NodeDeleted;
                case  3: return EventType.NodeDataChanged;
                case  4: return EventType.NodeChildrenChanged;

                default:
                    throw new RuntimeException("Invalid integer value for conversion to EventType");
            }
        }           
    }
}

}
说明:可以看到,Event接口并没有定义任何属性和方法,但其包含了KeeperState和EventType两个内部枚举类。
可以简化成:
public interface Event {}
3.2 接口方法  
abstract public void process(WatchedEvent event);
说明:其代表了实现Watcher接口时必须实现的的方法,即定义进行处理,WatchedEvent表示观察的事件。
四、Event源码分析(即3.1内部类)
4.1 内部类

  1. KeeperState 

public enum KeeperState { // 事件发生时Zookeeper的状态
/* Unused, this state is never generated by the server /
@Deprecated
// 未知状态,不再使用,服务器不会产生此状态
Unknown (-1),

/** The client is in the disconnected state - it is not connected
* to any server in the ensemble. */
// 断开
Disconnected (0),

/** Unused, this state is never generated by the server */
@Deprecated
// 未同步连接,不再使用,服务器不会产生此状态
NoSyncConnected (1),

/** The client is in the connected state - it is connected
* to a server in the ensemble (one of the servers specified
* in the host connection parameter during ZooKeeper client
* creation). */
// 同步连接状态
SyncConnected (3),

/**
* Auth failed state
*/
// 认证失败状态
AuthFailed (4),

/**
* The client is connected to a read-only server, that is the
* server which is not currently connected to the majority.
* The only operations allowed after receiving this state is
* read operations.
* This state is generated for read-only clients only since
* read/write clients aren't allowed to connect to r/o servers.
*/
// 只读连接状态
ConnectedReadOnly (5),

/**
* SaslAuthenticated: used to notify clients that they are SASL-authenticated,
* so that they can perform Zookeeper actions with their SASL-authorized permissions.
*/
// SASL认证通过状态
SaslAuthenticated(6),

/** The serving cluster has expired this session. The ZooKeeper
* client connection (the session) is no longer valid. You must
* create a new client connection (instantiate a new ZooKeeper
* instance) if you with to access the ensemble. */
// 过期状态
Expired (-112);

// 代表状态的整形值
private final int intValue;     // Integer representation of value
// for sending over wire


// 构造函数
KeeperState(int intValue) {
    this.intValue = intValue;
}

// 返回整形值
public int getIntValue() {
    return intValue;
}

// 从整形值构造相应的状态
public static KeeperState fromInt(int intValue) {
    switch(intValue) {
        case   -1: return KeeperState.Unknown;
        case    0: return KeeperState.Disconnected;
        case    1: return KeeperState.NoSyncConnected;
        case    3: return KeeperState.SyncConnected;
        case    4: return KeeperState.AuthFailed;
        case    5: return KeeperState.ConnectedReadOnly;
        case    6: return KeeperState.SaslAuthenticated;
        case -112: return KeeperState.Expired;

        default:
            throw new RuntimeException("Invalid integer value for conversion to KeeperState");
    }
}

}
说明:KeeperState是一个枚举类,其定义了在事件发生时Zookeeper所处的各种状态,其还定义了一个从整形值返回对应状态的方法fromInt。

  1. EventType 
    public enum EventType { // 事件类型
    // 无
    None (-1),
    // 结点创建
    NodeCreated (1),
    // 结点删除
    NodeDeleted (2),
    // 结点数据变化
    NodeDataChanged (3),
    // 结点子节点变化
    NodeChildrenChanged (4);

    // 代表事件类型的整形
    private final int intValue; // Integer representation of value
    // for sending over wire

    // 构造函数
    EventType(int intValue) {

     this.intValue = intValue;
    

    }

    // 返回整形
    public int getIntValue() {

     return intValue;
    

    }

    // 从整形构造相应的事件
    public static EventType fromInt(int intValue) {

     switch(intValue) {
         case -1: return EventType.None;
         case  1: return EventType.NodeCreated;
         case  2: return EventType.NodeDeleted;
         case  3: return EventType.NodeDataChanged;
         case  4: return EventType.NodeChildrenChanged;
    
         default:
             throw new RuntimeException("Invalid integer value for conversion to EventType");
     }
    

    }
    }
    说明:EventType是一个枚举类,其定义了事件的类型(如创建节点、删除节点等事件),同时,其还定义了一个从整形值返回对应事件类型的方法fromInt。
    五、WatchedEvent
    5.1 类的属性 
    public class WatchedEvent {
    // Zookeeper的状态
    final private KeeperState keeperState;
    // 事件类型
    final private EventType eventType;
    // 事件所涉及节点的路径
    private String path;
    }
    说明:WatchedEvent类包含了三个属性,分别代表事件发生时Zookeeper的状态、事件类型和发生事件所涉及的节点路径。
    5.2 构造函数
      1. public WatchedEvent(EventType eventType, KeeperState keeperState, String path)型构造函数 
    public WatchedEvent(EventType eventType, KeeperState keeperState, String path) {
    // 初始化属性
    this.keeperState = keeperState;
    this.eventType = eventType;
    this.path = path;
    }
      说明:构造函数传入了三个参数,然后分别对属性进行赋值操作。
      2. public WatchedEvent(WatcherEvent eventMessage)型构造函数  
    public WatchedEvent(WatcherEvent eventMessage) {
    // 从eventMessage中取出相应属性进行赋值
    keeperState = KeeperState.fromInt(eventMessage.getState());
    eventType = EventType.fromInt(eventMessage.getType());
    path = eventMessage.getPath();
    }
      说明:构造函数传入了WatcherEvent参数,之后直接从该参数中取出相应属性进行赋值操作。
    五总结:对于WatchedEvent类的方法而言,相对简单,包含了几个getXXX方法,用于获取相应的属性值。
    六、ClientWatchManager
    public Set materialize(Watcher.Event.KeeperState state,

                             Watcher.Event.EventType type, String path);
    

      说明:该方法表示事件发生时,返回需要被通知的Watcher集合,可能为空集合。
    七、ZKWatchManager(zookeeper内)
    7.1 类的属性
    private static class ZKWatchManager implements ClientWatchManager {

    // 数据变化的Watchers
    private final Map> dataWatches = new HashMap>();

    // 节点存在与否的Watchers
    private final Map> existWatches = new HashMap>();

    // 子节点变化的Watchers
    private final Map> childWatches = new HashMap>();
    }
     说明:ZKWatchManager实现了ClientWatchManager,并定义了三个Map键值对,键为节点路径,值为Watcher。分别对应数据变化的Watcher、节点是否存在的Watcher、子节点变化的Watcher。
    7.2 核心方法分析

  2. materialize方法
    public Set materialize(Watcher.Event.KeeperState state,

                             Watcher.Event.EventType type,
                             String clientPath)
    

    {
    // 新生成结果Watcher集合
    Set result = new HashSet();

    switch (type) { // 确定事件类型

     case None: // 无类型
         // 添加默认Watcher
         result.add(defaultWatcher);
         // 是否需要清空(提取对zookeeper.disableAutoWatchReset字段进行配置的值、
         // Zookeeper的状态是否为同步连接)
         boolean clear = ClientCnxn.getDisableAutoResetWatch() &&
             state != Watcher.Event.KeeperState.SyncConnected;
         // 同步块
         synchronized(dataWatches) { 
             for(Set<Watcher> ws: dataWatches.values()) {
                 // 添加至结果集合
                 result.addAll(ws);
             }
             if (clear) { // 是否需要清空
                 dataWatches.clear();
             }
         }
    
         // 同步块
         synchronized(existWatches) {  
             for(Set<Watcher> ws: existWatches.values()) {
                 // 添加至结果集合
                 result.addAll(ws);
             }
             if (clear) { // 是否需要清空
                 existWatches.clear();
             }
         }
    
         // 同步块
         synchronized(childWatches) { 
             for(Set<Watcher> ws: childWatches.values()) {
                 // 添加至结果集合
                 result.addAll(ws);
             }
             if (clear) { // 是否需要清空
                 childWatches.clear();
             }
         }
         // 返回结果
         return result;
     case NodeDataChanged: // 节点数据变化
     case NodeCreated: // 创建节点
         synchronized (dataWatches) { // 同步块
             // 移除clientPath对应的Watcher后全部添加至结果集合
             addTo(dataWatches.remove(clientPath), result);
         }
         synchronized (existWatches) { 
             // 移除clientPath对应的Watcher后全部添加至结果集合
             addTo(existWatches.remove(clientPath), result);
         }
         break;
     case NodeChildrenChanged: // 节点子节点变化
         synchronized (childWatches) {
             // 移除clientPath对应的Watcher后全部添加至结果集合
             addTo(childWatches.remove(clientPath), result);
         }
         break;
     case NodeDeleted: // 删除节点
         synchronized (dataWatches) { 
             // 移除clientPath对应的Watcher后全部添加至结果集合
             addTo(dataWatches.remove(clientPath), result);
         }
         // XXX This shouldn't be needed, but just in case
         synchronized (existWatches) {
             // 移除clientPath对应的Watcher
             Set<Watcher> list = existWatches.remove(clientPath);
             if (list != null) {
                 // 移除clientPath对应的Watcher后全部添加至结果集合
                 addTo(existWatches.remove(clientPath), result);
                 LOG.warn("We are triggering an exists watch for delete! Shouldn't happen!");
             }
         }
         synchronized (childWatches) {
             // 移除clientPath对应的Watcher后全部添加至结果集合
             addTo(childWatches.remove(clientPath), result);
         }
         break;
     default: // 缺省处理
         String msg = "Unhandled watch event type " + type
             + " with state " + state + " on path " + clientPath;
         LOG.error(msg);
         throw new RuntimeException(msg);
    

    }

    // 返回结果集合
    return result;
    }
    说明:该方法在事件发生后,返回需要被通知的Watcher集合。在该方法中,首先会根据EventType类型确定相应的事件类型,然后根据事件类型的不同做出相应的操作:
    如针对None类型,即无任何事件,则首先会从三个键值对中删除clientPath对应的Watcher,然后将剩余的Watcher集合添加至结果集合;
    针对NodeDataChanged和NodeCreated事件而言,其会从dataWatches和existWatches中删除clientPath对应的Watcher,然后将剩余的Watcher集合添加至结果集合。
    八、总结
      针对Watcher机制的第一部分的源码分析就已经完成,本章节需重点关注:
    ● 事件的变化,状态的定义依赖于Event内部类的两组枚举值
    ● 上下游调用关系图需记忆一下,为加强记忆,再最后再贴一下

相关文章
|
7月前
|
Java 开发工具 数据安全/隐私保护
项目《中州养老》
《中州养老》是一个面向养老院的后台管理系统,涵盖员工管理端与家属小程序端。系统功能包括预约参观、入住退住、计费管理、健康监测等模块。项目采用Vue3+TS+TDesign构建前端,后端基于SpringBoot(SSM),集成Redis缓存、Nginx部署、阿里云OSS与IoT平台。支持RBAC权限控制、智能设备数据实时监控、微信登录、定时任务处理及多线程高效通信,实现养老业务全流程数字化管理。(238字)
|
7月前
|
Arthas Java 测试技术
1.下载安装
Arthas是一款Java诊断工具,需先启动Java应用并安装JDK。通过curl下载arthas-boot.jar,运行后选择目标进程即可接入。支持多Java进程管理,适用于SpringBoot等应用的线上问题排查与动态诊断。
|
7月前
|
缓存 开发工具 git
QLExpress使用及源码分析
QLExpress是阿里开源的轻量级规则引擎,支持通过注解与YAML配置实现业务逻辑解耦。通过@QLAlias、@QLRule等注解定义规则接口,结合AST语法树解析与上下文执行,实现动态脚本编译与缓存,适用于复杂条件判断与动态计算场景,具备高扩展性与易维护性。
|
7月前
|
存储 算法 BI
xxljob本地运行
本文介绍XXL-JOB分布式任务调度的部署与使用。包含源码获取、数据库导入、服务端配置启动、客户端注册及执行器配置。详细说明各数据表作用、路由策略类型,并演示任务创建、参数设置与执行日志查看,助你快速上手XXL-JOB。
|
7月前
|
数据可视化 Java 关系型数据库
01-认识Activiti
工作流指如请假、报销等需审批的业务流程,通过可视化引擎实现多节点审批,广泛应用于CRM、WMS等系统。主流技术包括BPMN、Activiti和Flowable,其中Activiti为Java系开源引擎,支持复杂流程自动化,推动企业流程数字化。
|
7月前
|
存储 JSON NoSQL
MongoDB常用命令
本教程介绍MongoDB数据库操作,涵盖数据库与集合的创建、删除,文档的增删改查及分页排序查询。以文章评论系统为例,演示数据存储结构设计、CRUD操作、条件查询、投影与排序,以及分页实现方法,帮助快速掌握MongoDB基本用法。(238字)
|
7月前
|
存储 消息中间件 开发框架
应用架构图
在业务架构基础上,技术架构将产品需求转化为技术实现。它涵盖分层设计、技术选型与关键组件关系,包括单体四层结构(表现、业务、数据、基础层)和分布式应用间的调用关系,明确内外系统边界,形成完整技术体系图谱。(238字)
|
7月前
|
存储 负载均衡 Java
第九章 SpringCloud框架
Nacos支持服务注册发现与配置管理,基于心跳机制实现健康监测,提供分级存储模型(命名空间、组、DataId)实现多环境隔离。相比Eureka,Nacos具备更优的实时推送、更短心跳周期,并融合配置中心功能。OpenFeign结合LoadBalancer实现服务调用负载均衡,支持轮询、随机等策略。Sentinel提供流量控制、熔断降级,通过滑动窗口算法精准限流。Spring Cloud Gateway作为网关核心,实现路由匹配、断言过滤与请求处理,支持限流、鉴权、监控等功能,提升微服务架构稳定性与可维护性。(239字)
|
7月前
|
存储 NoSQL 关系型数据库
MongoDB索引知识
MongoDB索引通过B树结构提升查询效率,避免全表扫描。支持单字段、复合、地理空间、文本及哈希索引,适用于等值、范围、排序及全文搜索,显著提升大数据量下的查询性能。
|
7月前
|
SQL 缓存 Java
第八章 Mybatis及MybatisPlus
MyBatis通过配置文件与映射文件构建SqlSessionFactory,创建SqlSession执行SQL。基于Executor与Mapped Statement实现参数映射、结果集处理,支持resultMap复杂映射、@Param传参及分页插件PageHelper、MyBatis Plus,提供一二级缓存机制,简化数据库操作。