4.Watcher机制(一)

简介: 本文深入分析Zookeeper的Watcher机制,涵盖核心类与源码实现。重点解析Watcher、Event、WatchedEvent等接口与类,阐述其在状态监听与事件通知中的作用,并结合ZKWatchManager管理机制,揭示数据变更时的Watcher触发流程。

一、前言

  前面已经分析了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。

2. 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<Watcher> 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<String, Set<Watcher>> dataWatches = new HashMap<String, Set<Watcher>>();
    
    // 节点存在与否的Watchers
    private final Map<String, Set<Watcher>> existWatches = new HashMap<String, Set<Watcher>>();
    
    // 子节点变化的Watchers
    private final Map<String, Set<Watcher>> childWatches = new HashMap<String, Set<Watcher>>();
}

 说明:ZKWatchManager实现了ClientWatchManager,并定义了三个Map键值对,键为节点路径,值为Watcher。分别对应数据变化的Watcher、节点是否存在的Watcher、子节点变化的Watcher。

7.2 核心方法分析

1. materialize方法

public Set<Watcher> materialize(Watcher.Event.KeeperState state,
                                Watcher.Event.EventType type,
                                String clientPath)
{
    // 新生成结果Watcher集合
    Set<Watcher> result = new HashSet<Watcher>();
    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内部类的两组枚举值
  • 上下游调用关系图需记忆一下,为加强记忆,再最后再贴一下


相关文章
|
1天前
|
缓存 运维 监控
一场FullGC故障排查
本文记录了一次Java应用CPU使用率异常升高的排查过程。通过分析发现,问题根源为频繁Full GC导致CPU飙升,而Full GC是因用户上传的Excel数据被加载为大对象并长期驻留JVM内存所致。使用JProfiler分析堆内存,定位到List&lt;Map&lt;String, String&gt;&gt;结构造成内存膨胀,空间效率仅约13.4%。最终提出“治本”与“治标”两类解决方案:一是将大数据移出JVM内存,存入Redis;二是优化代码,及时清理无用字段以减小对象体积。文章总结了从监控识别、工具分析到根本解决的完整排查思路,对类似性能问题具有参考价值。(238字)
|
1天前
|
关系型数据库 应用服务中间件 nginx
容器化部署引擎Docker
Docker是一种容器化技术,通过镜像打包应用及依赖,实现跨环境一致部署。它利用容器隔离运行,解决开发、测试、生产环境差异与组件兼容性问题,具备启动快、资源占用少、易于迁移等优势,是现代微服务部署的核心工具。
 容器化部署引擎Docker
|
1天前
|
SQL 容灾 Nacos
Seata的部署和集成
本文介绍Seata TC服务器的部署与微服务集成,包括下载、配置、数据库表初始化及高可用集群搭建,实现基于Nacos的分布式事务管理与异地容灾支持。
|
1天前
|
自然语言处理 数据可视化 Docker
安装ES、Kibana、IK
本文介绍如何通过Docker部署单节点Elasticsearch与Kibana,并安装IK分词器。内容涵盖创建网络、加载镜像、运行容器、配置扩展词典与停用词典,以及常见启动报错处理,帮助快速搭建ES开发环境。
安装ES、Kibana、IK
|
1天前
|
JSON 自然语言处理 算法
DSL语法、搜索结果处理
本文介绍了Elasticsearch的DSL查询语法及RestClient实现方式,涵盖全文检索、精确查询、地理坐标查询和复合查询,并结合黑马旅游案例实现了搜索、分页、过滤与高亮功能。
 DSL语法、搜索结果处理
|
1天前
|
自然语言处理 关系型数据库 MySQL
数据聚合、自动补全、数据同步
本文介绍了Elasticsearch中的核心功能:数据聚合、自动补全与数据同步。聚合支持对数据分组(Bucket)、统计计算(Metric)及管道聚合,可高效实现品牌、价格等分析;通过拼音分词器与Completion Suggester实现搜索自动补全;并利用MQ异步通知机制实现MySQL与ES之间的数据同步,确保数据一致性,提升搜索实时性与准确性。(238字)
数据聚合、自动补全、数据同步
|
1天前
|
存储 监控 Docker
ElasticSearch集群
Elasticsearch集群通过分片和副本解决海量数据存储与单点故障问题。分片实现数据水平拆分,副本保障高可用,结合节点角色划分与故障转移机制,提升系统稳定性与性能。
 ElasticSearch集群
|
1天前
|
Kubernetes Java 应用服务中间件
1.开发篇(脚手架下载)
本文介绍基于SpringCloud + Kubernetes的微服务开发实践,重点分享EDAS 3.0在项目初始化与本地启动环节的优化体验。通过阿里云start.aliyun.com脚手架快速生成项目,结合Cloud Toolkit插件一键拉起本地注册中心,实现应用快速部署与联调,提升开发者效率。后续将深入讲解云端部署及端云互联能力。
|
1天前
|
Kubernetes IDE 应用服务中间件
2.部署篇(开发部署)
本文介绍如何将SpringCloud应用通过EDAS部署至Kubernetes集群。涵盖集群导入、应用初始化及IDE插件快速部署,助力开发者高效上云。
|
1天前
|
存储 缓存 负载均衡
Nacos注册中心
本文介绍Nacos的安装部署、服务注册中心整合、分级模型、负载均衡策略、权重控制、环境隔离及实例类型,详解其在微服务架构中的应用,帮助开发者掌握Nacos核心功能与最佳实践。
 Nacos注册中心