12. Zookeeper 实战一: 监控节点动态上下线

亦凉 2022-02-20 08:39 504阅读 0赞

利用Zookeeper 临时型节点特性(当断开连接时, 创建的临时性节点自动销毁), 可实现监听集群节点动态上下线的功能. 所谓监听监听节点动态上下线是指, 当有节点新增或删除时, 监听端都会收到信息. zookeeper 实现的监听节点动态上下线特点:

*  通过向节点注册监听器, 来监听该节点子节点的创建和删除
 *  当子节点有创建和删除操作时, 监听端会接收到事件, 但是却不知道具体是新增或删除哪个节点, 需要重新查询所有子节点数据, 来进行更新列表
 *  使用局部监听器时,需要注意局部监听器只能监听一次事件. 当监听到事件后, 需要将监听器自身再次注册为监听器, 才能保证继续监听
 *  zk 监听类似于广播, 当一个节点注册了多个监听器时, 当有事件发生时, 事件会通知给所有监听器

## 1. 注册端开发 ##

### 1.1 创建全局监听器 ###

笔者习惯于也建议, 创建一个全局监听器类, 而不是采用内部类实现, 纵使笔者的全局事件监听器并未做任何处理. 笔者觉得这样会使代码更清晰.

public class RegisterGlobalWatcher implements Watcher { 
        @Override
        public void process(WatchedEvent event) { 
        }
    }

### 1.2 创建zk 客户端连接,并注册监听器 ###

笔者采用静态代码块儿来初始化zk 连接, 这样能保证连zk 连接只会被创建一次. 当结合spring 使用时, 只需要使用@Component 将此类的实例注册为sprin 容器的一个组件即可.

public class ZkRegisteClient { 
    
        private static Logger logger = LoggerFactory.getLogger(ZkRegisteClient.class);
    
        private static ZooKeeper zooKeeper;
    
        // 使用static 方式, 在类加载时, 只执行一次
        static { 
            // 创建zookeeper 连接时, 设置全局监听器
            try { 
                zooKeeper = new ZooKeeper(ZkConstant.SERVER, ZkConstant.SESSION_TIMEOUT, new RegisterGlobalWatcher());
            } catch (IOException e) { 
                throw new RuntimeException("创建连接失败");
            }
    
            //注册节点
            rigeste();
        }
    
        /** * @Description: 注册节点 * @author: zongf * @time: 17:30:07 */
        private static void rigeste(){ 
    
            try { 
                // 如果nodes 节点不存在, 则创建nodes 节点
                if(null == zooKeeper.exists(ZkConstant.PATH_NODES, false)){ 
                    zooKeeper.create(ZkConstant.PATH_NODES, "集群节点列表".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                }
    
                // 获取本机主机ip
                String address = InetAddressUtil.getHostIp();
    
                // 节点路径 TODO 根据自居需要设置在数据内容
                String nodePath = ZkConstant.PATH_NODES + "/" + address + "-";
    
                // 节点数据 TODO 根据自居需要设置在数据内容
                String data = "{\"host\":\"" + address + "\", time:\"" + getDateTime() + "\"}";
    
                // 如果节点存在则先删除
                Stat stat = zooKeeper.exists(nodePath, false);
                if(stat != null){ 
                    zooKeeper.delete(nodePath, stat.getVersion());
                }
    
                // 创建节点(由于笔者是一台机器,所以ip相同,因此采用了临时有序节点)
                String nodeName = zooKeeper.create(nodePath, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    
                logger.info("zookeeper 创建节点成功, 节点路径:{}, 节点名称:{}", nodePath, nodeName);
            } catch (Exception e) { 
                logger.error("Zookeeper 创建nodes节点异常", e);
            }
        }
    
        /** * @Description: 获取当前时间字符串. * @return: String * @author: zongf * @time: 17:37:04 */
        private static String getDateTime() { 
            LocalDateTime now = LocalDateTime.now();
            DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.S");
            return dtf.format(now);
        }
    
    }

## 2. 监听端开发 ##

### 2.1 创建全局监听器 ###

和注册端类似, 笔者也单独创建一个全局事件监听器.

public class MonitorGlobalWatcher implements Watcher { 
        @Override
        public void process(WatchedEvent event) { 
    
        }
    }

### 2.2 创建监控节点监听器 ###

笔者单独创建一个/nodes 节点的监听器, 来处理/nodes 节点下子节点发生变化的事件, 并不使用全局监听器. 需要注意的时, 局部监听器只能进行一次监听, 所以当监听到事件时, 需要再次将自身注册为监听器, 以保证能继续监听.

public class ZkNodesWatcher implements Watcher { 
    
        private static Logger LOGGER = LoggerFactory.getLogger(ZkMonitorClient.class);
    
        private ZooKeeper zooKeeper;
    
        public ZkNodesWatcher(ZooKeeper zooKeeper) { 
            this.zooKeeper = zooKeeper;
    
            // 初始化数据
            reInitData();
        }
    
        /** * @Description: 重新初始化数据 * @author: zongf * @time: 18:08:49 */
        public void reInitData() { 
            try { 
                // 获取所有子节点. 并再次注册监听, 监听器为自身.
                List<String> children = zooKeeper.getChildren(ZkConstant.PATH_NODES, this);
    
                // 提取所有子节点数据
                for (String child : children) { 
                    String childPath = ZkConstant.PATH_NODES + "/" + child;
                    byte[] data = zooKeeper.getData(childPath, false, null);
    
                    ZkConstant.availableNodes.clear();
                    ZkConstant.availableNodes.put(childPath, new String(data));
                }
    
                // 打印当前存活节点信息
                printNodes();
            } catch (Exception e) { 
                LOGGER.error("重新初始化队列异常", e);
            }
        }
    
        @Override
        public void process(WatchedEvent event) { 
            // 监听子节点发生变化事件(新增子节点和删除子节点时触发)
            if (Event.EventType.NodeChildrenChanged == event.getType()) { 
                reInitData();
            }
        }
    
        /** * @Description: 输出可用节点列表 * @author: zongf * @time: 18:08:27 */
        private static void printNodes() { 
            StringBuffer sb = new StringBuffer();
            sb.append("\n********** 当前可用节点列表: **********\n");
            ZkConstant.availableNodes.forEach((key,value) ->sb.append(key + " --> " + value));
            sb.append("\n");
            LOGGER.info(sb.toString());
        }
    }

### 2.3 创建监控客户端 ###

和注册端类似, 使用静态代码块儿初始化zk 连接, 保证zk 连接只初始化一次. 当整合spring 时, 直接添加@Component 注解修饰即可.

public class ZkMonitorClient { 
    
        private static Logger LOGGER = LoggerFactory.getLogger(ZkMonitorClient.class);
    
        private static ZooKeeper zooKeeper;
    
        static  { 
            // 创建zookeeper 连接时, 设置全局监听器
            try { 
                zooKeeper = new ZooKeeper(ZkConstant.SERVER, ZkConstant.SESSION_TIMEOUT, new MonitorGlobalWatcher());
            } catch (Exception e) { 
                LOGGER.error("连接zookeeper 异常", e);
            }
    
            //注册监听
            try { 
                zooKeeper.getChildren(ZkConstant.PATH_NODES, new ZkNodesWatcher(zooKeeper));
            } catch (Exception e) { 
                LOGGER.error("获取节点数据异常,节点路径:" + ZkConstant.PATH_NODES, e);
            }
        }
    }

## 3. 辅助工具类 ##

### 3.1 常量类 ###

由于是测试, 笔者采用常量类类存储一些常量信息. 生产中, 建议将常量类替换为属性配置文件.

public  class ZkConstant { 
    
        // zookeeper 服务地址
        public static final String SERVER = "127.0.0.1:2181";
    
        // zookeeper 连接超时时间
        public static int SESSION_TIMEOUT = 3000;
    
        // 节点路径
        public static String PATH_NODES = "/nodes";
    
        // 存储存活节点数据
        public static Map<String,String> availableNodes = new HashMap<>();
    
    }

### 3.2 获取主机Ip 工具类 ###

public class InetAddressUtil { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(InetAddressUtil.class);
    
    
        /** * @Description: 获取主机地址 * @return: String * @author: zongf * @time: 2019-03-11 14:47:53 */
        public static String getHostIp(){ 
    
            String realIp = null;
    
            try { 
                InetAddress address = InetAddress.getLocalHost();
    
                // 如果是回环网卡地址, 则获取ipv4 地址
                if (address.isLoopbackAddress()) { 
                    address = getInet4Address();
                }
    
                realIp = address.getHostAddress();
    
                LOGGER.info("获取主机ip地址成功, 主机ip地址:{}", address);
                return address.getHostAddress();
            } catch (Exception e) { 
                LOGGER.error("获取主机ip地址异常", e);
            }
    
            return realIp;
        }
    
        /** * @Description: 获取IPV4网络配置 * @return: InetAddress * @author: zongf * @time: 2019-03-11 14:09:17 */
        private static InetAddress getInet4Address() throws SocketException { 
            // 获取所有网卡信息
            Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();
            while (networkInterfaces.hasMoreElements()) { 
                NetworkInterface netInterface = (NetworkInterface) networkInterfaces.nextElement();
                Enumeration<InetAddress> addresses = netInterface.getInetAddresses();
                while (addresses.hasMoreElements()) { 
                    InetAddress ip = (InetAddress) addresses.nextElement();
                    if (ip instanceof Inet4Address) { 
                        return ip;
                    }
                }
            }
            return null;
        }
    }

## 4. 测试 ##

### 4.1 启动监听端 ###

public static void main(String[] args) throws InterruptedException { 
    
        // 创建监听客户端
        new ZkMonitorClient();
    
        // 线程休眠
        Thread.sleep(Integer.MAX_VALUE);
    }

### 4.2 启动注册端 ###

public static void main(String[] args) throws InterruptedException { 
    
        // 创建注册客户端
        new ZkRegisteClient();
    
        // 线程休眠
        Thread.sleep(Integer.MAX_VALUE);
    }

### 4.3 监控日志输出 ###

测试步骤:

1.  启动一个监听实例
2.  启动一个注册实例, 并观察监听实例控制台输出
3.  启动一个注册实例, 并观察监听实例控制台输出
4.  停止一个注册实例, 并观察监听实例控制台输出
5.  停止一个注册实例, 并观察监听实例控制台输出

# 启动第一个注册实例
    ********** 当前可用节点列表: **********
    /nodes/192.168.1.224-0000000020 --> { "host":"192.168.1.224", time:"2019-03-25 18:16:38.7"}
    
    # 启动第二个注册实例
    ********** 当前可用节点列表: **********
    /nodes/192.168.1.224-0000000021 --> { "host":"192.168.1.224", time:"2019-03-25 18:16:44.0"}
    /nodes/192.168.1.224-0000000020 --> { "host":"192.168.1.224", time:"2019-03-25 18:16:38.7"}
    
    # 启动第三个注册实例
    ********** 当前可用节点列表: **********
    /nodes/192.168.1.224-0000000022 --> { "host":"192.168.1.224", time:"2019-03-25 18:16:49.5"}
    /nodes/192.168.1.224-0000000021 --> { "host":"192.168.1.224", time:"2019-03-25 18:16:44.0"}
    /nodes/192.168.1.224-0000000020 --> { "host":"192.168.1.224", time:"2019-03-25 18:16:38.7"}
    
    # 停止一个注册实例
    ********** 当前可用节点列表: **********
    /nodes/192.168.1.224-0000000022 --> { "host":"192.168.1.224", time:"2019-03-25 18:16:49.5"}
    /nodes/192.168.1.224-0000000021 --> { "host":"192.168.1.224", time:"2019-03-25 18:16:44.0"}
    
    # 停止两个注册实例
    ********** 当前可用节点列表: **********
    /nodes/192.168.1.224-0000000022 --> { "host":"192.168.1.224", time:"2019-03-25 18:16:49.5"}