hadoop系列：zookeeper（3）——zookeeper核心原理（事件）
package com.yinwenjie.test.zookeepertest.test;import java.io.FileNotFoundException;import java.io.IOException;import org.apache.commons.logging.Log;import org.apache.commons.logging.LogFactory;import org.apache.zookeeper.CreateMode;import org.apache.zookeeper.WatchedEvent;import org.apache.zookeeper.Watcher;import org.apache.zookeeper.Watcher.Event.EventType;import org.apache.zookeeper.Watcher.Event.KeeperState;import org.apache.zookeeper.ZooDefs.Ids;import org.apache.zookeeper.ZooKeeper;import org.springframework.util.Log4jConfigurer;import com.yinwenjie.test.zookeepertest.TestZookeeperAgainst;/*** 这个测试类测试多个watcher监控某一个znode节点的效果。
* servers：192.168.61.129:2181，192.168.61.130:2181，192.168.61.132:2181
* 为了验证zk集群的工作效果，我们选择一个节点进行连接(192.168.61.129)。
* @author yinwenjie
*/public class TestManyWatcher implements Runnable {
static { try {
Log4jConfigurer.initLogging("classpath:log4j.properties");
} catch (FileNotFoundException ex) {
System.err.println("Cannot Initialize log4j");
System.exit(-1);
}
} /**
* 日志
*/
private static final Log LOGGER = LogFactory.getLog(TestZookeeperAgainst.class); public static void main(String[] args) throws Exception {
TestManyWatcher testManyWatcher = new TestManyWatcher(); new Thread(testManyWatcher).start();
} public void run() { /*
* 验证过程如下：
* 1、验证一个节点X上使用exist方式注册的多个监听器（ManyWatcherOne、ManyWatcherTwo），
* 在节点X发生create事件时的事件通知情况
* 2、验证一个节点Y上使用getDate方式注册的多个监听器（ManyWatcherOne、ManyWatcherTwo），
* 在节点X发生create事件时的事件通知情况
* */
//默认监听：注册默认监听是为了让None事件都由默认监听处理，
//不干扰ManyWatcherOne、ManyWatcherTwo的日志输出
ManyWatcherDefault watcherDefault = new ManyWatcherDefault();
ZooKeeper zkClient = null; try {
zkClient = new ZooKeeper("192.168.61.129:2181", 120000, watcherDefault);
} catch (IOException e) {
TestManyWatcher.LOGGER.error(e.getMessage(), e); return;
} //默认监听也可以使用register方法注册
//zkClient.register(watcherDefault);
//1、========================================================
TestManyWatcher.LOGGER.info("=================以下是第一个实验");
String path = "/X";
ManyWatcherOne watcherOneX = new ManyWatcherOne(zkClient , path);
ManyWatcherTwo watcherTwoX = new ManyWatcherTwo(zkClient , path); //注册监听，注意，这里两次exists方法的执行返回都是null，因为“X”节点还不存在
try {
zkClient.exists(path, watcherOneX);
zkClient.exists(path, watcherTwoX); //创建"X"节点，为了简单起见，我们忽略权限问题。
//并且创建一个临时节点，这样重复跑代码的时候，不用去server上手动删除)
zkClient.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
} catch (Exception e) {
TestManyWatcher.LOGGER.error(e.getMessage(), e); return;
} //TODO 注意观察日志，根据原理我们猜测理想情况下ManyWatcherTwo和ManyWatcherOne都会被通知。
//2、========================================================
TestManyWatcher.LOGGER.info("=================以下是第二个实验");
path = "/Y";
ManyWatcherOne watcherOneY = new ManyWatcherOne(zkClient , path);
ManyWatcherTwo watcherTwoY = new ManyWatcherTwo(zkClient , path); //注册监听，注意，这里使用两次getData方法注册监听，"Y"节点目前并不存在
try {
zkClient.getData(path, watcherOneY, null);
zkClient.getData(path, watcherTwoY, null);
} catch (Exception e) {
TestManyWatcher.LOGGER.error(e.getMessage(), e);
} //TODO 注意观察日志，因为"Y"节点不存在，所以getData就会出现异常。watcherOneY、watcherTwoY的注册都不起任何作用。
//然后我们在报了异常的情况下，创建"Y"节点，根据原理，不会有任何watcher响应"Y"节点的create事件
try {
zkClient.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
} catch (Exception e) {
TestManyWatcher.LOGGER.error(e.getMessage(), e); return;
} //下面这段代码可以忽略，是为了观察zk的原理。让守护线程保持不退出
synchronized(this) { try { this.wait();
} catch (InterruptedException e) {
TestManyWatcher.LOGGER.error(e.getMessage(), e);
System.exit(-1);
}
}
}
}/**
* 这是默认的watcher实现。
* @author yinwenjie
*/class ManyWatcherDefault implements Watcher { /**
* 日志
*/
private static Log LOGGER = LogFactory.getLog(ManyWatcherDefault.class); public void process(WatchedEvent event) {
KeeperState keeperState = event.getState();
EventType eventType = event.getType();
ManyWatcherDefault.LOGGER.info("=========默认监听到None事件：keeperState = "
+ keeperState + ":eventType = " + eventType);
}
}/**
* 这是第一种watcher
* @author yinwenjie
*/class ManyWatcherOne implements Watcher { /**
* 日志
*/
private static Log LOGGER = LogFactory.getLog(ManyWatcherOne.class); private ZooKeeper zkClient; /**
* 被监控的znode地址
*/
private String watcherPath; public ManyWatcherOne(ZooKeeper zkClient , String watcherPath) { this.zkClient = zkClient; this.watcherPath = watcherPath;
} public void process(WatchedEvent event) { try { this.zkClient.exists(this.watcherPath, this);
} catch (Exception e) {
ManyWatcherOne.LOGGER.error(e.getMessage(), e);
}
KeeperState keeperState = event.getState();
EventType eventType = event.getType(); //这个属性是发生事件的path
String eventPath = event.getPath();
ManyWatcherOne.LOGGER.info("=========ManyWatcherOne监听到" + eventPath + "地址发生事件："
+ "keeperState = " + keeperState + ":eventType = " + eventType);
}
}/**
* 这是第二种watcher
* @author yinwenjie
*/class ManyWatcherTwo implements Watcher { /**
* 日志
*/
private static Log LOGGER = LogFactory.getLog(ManyWatcherOne.class); private ZooKeeper zkClient; /**
* 被监控的znode地址
*/
private String watcherPath; public ManyWatcherTwo(ZooKeeper zkClient, String watcherPath) { this.zkClient = zkClient; this.watcherPath = watcherPath;
} public void process(WatchedEvent event) { try { this.zkClient.exists(this.watcherPath, this);
} catch (Exception e) {
ManyWatcherTwo.LOGGER.error(e.getMessage(), e);
}
KeeperState keeperState = event.getState();
EventType eventType = event.getType(); //这个属性是发生事件的path
String eventPath = event.getPath();
ManyWatcherTwo.LOGGER.info("=========ManyWatcherTwo监听到" + eventPath + "地址发生事件："
+ "keeperState = " + keeperState + ":eventType = " + eventType);
}
}123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209
页:
[1]