背景

项目中使用了zookeeper进行的类似工作流引擎的工作流转，将一次工作请求拆分了4个节点(S/E/T/L)。S阶段做完后，通过zk的watcher触发下一个E节点进行处理，S和E可能为不同的jvm上，所以需要走一个分布式的消息进行通知。

思路

基于zookeeper做持久化watcher，项目中直接使用zookeeper官方api，大致的工作模型：

1.private synchronized void initNodes(List<String> nodes) {
2.    // 根据zk节点，判断是否需要处理
3.}
4.
5.private void syncNodes() {
6.        try {
7.            List<String> nodes = zookeeper.getChildren(ArbitrateConstants.NODE_NID_ROOT, new AsyncWatcher() {
8.
9.                public void asyncProcess(WatchedEvent event) {
10.                    syncNodes();// 继续关注node节点变化
11.                }
12.            });
13.
14.            initNodes(nodes);
15.        } catch (KeeperException e) {
16.            syncNodes();
17.        } catch (InterruptedException e) {
18.            // ignore
19.        }
20.    }

• 有两个方法initNodes 和 syncNodes，  syncNodes主要是监听zookeeper的节点变化
• syncNodes会通过级联方式，在每次watcher被触发后，就会再挂一次watcher。完成了一个类似链式触发的功能

遇到的问题

系统上线运行后，跑了几天时间，跑出了一个OutOfMemory的问题，jmap dump了下对应的内存数据文件，发现了一个zk使用上的问题.

a. 通过mat分析查看了下jvm中占用内存最大的对象，居然是zookeeper中的一个waitingEvents.：

b. waitingEvents中的WatcherSetEventPair对象中，包含了一个待响应的watchers和对应的响应event事件对象，对应的watchers数量居然有300W个

问题分析：

分析了下WatcherSetEventPair中的处理机制。

Event响应中对应EventType的枚举类型：(存在一个特殊的None类型)

• None (-1),
• NodeCreated (1),
• NodeDeleted (2),
• NodeDataChanged (3),
• NodeChildrenChanged (4);

1.eventThread.queueEvent(new WatchedEvent(
2.                        Watcher.Event.EventType.None,
3.                        Watcher.Event.KeeperState.Expired, null));

针对None类型，在获取对应的watcher响应时：

1.public Set<Watcher> materialize(Watcher.Event.KeeperState state,
2.                                        Watcher.Event.EventType type,
3.                                        String clientPath)
4.        {
5.            Set<Watcher> result = new HashSet<Watcher>();
6.
7.            switch (type) {
8.            case None:
9.                result.add(defaultWatcher);
10.                for(Set<Watcher> ws: dataWatches.values()) {
11.                    result.addAll(ws);
12.                }
13.                for(Set<Watcher> ws: existWatches.values()) {
14.                    result.addAll(ws);
15.                }
16.                for(Set<Watcher> ws: childWatches.values()) {
17.                    result.addAll(ws);
18.                }
19.
20.                // clear the watches if auto watch reset is not enabled
21.                if (ClientCnxn.getDisableAutoResetWatch() &&
22.                        state != Watcher.Event.KeeperState.SyncConnected)
23.                {
24.                    synchronized(dataWatches) {
25.                        dataWatches.clear();
26.                    }
27.                    synchronized(existWatches) {
28.                        existWatches.clear();
29.                    }
30.                    synchronized(childWatches) {
31.                        childWatches.clear();
32.                    }
33.                }
34.
35.                return result;

总结

a.  需要明确watcher的触发条件和触发case场景。特别注意，None类型可能会引起触发2次watcher调用

(截取了淘宝同学的blog : http://rdc.taobao.com/team/jm/archives/1047)

event For “/path”defaultWatcherexists
(“/path”)getData
(“/path”)getChildren
(“/path”)

b.  出现session expired，需要重建zookeeper connector，对应的watcher会失效。因为watcher在client的存储是和对应的zookeeper client绑定，不同的client有不同的watcher列表。