网站做百度百科的好处,做淘宝客网站备案要怎么写,那个网站做扑克牌便宜,网站虚拟机从头做有影响吗Zookeeper的Java API操作一、先启动Zookeeper集群二、IDEA 环境搭建三、创建子节点四、获取子节点并监听节点变化五、判断 Znode 是否存在六、Watcher工作流程一、先启动Zookeeper集群
二、IDEA 环境搭建
1.创建一个Maven工程#xff1a;ZookeeperProject 2.在pom.xml文件添…
Zookeeper的Java API操作一、先启动Zookeeper集群二、IDEA 环境搭建三、创建子节点四、获取子节点并监听节点变化五、判断 Znode 是否存在六、Watcher工作流程一、先启动Zookeeper集群
二、IDEA 环境搭建
1.创建一个Maven工程ZookeeperProject 2.在pom.xml文件添加如下内容:
dependenciesdependencygroupIdjunit/groupIdartifactIdjunit/artifactIdversionRELEASE/version/dependencydependencygroupIdorg.apache.logging.log4j/groupIdartifactIdlog4j-core/artifactIdversion2.8.2/version/dependencydependencygroupIdorg.apache.zookeeper/groupIdartifactIdzookeeper/artifactIdversion3.5.7/version/dependency
/dependencies3.拷贝log4j.properties文件到项目根目录 需要在项目的 src/main/resources 目录下新建一个文件命名为“log4j.properties”在文件中填入:
log4j.rootLoggerINFO, stdout
log4j.appender.stdoutorg.apache.log4j.ConsoleAppender
log4j.appender.stdout.layoutorg.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern%d %p [%c] - %m%n
log4j.appender.logfileorg.apache.log4j.FileAppender
log4j.appender.logfile.Filetarget/spring.log
log4j.appender.logfile.layoutorg.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern%d %p [%c] - %m%n三、创建子节点
package com.hyj.zk;import org.apache.zookeeper.*;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;import java.io.IOException;
import java.util.List;public class CreateZnode {//注意:逗号前后不能有空格 指定Zookeeper服务器列表private static String connectStringhadoop102:2181,hadoop103:2181,hadoop104:2181;/* sessionTimeout指会话的超时时间是一个以“毫秒”为单位的整型值。在ZooKeeper中有会话的概念在一个会话周期内ZooKeeper客户端和服务端之间会通过心跳检测机制来维持会话的有效性一旦在sessionTimeout时间内没有进行有效的心跳检测会话就会失效。*/private static int sessionTimeout100000;private ZooKeeper zkClientnull;Beforepublic void init() throws IOException {//创建一个Zookeeper实例来连接Zookeeper服务器 Watcher会话监听器,服务端将会触发监听zkClientnew ZooKeeper(connectString, sessionTimeout, new Watcher() {Override //收到事件通知后的回调函数(用户的业务逻辑)public void process(WatchedEvent watchedEvent) {}});}Test //创建子节点public void create() throws InterruptedException, KeeperException {/* 参数 1要创建的节点的路径 参数 2节点数据(一个字节数组) 参数 3节点权限 ZooDefs.Ids.OPEN_ACL_UNSAFE表示以后对这个节点的任何操作都不受权限控制参数 4节点的类型 持久无序号节点PERSISTENT 持久带序号节点 PERSISTENT_SEQUENTIAL (persistent_sequential)短暂无序号节点EPHEMERAL 短暂带序号节点 EPHEMERAL_SEQUENTIAL (ephemeral_sequential)*/String s zkClient.create(/sanguo/xiyouji, sunwu.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);}Afterpublic void close() throws InterruptedException {zkClient.close();}
}从ZookKeeper系列watch机制截的一张图
监听的事件类型有:
None 客户端连接状态发生改变的时候,会收到None事件通知(如连接成功,连接失败,session会话过期等)NodeCreated 节点被创建NodeDeleted 节点被删除NodeDataChanged 节点数据被修改NodeChildrenChanged 子节点被创建或删除
四、获取子节点并监听节点变化
package com.hyj.zk;import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;import java.io.IOException;
import java.util.List;public class GetChildren {//注意:逗号前后不能有空格 指定Zookeeper服务器列表private static String connectStringhadoop102:2181,hadoop103:2181,hadoop104:2181;/* sessionTimeout指会话的超时时间是一个以“毫秒”为单位的整型值。在ZooKeeper中有会话的概念在一个会话周期内ZooKeeper客户端和服务端之间会通过心跳检测机制来维持会话的有效性一旦在sessionTimeout时间内没有进行有效的心跳检测会话就会失效。*/private static int sessionTimeout100000;private ZooKeeper zkClientnull;Beforepublic void init() throws IOException {//创建一个Zookeeper实例来连接Zookeeper服务器 Watcher会话监听器,服务端将会触发监听zkClientnew ZooKeeper(connectString, sessionTimeout, new Watcher() {Override //收到事件通知后的回调函数(用户的业务逻辑)public void process(WatchedEvent watchedEvent) {if(watchedEvent.getType() Event.EventType.None){if(watchedEvent.getState() Event.KeeperState.SyncConnected){System.out.println(Zookeeper连接成功!!!);}else if(watchedEvent.getState() Event.KeeperState.Disconnected){System.out.println(客户端和服务器的连接断开!!!);}else if (watchedEvent.getState() Event.KeeperState.Expired){System.out.println(session会话过期!!!);}}else{System.out.println(watchedEvent.getType() -- watchedEvent.getPath());try {//再次监听(注册一次,监听一次)ListString children zkClient.getChildren(/, true); //false表示不监听,true表示使用默认的watcherfor (String child : children) {System.out.println(child);}} catch (KeeperException | InterruptedException e) {e.printStackTrace();}}}});}Test //获取子节点并监听节点路径变化public void getChildren() throws InterruptedException, KeeperException {// 参数1: 表示监听的节点 参数2: true表示监听 ,false表示不监听ListString children zkClient.getChildren(/, true); //使用默认的watcherfor (String child : children) {System.out.println(child);}//延时阻塞Thread.sleep(Long.MAX_VALUE);}Test //获取子节点不监听节点路径变化public void getChildren2() throws InterruptedException, KeeperException {// 参数1: 表示监听的节点 参数2: true表示监听 ,false表示不监听ListString children zkClient.getChildren(/, false); //不注册watcherfor (String child : children) {System.out.println(child);}}Test //获取子节点并监听节点路径变化public void getChildren3() throws InterruptedException, KeeperException {// 参数1: 表示监听的节点 参数2: true表示监听 ,false表示不监听ListString children zkClient.getChildren(/, new Watcher() { //注册新的watcherOverride //收到事件通知后的回调函数(用户的业务逻辑)public void process(WatchedEvent watchedEvent) {System.out.println(watchedEvent.getType() ------ watchedEvent.getPath());try {ListString children zkClient.getChildren(/, false); //这里若是true它还是会使用默认的watcherfor (String child : children) {System.out.println(child);}} catch (KeeperException | InterruptedException e) {e.printStackTrace();}}});for (String child : children) {System.out.println(child);}//延时阻塞Thread.sleep(Long.MAX_VALUE);}Afterpublic void close() throws InterruptedException {zkClient.close();}
}
五、判断 Znode 是否存在
package com.hyj.zk;import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;import java.io.IOException;
import java.util.List;public class IsExistNode {//注意:逗号前后不能有空格 指定Zookeeper服务器列表private static String connectString hadoop102:2181,hadoop103:2181,hadoop104:2181;/* sessionTimeout指会话的超时时间是一个以“毫秒”为单位的整型值。在ZooKeeper中有会话的概念在一个会话周期内ZooKeeper客户端和服务端之间会通过心跳检测机制来维持会话的有效性一旦在sessionTimeout时间内没有进行有效的心跳检测会话就会失效。*/private static int sessionTimeout 100000;private ZooKeeper zkClient null;Beforepublic void init() throws IOException {//创建一个Zookeeper实例来连接Zookeeper服务器 Watcher会话监听器,服务端将会触发监听zkClient new ZooKeeper(connectString, sessionTimeout, new Watcher() {Override //收到事件通知后的回调函数(用户的业务逻辑)public void process(WatchedEvent watchedEvent) {if (watchedEvent.getType() ! Event.EventType.None) {System.out.println(watchedEvent.getType() -- watchedEvent.getPath());try {ListString children zkClient.getChildren(/, false); //false表示不监听,true表示使用默认的watcherfor (String child : children) {System.out.println(child);}} catch (KeeperException | InterruptedException e) {e.printStackTrace();}}}});}Testpublic void exist() throws InterruptedException, KeeperException {// 参数1: 表示要判断的节点 参数2: true表示监听 ,false表示不监听Stat stat zkClient.exists(/sanguo, false); //不注册watcherSystem.out.println(stat null ? not exist : exist);}Testpublic void exist2() throws InterruptedException, KeeperException {// 参数1: 表示要判断的节点 参数2: true表示监听此节点变化 ,false表示不监听Stat stat zkClient.exists(/sanguo, true); //使用默认的watcherSystem.out.println(stat null ? not exist : exist);//延时阻塞Thread.sleep(Long.MAX_VALUE);}Testpublic void exist3() throws InterruptedException, KeeperException {// 参数1: 表示要判断的节点 参数2: true表示监听此节点变化 ,false表示不监听Stat stat zkClient.exists(/sanguo, new Watcher() {Override //收到事件通知后的回调函数(用户的业务逻辑)public void process(WatchedEvent watchedEvent) { //注册新的watcherSystem.out.println(watchedEvent.getType() -- watchedEvent.getPath());try {ListString children zkClient.getChildren(/, false); //false表示不监听,true表示使用默认的watcherfor (String child : children) {System.out.println(child);}} catch (KeeperException | InterruptedException e) {e.printStackTrace();}}});System.out.println(stat null ? not exist : exist);//延时阻塞Thread.sleep(Long.MAX_VALUE);}Afterpublic void close() throws InterruptedException {zkClient.close();}
}
六、Watcher工作流程
Client 向 Zookeeper 服务端注册一个 Watcher ,同时将Watcher对象存储在客户端的 WatcherManager 中。当Zookeeper 服务端的一些指定事件触发了 Watcher 事件时,就会向客户端发送事件通知,客户端就会从WatcherManager 中取出对应的 Watcher 进行回调。
Watcher工作机制分为三个过程: 客户端注册Watcher 服务端处理Watcher 客户端回调Watcher
