Zookeeper Java客户端实例以及设置循环监听

Zookeeper Java 客户端实例

Zookeeper 的client端和服务端的代码没有分离，所以只用引用一个jar就行。这里版本最好与你服务端安装的zookeeper版本一致，以防一些兼容性问题。

<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.5.8</version>
</dependency>

链接到zookeeper的服务器：

    public static void connectZK() throws IOException, InterruptedException {
        final CountDownLatch countDown = new CountDownLatch(1);
        zk = new ZooKeeper(ZK_ADDRESS,5000,event -> {
            if (event.getState() == Watcher.Event.KeeperState.SyncConnected && event.getType() == Watcher.Event.EventType.None){
                countDown.countDown();
                System.out.println("connected");
            }
        });
        System.out.println("connecting please wait...");
        countDown.await();
    }

同步方式创建客户端的zk实例：

    public static void createNote() throws KeeperException, InterruptedException {
        zk.create("/AllenClient","Hello World".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("/AllenClient Created" );
    }

异步方式创建节点，客户端实例：

    public static void createNoteSync() throws KeeperException, InterruptedException {
        zk.create("/AllenClientSync","Hello World".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT,(rc,path,ctx,name)->{
            System.out.println(rc + " "+ path+" " + name+" " + ctx);

        },"context");
        System.out.println("/AllenClientSync Created" );

    }

式修改zk数据：

   public static void changeData() throws KeeperException, InterruptedException {
        Stat stat = new Stat();
        String path = "/AllenClient";
        byte [] data = zk.getData(path,false,stat);
        System.out.println("before the Change" + new String(data) );
        zk.setData(path,"Hello Allen".getBytes(),stat.getVersion());
        byte [] data1 = zk.getData(path,false,stat);
        System.out.println("after the Change" + new String(data1) );
    }

测试以上方法：

    public static final String ZK_ADDRESS="192.168.0.101:2181";
    public static ZooKeeper zk;
    public static void main(String[] args) {
        try {
            connectZK();
            createNote();
            changeData();
            createNoteSync();
        } catch (IOException | KeeperException e) {
            e.printstacktrace();
        } catch (InterruptedException e) {
            e.printstacktrace();
        }
    }

执行结构：

"C:\Program Files\Java\jdk1.8.0_202\bin\java.exe" "-javaagent:C:\Program Files\JetBrains\IntelliJ IDEA 2018.2.2\lib\idea_rt.jar=59861:C:\Program Files\JetBrains\IntelliJ IDEA 2018.2.2\bin" -Dfile.encoding=UTF-8 -classpath "C:\Program Files\Java\jdk1.8.0_202\jre\lib\charsets.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\deploy.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\ext\access-bridge-64.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\ext\cldrdata.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\ext\dnsns.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\ext\jaccess.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\ext\jfxrt.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\ext\localedata.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\ext\nashorn.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\ext\sunec.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\ext\sunjce_provider.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\ext\sunmscapi.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\ext\sunpkcs11.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\ext\zipfs.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\javaws.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\jce.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\jfr.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\jfxswt.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\jsse.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\management-agent.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\plugin.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\resources.jar;C:\Program Files\Java\jdk1.8.0_202\jre\lib\rt.jar;C:\Users\Administrator\IdeaProjects\javaAA_L1\zookeeperTrainning\target\classes;C:\Users\Administrator\.m2\repository\org\apache\zookeeper\zookeeper\3.5.8\zookeeper-3.5.8.jar;C:\Users\Administrator\.m2\repository\org\apache\zookeeper\zookeeper-jute\3.5.8\zookeeper-jute-3.5.8.jar;C:\Users\Administrator\.m2\repository\org\apache\yetus\audience-annotations\0.5.0\audience-annotations-0.5.0.jar;C:\Users\Administrator\.m2\repository\io\netty\netty-handler\4.1.48.Final\netty-handler-4.1.48.Final.jar;C:\Users\Administrator\.m2\repository\io\netty\netty-common\4.1.48.Final\netty-common-4.1.48.Final.jar;C:\Users\Administrator\.m2\repository\io\netty\netty-resolver\4.1.48.Final\netty-resolver-4.1.48.Final.jar;C:\Users\Administrator\.m2\repository\io\netty\netty-buffer\4.1.48.Final\netty-buffer-4.1.48.Final.jar;C:\Users\Administrator\.m2\repository\io\netty\netty-transport\4.1.48.Final\netty-transport-4.1.48.Final.jar;C:\Users\Administrator\.m2\repository\io\netty\netty-codec\4.1.48.Final\netty-codec-4.1.48.Final.jar;C:\Users\Administrator\.m2\repository\io\netty\netty-transport-native-epoll\4.1.48.Final\netty-transport-native-epoll-4.1.48.Final.jar;C:\Users\Administrator\.m2\repository\io\netty\netty-transport-native-unix-common\4.1.48.Final\netty-transport-native-unix-common-4.1.48.Final.jar;C:\Users\Administrator\.m2\repository\org\slf4j\slf4j-api\1.7.25\slf4j-api-1.7.25.jar;C:\Users\Administrator\.m2\repository\org\slf4j\slf4j-log4j12\1.7.25\slf4j-log4j12-1.7.25.jar;C:\Users\Administrator\.m2\repository\log4j\log4j\1.2.17\log4j-1.2.17.jar" allen.trainning.zk.ZKOfficialSimple
log4j:WARN No appenders Could be found for logger (org.apache.zookeeper.ZooKeeper).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
connecting please wait...
connected
/AllenClient Created
before the ChangeHello World
after the ChangeHello Allen
/AllenClientSync Created

Process finished with exit code 0

在zk上查看执行结构：

WatchedEvent state:SyncConnected type:None path:null
[zk: localhost:2181(CONNECTED) 0] ls /
[Allen, AllenClient, AllenClientSync, zookeeper]
[zk: localhost:2181(CONNECTED) 1] get /AllenClient
Hello Allen
[zk: localhost:2181(CONNECTED) 3] get /AllenClientSync
Hello World
[zk: localhost:2181(CONNECTED) 4]

设置循环监听

在节点上添加监听，并实现循环监听：

1. 引入一个jason与object相互转换的工具

    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
        <version>2.8.3</version>
    </dependency>

2. 在代码中定义一个Watcher

Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            if(event.getType()==Event.EventType.NodeDataChanged
            && event.getPath()!=null
            && event.getPath().equals(path)){
                System.out.println(event.getPath() + "date changed, please check !");
                byte [] data=null;
                try {
                     data = zk.getData(path,this,null);
                    Person allen = om.readValue(new String(data),Person.class);
                    System.out.println("Allen personal info changed to "+ allen.toString());
                } catch (Exception e) {
                    e.printstacktrace();
                }
            }
        }
    };

3. 在代码中定义一个用来装数据的实体类


class Person{
    private String name;
    private String gender;
    private int age;

    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public String getGender() {
        return gender;
    }
    public void setGender(String gender) {
        this.gender = gender;
    }
    public int getAge() {
        return age;
    }
    public void setAge(int age) {
        this.age = age;
    }
    public String toString(){
        return "name:"+ name +"  gender:"+ gender +"  age:"+ age;
    }

4. 设置循环监听节点


public static void listenNote() throws IOException, KeeperException, InterruptedException {
    String path = "/AllenInfo";
    Person allen =    new Person();
    allen.setName("Allen");
    allen.setGender("Female");
    allen.setAge(18);
    ObjectMapper om = new ObjectMapper();
    byte [] bytes = om.writeValueAsBytes(allen);
    zk.create(path,bytes,ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);

    #define Watcher

    byte [] data =zk.getData(path,watcher,null);
    Person allenNew = om.readValue(new String(data),Person.class);
    System.out.println("Allen personal info is "+ allenNew.toString());
    TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
}

测试方法：

   public static final String ZK_ADDRESS="192.168.0.101:2181";
    public static ZooKeeper zk;
    public static void main(String[] args) {
        try {
            connectZK();
            listenNote();
        } catch (IOException | KeeperException e) {
            e.printstacktrace();
        } catch (InterruptedException e) {
            e.printstacktrace();
        }
    }

执行结构：

当在zookeeper中去修改数据时候，代码会自动的打印修改结构：

打印结果：

log4j:WARN No appenders Could be found for logger (org.apache.zookeeper.ZooKeeper).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
connecting please wait...
connected
Allen personal info is name:Allen  gender:Female  age:18
/AllenInfodate changed, please check !
Allen personal info changed to name:Allen  gender:Female  age:19
/AllenInfodate changed, please check !
Allen personal info changed to name:Allen  gender:Female  age:20
/AllenInfodate changed, please check !
Allen personal info changed to name:Allen  gender:Female  age:21

Zookeeper Java客户端实例以及设置循环监听

Zookeeper Java 客户端实例

设置循环监听

相关推荐