创建maven工程
略
pom.xml
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.8</version>
</dependency>
log4j.properties
打印 hadoop debug 信息
在 resources 文件夹下创建 log4j.properties 文件，内容如下：
log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
java
ZkUtil
zookeeper工具类，方便连接 zookeeper、注册节点
package std;
import java.io.IOException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
public class ZkUtil {
private static String connectString = "hadoop1:2181,hadoop2:2181,hadoop3:2181";
private static int sessionTimeout = 2000;
//根节点名
private static String ROOT_NODE="/myserver";
// 服务名
private static String SERVICE_NAME="notice";
private ZooKeeper zk = null;
// 创建到 zk 的客户端连接
public void connect() throws IOException {
zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
}
});
}
// 注册服务器
public void regist(String nodeData) throws Exception {
//判断根节点是否存在，false表示不需要监听
Stat stat = zk.exists(ROOT_NODE, false);
if(stat==null){//不存在就创建
zk.create(
ROOT_NODE, //节点路径
null, //节点数据
Ids.OPEN_ACL_UNSAFE, //节点权限
CreateMode.PERSISTENT //持久化目录节点
);
}
//创建子节点
String res = zk.create(
ROOT_NODE+"/"+SERVICE_NAME, //节点路径
nodeData.getBytes(), //节点数据
Ids.OPEN_ACL_UNSAFE, //节点权限
CreateMode.EPHEMERAL_SEQUENTIAL //临时顺序编号目录节点，由zk进行顺序编号，且断开后，该节点被删除
);
System.out.println("节点："+res);
}
}
notice1
模拟 notice1
package std;
public class NoticeService1 {
public static void main(String[] args) throws Exception {
String data="hadoop1:8080";
ZkUtil zk = new ZkUtil();
zk.connect();
zk.regist(data);
System.out.println(data+" 启动成功！");
//休眠，否则退出程序，由于创建的是临时节点，会自动删除节点
Thread.sleep(1000000);
}
}
notice2
package std;
public class NoticeService2 {
public static void main(String[] args) throws Exception {
String data="hadoop2:8080";
ZkUtil zk = new ZkUtil();
zk.connect();
zk.regist(data);
System.out.println(data+" 启动成功！");
Thread.sleep(1000000);//休眠，否则退出程序
}
}
notice3
package std;
public class NoticeService3 {
public static void main(String[] args) throws Exception {
String data="hadoop3:8080";
ZkUtil zk = new ZkUtil();
zk.connect();
zk.regist(data);
System.out.println(data+" 启动成功！");
Thread.sleep(1000000);//休眠，否则退出程序
}
}
GuanzhuService
启动后，向 zookeeper 发请求，获取服务器列表，并注册监听
package std;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.time.LocalTime;
import java.util.List;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
public class GuanzhuService {
private static String connectString = "hadoop1:2181,hadoop2:2181,hadoop3:2181";
private static int sessionTimeout = 2000;
private ZooKeeper zk = null;
// 根节点名
private static String ROOT_NODE = "/myserver";
// 创建到 zk 的客户端连接
public void connect() throws IOException {
zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
try {
LocalTime time= LocalTime.now();
System.out.println("监听开始：==============="+time);
getServers();
System.out.println("监听结束===============\n\n\n");
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
}
// 获取服务器信息
public void getServers() throws Exception {
// 获取子节点信息，true表示对节点进行监听
List<String> children = zk.getChildren(ROOT_NODE, true);
// 遍历所有节点，获取节点中的数据
for (String child : children) {
byte[] data = zk.getData(ROOT_NODE + "/" + child, false, null);
System.out.println("notice微服务："+new String(data));
}
System.out.println("=====");
}
public static void main(String[] args) throws Exception {
GuanzhuService service = new GuanzhuService();
service.connect();
// ？？？？必须调用下面方法，否则 connect() 方法中的 getServers() 不会执行
// service.getServers();
// 休眠，否则退出程序，退出后，无法监听
Thread.sleep(1000000);//休眠，否则退出程序
}
}
启动
先启动 notice1、notice2、notice3
最后启动 GuanzhuService，启动后，读取 notice 列表信息
测试 notice3 下线
停止 notice3，zk会删除相应的节点数据，并向 GuanzhuService 发送新的节点数据
测试 notice3 上线
再次启动 notice3，向 zk 写入节点数据，zk向 GuanzhuService 发送新的节点数据