zookeeper（10）源码分析-事件监听Watcher

Posted 2020-11-24

今天继续源码分析，分析一下org.apache.zookeeper.server下的WatchManager类。

WatcherManager类用于管理watchers和相应的触发器。

类的属性

//watchTable表示从节点路径到watcher集合的映射
    private final HashMap<String, HashSet<Watcher>> watchTable =
        new HashMap<String, HashSet<Watcher>>();
    //watch2Paths则表示从watcher到所有节点路径集合的映射
    private final HashMap<Watcher, HashSet<String>> watch2Paths =
        new HashMap<Watcher, HashSet<String>>();

核心方法

1. size方法

size方法是同步的，因此在多线程环境下是安全的，其主要作用是获取watchTable的大小，即遍历watchTable的值集合，每个集合大小累加。

synchronized int size(){
        int result = 0;
        for(Set<Watcher> watches : watchTable.values()) {
            result += watches.size();
        }
        return result;
    }

2、addWatch方法

addWatch方法同样是同步的，主要用来更新类里面的上面提到的两个集合属性。

synchronized void addWatch(String path, Watcher watcher) {
        //通过传入的path（节点路径）从watchTable获取相应的watcher集合
        HashSet<Watcher> list = watchTable.get(path);
        if (list == null) { //watcher是否为空，若为空
            // don‘t waste memory if there are few watches on a node
            // rehash when the 4th entry is added, doubling size thereafter
            // seems like a good compromise
            //新生成watcher集合，并将路径path和此集合添加至watchTable中
            list = new HashSet<Watcher>(4);
            watchTable.put(path, list);
        }
        //将传入的watcher添加至watcher集合，即完成了path和watcher添加至watchTable的步骤
        list.add(watcher);
        //通过传入的watcher从watch2Paths中获取相应的path集合
        HashSet<String> paths = watch2Paths.get(watcher);
        if (paths == null) {// 判断path集合是否为空，若为空
            // cnxns typically have many watches, so use default cap here
            //新生成path集合，并将watcher和paths添加至watch2Paths中
            paths = new HashSet<String>();
            watch2Paths.put(watcher, paths);
        }
        // 将传入的path（节点路径）添加至path集合，即完成了path和watcher添加至watch2Paths的步骤
        paths.add(path);
    }

3、removeWatcher

removeWatcher用作从watch2Paths和watchTable中中移除该watcher

synchronized void removeWatcher(Watcher watcher) {
        //从wach2Paths中移除watcher，并返回watcher对应的path集合
        HashSet<String> paths = watch2Paths.remove(watcher);
        if (paths == null) {
            return;
        }
        for (String p : paths) {
            //从watcherTable中根据路径取出相应的watcher集合
            HashSet<Watcher> list = watchTable.get(p);
            if (list != null) {
                // 从list中移除该watcher
                list.remove(watcher);
                // 移除后list为空，则从watchTable中移出path
                if (list.size() == 0) {
                    watchTable.remove(p);
                }
            }
        }
    }

4、triggerWatch

该方法主要用于触发watch事件，并对事件进行处理。

Set<Watcher> triggerWatch(String path, EventType type) {
        return triggerWatch(path, type, null);
    }

    Set<Watcher> triggerWatch(String path, EventType type, Set<Watcher> supress) {
        // 根据事件类型、连接状态、节点路径创建WatchedEvent
        WatchedEvent e = new WatchedEvent(type,
                KeeperState.SyncConnected, path);
        HashSet<Watcher> watchers;
        synchronized (this) {
            // 从watcher表中移除path，并返回其对应的watcher集合
            watchers = watchTable.remove(path);
            if (watchers == null || watchers.isEmpty()) {
                if (LOG.isTraceEnabled()) {
                    ZooTrace.logTraceMessage(LOG,
                            ZooTrace.EVENT_DELIVERY_TRACE_MASK,
                            "No watchers for " + path);
                }
                return null;
            }
            // 遍历watcher集合
            for (Watcher w : watchers) {
                // 根据watcher从watcher表中取出路径集合
                HashSet<String> paths = watch2Paths.get(w);
                if (paths != null) {
                                // 如果paths不为空，则移除传入路径path
                    paths.remove(path);
                }
            }
        }
        // 遍历watcher集合
        for (Watcher w : watchers) {
            if (supress != null && supress.contains(w)) {
                continue;
            }
            // watcher进行处理
            w.process(e);
        }
        return watchers;
    }