一、前言

　　前面已经分析了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内部类的两组枚举值
上下游调用关系图需记忆一下，为加强记忆，再最后再贴一下

Watcher机制（一）

一、前言

二、总体框图

三、Watcher源码分析

3.1 内部类

3.2 接口方法

四、Event源码分析(即3.1内部类)

4.1 内部类

1. KeeperState

2. EventType

五、WatchedEvent

5.1 类的属性

5.2 构造函数

六、ClientWatchManager

七、ZKWatchManager(zookeeper内)

7.1 类的属性

7.2 核心方法分析

1. materialize方法

八、总结

热门文章

最新文章

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

Watcher机制（一）

一、前言

二、总体框图

三、Watcher源码分析

3.1 内部类

3.2 接口方法

四、Event源码分析(即3.1内部类)

4.1 内部类

1. KeeperState

2. EventType

五、WatchedEvent

5.1 类的属性

5.2 构造函数

六、ClientWatchManager

七、ZKWatchManager(zookeeper内)

7.1 类的属性

7.2 核心方法分析

1. materialize方法

八、总结

热门文章

最新文章

相关电子书

3.2 接口方法　　

1. KeeperState　

2. EventType　

5.1 类的属性　