hadoop|Zookeeper 应用案例（一）之服务器上下线动态感知服务器动态感知|zookeeper

Zookeeper 应用案例（一）之服务器上下线动态感知

1、需求描述
2、设计思路
3、代码开发
- 3.1、服务器端处理
- 3.2、客户端处理
- 3.3、运行效果
4、学习内容

1、需求描述某分布式系统中，主节点可以有多台，可以动态上下线。
任意一台客户端都能实时感知到主节点服务器的上下线。
2、设计思路（1）设计服务器端存入服务器上线，下线的信息，比如都写入到 servers 节点下；
（2）设计客户端监听该 servers 节点，获取该服务器集群的在线服务器列表；
（3）服务器一上线，就往 zookeeper 文件系统中的一个统一的节点比如 servers 下写入一个临时节点，记录下服务器的信息（思考，该节点最好采用什么类型的节点？）
（4）服务器一下线，则删除 servers 节点下的该服务器的信息，则客户端因为监听了该节点的数据变化，所以将第一时间得知服务器的在线状态。
3、代码开发加入依赖：

org.apache.zookeeper zookeeper 3.4.10

3.1、服务器端处理

package com.zc.bigdata.demo1; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.Stat; /** * @作者： zc * @时间： 2021/3/4 9:50 * @描述：用来模拟服务器的动态上线下线。 *总体思路就是服务器上线就上 zookeeper 集群创建一个临时节点， *然后监听了该数据节点的个数变化的客户端都收到通知下线， *则该临时节点自动删除，监听了该数据节点的个数变化的客户端也都收到通知。 */ public class DistributeServer { // 链接地址 private static final String connectStr = "hadoop01:2181,hadoop02:2181,hadoop03:2181,hadoop04:2181,hadoop05:2181"; // 链接请求时间 private static final int sessionTimeout = 4000; // 准备的测试服务动态感知的 ZK 父znode private static final String PARENT_NODE = "/server_test"; // 初始化 ZK 操作对象 static ZooKeeper zk = null; public static void main(String[] args) throws Exception { DistributeServer ds = new DistributeServer(); ds.getZookeeperConnect(); ds.registeServer("hadoop01"); Thread.sleep(10000); ds.registeServer("hadoop02"); Thread.sleep(Long.MAX_VALUE }/** * 拿到 zookeeper 进群的链接 * @throws Exception 链接 ZK 异常 */ public void getZookeeperConnect() throws Exception { zk = new ZooKeeper(connectStr, sessionTimeout, null); }/** * 服务器上线就注册，掉线就自动删除，所以创建的是临时顺序节点 * @param hostname 注册的主机名 * @throws Exception 判断节点是否存在或者创建节点异常 */ public void registeServer(String hostname) throws Exception { // ZK 没有链接，直接返回 if (zk == null) { System.out.println("zk 链接为空"); return; }// 看父znode 是否存在 Stat exists = zk.exists(PARENT_NODE, false); if (exists == null) { // 父znode 不存在,创建父 znode System.out.println("父znode 不存在,创建父 znode"); zk.create(PARENT_NODE, "server_parent_node".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); }zk.create(PARENT_NODE + "/" + hostname, (hostname + "_node").getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); System.out.println(hostname + " is online, start working..."); } }

3.2、客户端处理

package com.zc.bigdata.demo1; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import java.util.ArrayList; import java.util.List; /** * @作者： zc * @时间： 2021/3/4 10:11 * @描述：用来模拟用户端的操作： *连上 zookeeper 进群，实时获取服务器动态上下线的节点信息。 *总体思路就是每次该 server 节点下有增加或者减少节点数， *我就打印出来该 server 节点下的所有节点。 */ public class DistributeClient { // 链接地址 private static final String connectStr = "hadoop01:2181,hadoop02:2181,hadoop03:2181,hadoop04:2181,hadoop05:2181"; // 链接请求时间 private static final int sessionTimeout = 4000; // 准备的测试服务动态感知的 ZK 父znode private static final String PARENT_NODE = "/server_test"; // 初始化 ZK 操作对象 static ZooKeeper zk = null; public static void main(String[] args) throws Exception { DistributeClient dc = new DistributeClient(); dc.getZookeeperConnect(); Thread.sleep(Long.MAX_VALUE); }public void getZookeeperConnect() throws Exception { zk = new ZooKeeper(connectStr, sessionTimeout, event -> { //获取触发的事件类型 Watcher.Event.EventType type = event.getType(); //获取触发的事件路径 String path = event.getPath(); System.out.println(type+"@@@@@@@@@@@@@@@@@@"+path); try { // 获取父znode 下所有子节点，即是所有正上线服务的服务器节点 List> children = zk.getChildren(PARENT_NODE, true); List> servers = new ArrayList<>(); for (String ch : children) { // 取出父znode 下子节点的数据，放入到 list 里 String server = new String(zk.getData(PARENT_NODE + "/" + ch, false,null), "UTF-8"); servers.add(server); } // 打印 list 里面的元素 System.out.println(servers); } catch (Exception e) { e.printStackTrace(); } }); System.out.println("Client is online, start working..."); } }

3.3、运行效果启动 zk 服务：