2. 程式人生 > >Zookeeper使用（二）--Java API

Zookeeper使用（二）--Java API

阿新 • • 發佈：2018-12-13

一、前言

　　上一篇部落格我們通過命令列來操作Zookeper的客戶端和服務端並進行相應的操作，這篇主要介紹如何通過API（JAVA）來操作Zookeeper。

二、開發環境配置

　　首先開啟Zookeeper服務端（上一篇部落格有具體的方法），方便客戶端連線。

　　配置開發環境環境可以有兩種方式：① 直接下載相關的依賴Jar包，然後在IDE中新增依賴 ② 建立maven專案，使用maven進行依賴管理。

　　① 手動新增依賴至IDE

　　步驟一：點選這裡下載對應版本的Jar包，包括（jar、javadoc.jar、sources.jar），筆者對應下載的Zookeeper3.4.6版本。

　　步驟二：開啟IDE（筆者使用eclispe），新建名為zookeeper_examples_none_maven的java專案。由於需要單獨新增依賴，為了方便管理，筆者在專案下新建了jar資料夾，用於存放本專案的jar包（將步驟一下載的3個jar包存放至此資料夾下）。

　　

　　步驟三：在eclipse中新增依賴

　　

　　步驟四：新建包、Java類進行測試

　　Zookeeper_Constructor_Usage_Simple.java　　

複製程式碼

package com.hust.grid.leesf.examples;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;

public class Zookeeper_Constructor_Usage_Simple implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);

    @Override
    public void process(WatchedEvent event) {
        System.out.println("Receive watched event : " + event);
        if (KeeperState.SyncConnected == event.getState()) {
            connectedSemaphore.countDown();
        }
    }

    public static void main(String[] args) throws IOException {
        ZooKeeper zookeeper = new ZooKeeper("127.0.0.1:2181", 5000, new Zookeeper_Constructor_Usage_Simple());
        System.out.println(zookeeper.getState());
        try {
            connectedSemaphore.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Zookeeper session established");
    }
}

複製程式碼

　　執行結果如下　　

複製程式碼

Exception in thread "main" java.lang.NoClassDefFoundError: org/slf4j/LoggerFactory
    at org.apache.zookeeper.ZooKeeper.<clinit>(ZooKeeper.java:94)
    at com.hust.grid.leesf.examples.Zookeeper_Constructor_Usage_Simple.main(Zookeeper_Constructor_Usage_Simple.java:23)
Caused by: java.lang.ClassNotFoundException: org.slf4j.LoggerFactory
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    ... 2 more

複製程式碼

　　結果表明缺失LoggerFactory類，經筆者查閱資料，只需將zookeeper的lib資料夾下log4j-1.2.16.jar、slf4j-api-1.6.1.jar放如zookeeper_examples_none_maven的jar資料夾下，然後再次將其新增至IDE即可。

　　再次執行結果如下

CONNECTING
Receive watched event : WatchedEvent state:SyncConnected type:None path:null
Zookeeper session established

　　表示客戶端已經成功連線至伺服器了。

　　可以看到方法一相對而言比較麻煩，需要手動管理不同的依賴jar包，可以採用更成熟的依賴管理方法，即使用maven來管理Jar包。

　　② 使用maven管理依賴

　　步驟一：新建maven專案

　　

　　

　　

　　步驟二：配置pom.xml檔案如下　

複製程式碼

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.hust.grid.leesf</groupId>
  <artifactId>zookeeper_examples</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>zookeeper_examples</name>
  <url>http://maven.apache.org</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>

  <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>3.8.1</version>
            <scope>test</scope>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.6</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/log4j/log4j -->
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.16</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api -->

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.6.1</version>
        </dependency>

    </dependencies>
</project>

複製程式碼

　　步驟三：新建java類進行測試

　　Zookeeper_Constructor_Usage_Simple.java，程式碼同上。

　　執行結果如下　　

CONNECTING
Receive watched event : WatchedEvent state:SyncConnected type:None path:null
Zookeeper session established

　　結果也表示客戶端已經成功連線至伺服器。

三、操作示例

　　3.1  建立節點

　　建立節點有非同步和同步兩種方式。無論是非同步或者同步，Zookeeper都不支援遞迴呼叫，即無法在父節點不存在的情況下建立一個子節點，如在/zk-ephemeral節點不存在的情況下建立/zk-ephemeral/ch1節點；並且如果一個節點已經存在，那麼建立同名節點時，會丟擲NodeExistsException異常。

　　① 同步方式

複製程式碼

package com.hust.grid.leesf.examples;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class Zookeeper_Create_API_Sync_Usage implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    
    public static void main(String[] args) throws Exception {
        ZooKeeper zookeeper = new ZooKeeper("127.0.0.1:2181", 5000, new Zookeeper_Create_API_Sync_Usage());
        System.out.println(zookeeper.getState());
        connectedSemaphore.await();
        
        String path1 = zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);    
        System.out.println("Success create znode: " + path1);
        
        String path2 = zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);    
        System.out.println("Success create znode: " + path2);
    }

    public void process(WatchedEvent event) {
        if (KeeperState.SyncConnected == event.getState()) {
            connectedSemaphore.countDown();
        }
    }
}

複製程式碼

　　執行結果如下　　

CONNECTING
Success create znode: /zk-test-ephemeral-
Success create znode: /zk-test-ephemeral-0000000043

　　結果表明已經成功建立了臨時節點和臨時順序節點，在建立順序節點時，系統會在後面自動增加一串數字。

　　② 非同步方式

　　使用非同步方式於同步方式的區別在於節點的建立過程（包括網路通訊和服務端的節點建立過程）是非同步的，在同步介面呼叫過程中，開發者需要關注介面丟擲異常的可能，但是在非同步介面中，介面本身不會丟擲異常，所有異常都會在回撥函式中通過Result Code來體現。

複製程式碼

package com.hust.grid.leesf.examples;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;

public class Zookeeper_Create_API_ASync_Usage implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);

    public static void main(String[] args) throws Exception {
        ZooKeeper zookeeper = new ZooKeeper("127.0.0.1:2181", 5000, new Zookeeper_Create_API_ASync_Usage());
        System.out.println(zookeeper.getState());
        connectedSemaphore.await();

        zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
                new IStringCallback(), "I am context. ");

        zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
                new IStringCallback(), "I am context. ");

        zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,
                new IStringCallback(), "I am context. ");
        Thread.sleep(Integer.MAX_VALUE);
    }

    public void process(WatchedEvent event) {
        if (KeeperState.SyncConnected == event.getState()) {
            connectedSemaphore.countDown();
        }
    }
}

class IStringCallback implements AsyncCallback.StringCallback {
    public void processResult(int rc, String path, Object ctx, String name) {
        System.out.println("Create path result: [" + rc + ", " + path + ", " + ctx + ", real path name: " + name);
    }
}

複製程式碼

　　執行結果如下　

CONNECTING
Create path result: [0, /zk-test-ephemeral-, I am context. , real path name: /zk-test-ephemeral-
Create path result: [-110, /zk-test-ephemeral-, I am context. , real path name: null
Create path result: [0, /zk-test-ephemeral-, I am context. , real path name: /zk-test-ephemeral-0000000045

　　結果表明已經成功使用非同步方式建立了相應節點。

　　3.2 刪除節點　

　　只允許刪除葉子節點，即一個節點如果有子節點，那麼該節點將無法直接刪除，必須先刪掉其所有子節點。同樣也有同步和非同步兩種方式。　

　　① 同步方式

複製程式碼

package com.hust.grid.leesf.examples;

import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class Delete_API_Sync_Usage implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    private static ZooKeeper zk;

    public static void main(String[] args) throws Exception {
        String path = "/zk-book";
        zk = new ZooKeeper("127.0.0.1:2181", 5000,
                new Delete_API_Sync_Usage());
        connectedSemaphore.await();

        zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("success create znode: " + path);
        zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("success create znode: " + path + "/c1");
        try {
            zk.delete(path, -1);
        } catch (Exception e) {
            System.out.println("fail to delete znode: " + path);
        }
        
        zk.delete(path + "/c1", -1);
        System.out.println("success delete znode: " + path + "/c1");
        zk.delete(path, -1);        
        System.out.println("success delete znode: " + path);

        Thread.sleep(Integer.MAX_VALUE);
    }

    public void process(WatchedEvent event) {
        if (KeeperState.SyncConnected == event.getState()) {
            if (EventType.None == event.getType() && null == event.getPath()) {
                connectedSemaphore.countDown();
            }
        }
    }
}

複製程式碼

　　執行結果如下　

success create znode: /zk-book
success create znode: /zk-book/c1
fail to delete znode: /zk-book
success delete znode: /zk-book/c1
success delete znode: /zk-book

　　結果表明若節點有子節點，則無法將其刪除，必須先刪除其所有子節點。

　　② 非同步方式　

複製程式碼

package com.hust.grid.leesf.examples;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;

public class Delete_API_ASync_Usage implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    private static ZooKeeper zk;

    public static void main(String[] args) throws Exception {
        String path = "/zk-book";
        zk = new ZooKeeper("127.0.0.1:2181", 5000,
                new Delete_API_ASync_Usage());
        connectedSemaphore.await();

        zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("success create znode: " + path);
        zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("success create znode: " + path + "/c1");

        zk.delete(path, -1, new IVoidCallback(), null);        
        zk.delete(path + "/c1", -1, new IVoidCallback(), null);
        zk.delete(path, -1, new IVoidCallback(), null);        

        Thread.sleep(Integer.MAX_VALUE);
    }

    public void process(WatchedEvent event) {
        if (KeeperState.SyncConnected == event.getState()) {
            if (EventType.None == event.getType() && null == event.getPath()) {
                connectedSemaphore.countDown();
            }
        }
    }
}

class IVoidCallback implements AsyncCallback.VoidCallback {
    public void processResult(int rc, String path, Object ctx) {
        System.out.println(rc + ", " + path + ", " + ctx);
    }
}

複製程式碼

　　執行結果如下　　

success create znode: /zk-book
success create znode: /zk-book/c1
-111, /zk-book, null
0, /zk-book/c1, null
0, /zk-book, null

　　結果結果表明第一次刪除/zk-book的時異常，ResultCode為-111。

　　3.3 子節點獲取

　　讀取節點的子節點列表，同樣可以使用同步和非同步的方式進行操作。

　　① 同步方式　　

複製程式碼

package com.hust.grid.leesf.examples;

import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;

public class Zookeeper_GetChildren_API_Sync_Usage implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    private static ZooKeeper zk = null;

    public static void main(String[] args) throws Exception {
        String path = "/zk-book-1";
        zk = new ZooKeeper("127.0.0.1:2181", 5000, new Zookeeper_GetChildren_API_Sync_Usage());
        connectedSemaphore.await();

        zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("success create znode: " + path);
        zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("success create znode: " + path + "/c1");
        List<String> childrenList = zk.getChildren(path, true);
        System.out.println(childrenList);

        zk.create(path + "/c2", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("success create znode: " + path + "/c2");
        Thread.sleep(1000);
        zk.create(path + "/c3", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("success create znode: " + path + "/c3");
        Thread.sleep(Integer.MAX_VALUE);
    }

    public void process(WatchedEvent event) {
        if (KeeperState.SyncConnected == event.getState()) {
            if (EventType.None == event.getType() && null == event.getPath()) {
                connectedSemaphore.countDown();
            } else if (event.getType() == EventType.NodeChildrenChanged) {
                try {
                    System.out.println("ReGet Child:" + zk.getChildren(event.getPath(), true));
                } catch (Exception e) {
                }
            }
        }
    }
}

複製程式碼

　　執行結果如下　

複製程式碼

success create znode: /zk-book-1
success create znode: /zk-book-1/c1
[c1]
success create znode: /zk-book-1/c2
ReGet Child:[c1, c2]
success create znode: /zk-book-1/c3
ReGet Child:[c3, c1, c2]

複製程式碼

　　值得注意的是，Watcher通知是一次性的，即一旦觸發一次通知後，該Watcher就失效了，因此客戶端需要反覆註冊Watcher，即程式中在process裡面又註冊了Watcher，否則，將無法獲取c3節點的建立而導致子節點變化的事件。

　　② 非同步方式　　

複製程式碼

package com.hust.grid.leesf.examples;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class Zookeeper_GetChildren_API_ASync_Usage implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    private static ZooKeeper zk = null;

    public static void main(String[] args) throws Exception {
        String path = "/zk-book";
        zk = new ZooKeeper("127.0.0.1:2181", 5000, new Zookeeper_GetChildren_API_ASync_Usage());
        connectedSemaphore.await();
        zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("success create znode: " + path);
        zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("success create znode: " + path + "/c1");

        zk.getChildren(path, true, new IChildren2Callback(), null);

        zk.create(path + "/c2", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("success create znode: " + path + "/c2");

        Thread.sleep(Integer.MAX_VALUE);
    }

    public void process(WatchedEvent event) {
        if (KeeperState.SyncConnected == event.getState()) {
            if (EventType.None == event.getType() && null == event.getPath()) {
                connectedSemaphore.countDown();
            } else if (event.getType() == EventType.NodeChildrenChanged) {
                try {
                    System.out.println("ReGet Child:" + zk.getChildren(event.getPath(), true));
                } catch (Exception e) {
                }
            }
        }
    }
}

class IChildren2Callback implements AsyncCallback.Children2Callback {
    public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
        System.out.println("Get Children znode result: [response code: " + rc + ", param path: " + path + ", ctx: "
                + ctx + ", children list: " + children + ", stat: " + stat);
    }
}

複製程式碼

　　執行結果如下　　

複製程式碼

success create znode: /zk-book
success create znode: /zk-book/c1
Get Children znode result: [response code: 0, param path: /zk-book, ctx: null, children list: [c1], stat: 2901,2901,1478226062843,1478226062843,0,1,0,0,0,1,2902

success create znode: /zk-book/c2
ReGet Child:[c1, c2]

複製程式碼

　　結果表示通過非同步的方式可以獲取子節點資訊。

　　3.4 資料節點獲取

　　對於節點的資料獲取，同樣存在同步和非同步兩種方式。

　　① 同步方式　　　

複製程式碼

package com.hust.grid.leesf.examples;

import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class GetData_API_Sync_Usage implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    private static ZooKeeper zk = null;
    private static Stat stat = new Stat();

    public static void main(String[] args) throws Exception {
        String path = "/zk-book";
        zk = new ZooKeeper("127.0.0.1:2181", 5000, 
                new GetData_API_Sync_Usage());
        connectedSemaphore.await();
        zk.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("success create znode: " + path);
        
        System.out.println("the data of znode " + path + " is : " + new String(zk.getData(path, true, stat)));
        System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion());
        
        zk.setData(path, "123".getBytes(), -1);

        Thread.sleep(Integer.MAX_VALUE);
    }

    public void process(WatchedEvent event) {
        if (KeeperState.SyncConnected == event.getState()) {
            if (EventType.None == event.getType() && null == event.getPath()) {
                connectedSemaphore.countDown();
            } else if (event.getType() == EventType.NodeDataChanged) {
                try {
                    System.out.println("the data of znode " + event.getPath() + " is : " + new String(zk.getData(event.getPath(), true, stat)));
                    System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion());
                } catch (Exception e) {
                }
            }
        }
    }
}

複製程式碼

　　執行結果如下　　

success create znode: /zk-book
the data of /zk-book is : 123
czxID: 2924, mzxID: 2924, version: 0
the data of /zk-book is : 123
czxID: 2924, mzxID: 2925, version: 1

　　結果表明可以使用getData函式獲取節點的資料。

　　② 非同步方式　　

複製程式碼

package com.hust.grid.leesf.examples;

import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class GetData_API_ASync_Usage implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    private static ZooKeeper zk;

    public static void main(String[] args) throws Exception {
        String path = "/zk-book";
        zk = new ZooKeeper("127.0.0.1:2181", 5000, 
                new GetData_API_ASync_Usage());
        connectedSemaphore.await();

        zk.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("success create znode: " + path);
        
        zk.getData(path, true, new IDataCallback(), null);

        zk.setData(path, "123".getBytes(), -1);

        Thread.sleep(Integer.MAX_VALUE);
    }

    public void process(WatchedEvent event) {
        if (KeeperState.SyncConnected == event.getState()) {
            if (EventType.None == event.getType() && null == event.getPath()) {
                connectedSemaphore.countDown();
            } else if (event.getType() == EventType.NodeDataChanged) {
                try {
                    zk.getData(event.getPath(), true, new IDataCallback(), null);
                } catch (Exception e) {
                }
            }
        }
    }
}

class IDataCallback implements AsyncCallback.DataCallback {
    public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
        System.out.println("rc: " + rc + ", path: " + path + ", data: " + new String(data));
        System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion());
    }
}

複製程式碼

　　執行結果如下　　

success create znode: /zk-book
rc: 0, path: /zk-book, data: 123
czxID: 2932, mzxID: 2932, version: 0
rc: 0, path: /zk-book, data: 123
czxID: 2932, mzxID: 2933, version: 1

　　結果表明採用非同步方式同樣可方便獲取節點的資料。

　　3.5 更新資料

　　在更新資料時，setData方法存在一個version引數，其用於指定節點的資料版本，表明本次更新操作是針對指定的資料版本進行的，但是，在getData方法中，並沒有提供根據指定資料版本來獲取資料的介面，那麼，這裡為何要指定資料更新版本呢，這裡方便理解，可以等效於CAS（compare and swap），對於值V，每次更新之前都會比較其值是否是預期值A，只有符合預期，才會將V原子化地更新到新值B。Zookeeper的setData介面中的version引數可以對應預期值，表明是針對哪個資料版本進行更新，假如一個客戶端試圖進行更新操作，它會攜帶上次獲取到的version值進行更新，而如果這段時間內，Zookeeper伺服器上該節點的資料已經被其他客戶端更新，那麼其資料版本也會相應更新，而客戶端攜帶的version將無法匹配，無法更新成功，因此可以有效地避免分散式更新的併發問題。

　　① 同步方式　　

複製程式碼

package com.hust.grid.leesf.examples;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class SetData_API_Sync_Usage implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    private static ZooKeeper zk;

    public static void main(String[] args) throws Exception {
        String path = "/zk-book";
        zk = new ZooKeeper("127.0.0.1:2181", 5000, new SetData_API_Sync_Usage());
        connectedSemaphore.await();

        zk.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("success create znode: " + path);
        zk.getData(path, true, null);

        Stat stat = zk.setData(path, "456".getBytes(), -1);
        System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion());
        Stat stat2 = zk.setData(path, "456".getBytes(), stat.getVersion());
        System.out.println("czxID: " + stat2.getCzxid() + ", mzxID: " + stat2.getMzxid() + ", version: " + stat2.getVersion());
        try {
            zk.setData(path, "456".getBytes(), stat.getVersion());
        } catch (KeeperException e) {
            System.out.println("Error: " + e.code() + "," + e.getMessage());
        }
        Thread.sleep(Integer.MAX_VALUE);
    }

    public void process(WatchedEvent event) {
        if (KeeperState.SyncConnected == event.getState()) {
            if (EventType.None == event.getType() && null == event.getPath()) {
                connectedSemaphore.countDown();
            }
        }
    }
}

複製程式碼

　　執行結果如下　

success create znode: /zk-book
czxID: 2936, mzxID: 2937, version: 1
czxID: 2936, mzxID: 2938, version: 2
Error: BADVERSION,KeeperErrorCode = BadVersion for /zk-book

　　結果表明由於攜帶的資料版本不正確，而無法成功更新節點。其中，setData中的version引數設定-1含義為客戶端需要基於資料的最新版本進行更新操作。

　　② 非同步方式　　

複製程式碼

package com.hust.grid.leesf.examples;

import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class SetData_API_ASync_Usage implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    private static ZooKeeper zk;

    public static void main(String[] args) throws Exception {
        String path = "/zk-book";
        zk = new ZooKeeper("127.0.0.1:2181", 5000, new SetData_API_ASync_Usage());
        connectedSemaphore.await();

        zk.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("success create znode: " + path);
        zk.setData(path, "456".getBytes(), -1, new IStatCallback(), null);

        Thread.sleep(Integer.MAX_VALUE);
    }

    public void process(WatchedEvent event) {
        if (KeeperState.SyncConnected == event.getState()) {
            if (EventType.None == event.getType() && null == event.getPath()) {
                connectedSemaphore.countDown();
            }
        }
    }
}

class IStatCallback implements AsyncCallback.StatCallback {
    public void processResult(int rc, String path, Object ctx, Stat stat) {
        System.out.println("rc: " + rc + ", path: " + path + ", stat: " + stat);
    }
}

複製程式碼

　　執行結果如下　　

success create znode: /zk-book
rc: 0, path: /zk-book, stat: 2942,2943,1478228414526,1478228414545,1,0,0,96876700808708136,3,0,2942

　　rc（ResultCode）為0，表明成功更新節點資料。

　　3.6 檢測節點是否存在

　　在呼叫介面時註冊Watcher的話，還可以對節點是否存在進行監聽，一旦節點被建立、被刪除、資料更新，都會通知客戶端。

　　① 同步方式　

複製程式碼

package com.hust.grid.leesf.examples;

import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class Exist_API_Sync_Usage implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    private static ZooKeeper zk;

    public static void main(String[] args) throws Exception {
        String path = "/zk-book";
        zk = new ZooKeeper("127.0.0.1:2181", 5000, //
                new Exist_API_Sync_Usage());
        connectedSemaphore.await();

        zk.exists(path, true);

        zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zk.setData(path, "123".getBytes(), -1);

        zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("success create znode: " + path + "/c1");

        zk.delete(path + "/c1", -1);
        zk.delete(path, -1);

        Thread.sleep(Integer.MAX_VALUE);
    }

    public void process(WatchedEvent event) {
        try {
            if (KeeperState.SyncConnected == event.getState()) {
                if (EventType.None == event.getType() && null == event.getPath()) {
                    connectedSemaphore.countDown();
                } else if (EventType.NodeCreated == event.getType()) {
                    System.out.println("success create znode: " + event.getPath());
                    zk.exists(event.getPath(), true);
                } else if (EventType.NodeDeleted == event.getType()) {
                    System.out.println("success delete znode: " + event.getPath());
                    zk.exists(event.getPath(), true);
                } else if (EventType.NodeDataChanged == event.getType()) {
                    System.out.println("data changed of znode: " + event.getPath());
                    zk.exists(event.getPath(), true);
                }
            }
        } catch (Exception e) {
        }
    }
}

複製程式碼

　　執行結果如下　

success create znode: /zk-book
data changed of znode: /zk-book
success create znode: /zk-book/c1
success delete znode: /zk-book

　　結果表明：

　　· 無論節點是否存在，都可以通過exists介面註冊Watcher。

　　· 註冊的Watcher，對節點建立、刪除、資料更新事件進行監聽。

　　· 對於指定節點的子節點的各種變化，不會通知客戶端。

　　② 非同步方式　　

複製程式碼

package com.hust.grid.leesf.examples;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class Exist_API_ASync_Usage implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    private static ZooKeeper zk;

    public static void main(String[] args) throws Exception {
        String path = "/zk-book";
        zk = new ZooKeeper("127.0.0.1:2181", 5000, 
                new Exist_API_ASync_Usage());
        connectedSemaphore.await();

        zk.exists(path, true, new IIStatCallback(), null);

        zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zk.setData(path, "123".getBytes(), -1);

        zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("success create znode: " + path + "/c1");

        zk.delete(path + "/c1", -1);
        zk.delete(path, -1);

        Thread.sleep(Integer.MAX_VALUE);
    }

    public void process(WatchedEvent event) {
        try {
            if (KeeperState.SyncConnected == event.getState()) {
                if (EventType.None == event.getType() && null == event.getPath()) {
                    connectedSemaphore.countDown();
                } else if (EventType.NodeCreated == event.getType()) {
                    System.out.println("success create znode: " + event.getPath());
                    zk.exists(event.getPath(), true, new IIStatCallback(), null);
                } else if (EventType.NodeDeleted == event.getType()) {
                    System.out.println("success delete znode: " + event.getPath());
                    zk.exists(event.getPath(), true, new IIStatCallback(), null);
                } else if (EventType.NodeDataChanged == event.getType()) {
                    System.out.println("data changed of znode: " + event.getPath());
                    zk.exists(event.getPath(), true, new IIStatCallback(), null);
                }
            }
        } catch (Exception e) {
        }
    }
}

class IIStatCallback implements AsyncCallback.StatCallback {
    public void processResult(int rc, String path, Object ctx, Stat stat) {
        System.out.println("rc: " + rc + ", path: " + path + ", stat: " + stat);
    }    
}

複製程式碼

　　執行結果如下

複製程式碼

rc: -101, path: /zk-book, stat: null
success create znode: /zk-book
rc: 0, path: /zk-book, stat: 2974,2974,1478229717889,1478229717889,0,0,0,0,0,0,2974

data changed of znode: /zk-book
rc: 0, path: /zk-book, stat: 2974,2975,1478229717889,1478229717922,1,0,0,0,3,0,2974

success create znode: /zk-book/c1
success delete znode: /zk-book
rc: -101, path: /zk-book, stat: null

複製程式碼

　　結果表明當節點不存在時，其rc（ResultCode）為-101。

　　3.7 許可權控制

　　通過設定Zookeeper伺服器上資料節點的ACL控制，就可以對其客戶端對該資料節點的訪問許可權：如果符合ACL控制，則可以進行訪問，否則無法訪問。

　　① 使用無許可權資訊的Zookeeper會話訪問含許可權資訊的資料節點　　

複製程式碼

package com.hust.grid.leesf.examples;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class AuthSample_Get {
    final static String PATH = "/zk-book-auth_test";

    public static void main(String[] args) throws Exception {

        ZooKeeper zookeeper1 = new ZooKeeper("127.0.0.1:2181", 5000, null);
        zookeeper1.addAuthInfo("digest", "foo:true".getBytes());
        zookeeper1.create(PATH, "init".getBytes(), Ids.CREATOR_ALL_ACL, CreateMode.EPHEMERAL);
        System.out.println("success create znode: " + PATH);
        ZooKeeper zookeeper2 = new ZooKeeper("127.0.0.1:2181", 5000, null);
        zookeeper2.getData(PATH, false, null);
    }
}

複製程式碼

　　執行結果如下　

複製程式碼

success create znode: /zk-book-auth_test
Exception in thread "main" org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth for /zk-book-auth_test
    at org.apache.zookeeper.KeeperException.create(KeeperException.java:113)
    at org.apache.zookeeper.KeeperException.create(KeeperException.java:51)
    at org.apache.zookeeper.ZooKeeper.getData(ZooKeeper.java:1155)
    at org.apache.zookeeper.ZooKeeper.getData(ZooKeeper.java:1184)
    at com.hust.grid.leesf.examples.AuthSample_Get.main(AuthSample_Get.java:17)

複製程式碼

　　表示許可權不夠，不能進行操作。

　　② 刪除帶許可權控制的節點　　

複製程式碼

package com.hust.grid.leesf.examples;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class AuthSample_Delete {
    final static String PATH = "/zk-book-auth_test";
    final static String PATH2 = "/zk-book-auth_test/child";

    public static void main(String[] args) throws Exception {
        ZooKeeper zookeeper1 = new ZooKeeper("127.0.0.1:2181", 5000, null);
        zookeeper1.addAuthInfo("digest", "foo:true".getBytes());
        zookeeper1.create(PATH, "init".getBytes(), Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
        
        zookeeper1.create(PATH2, "init".getBytes(), Ids.CREATOR_ALL_ACL, CreateMode.EPHEMERAL);

        try {
            ZooKeeper zookeeper2 = new ZooKeeper("127.0.0.1:2181", 5000, null);
            zookeeper2.delete(PATH2, -1);
        } catch (Exception e) {
            System.out.println("fail to delete: " + e.getMessage());
        }

        ZooKeeper zookeeper3 = new ZooKeeper("127.0.0.1:2181", 5000, null);
        zookeeper3.addAuthInfo("digest", "foo:true".getBytes());
        zookeeper3.delete(PATH2, -1);
        System.out.println("success delete znode: " + PATH2);

        ZooKeeper zookeeper4 = new ZooKeeper("127.00.1:2181", 5000, null);
        zookeeper4.delete(PATH, -1);
        System.out.println("success delete znode: " + PATH);
    }
}

複製程式碼

　　執行結果如下　　

fail to delete: KeeperErrorCode = NoAuth for /zk-book-auth_test/child
success delete znode: /zk-book-auth_test/child
success delete znode: /zk-book-auth_test

　　結果表明若沒有許可權，則無法刪除節點。

四、總結

　　基於原生態的JAVA API的呼叫相對較簡單，筆者後續會對原始碼進行分析。本部落格的所有程式碼也同步上傳至github，也謝謝各位園友的觀看～

出處：http://www.cnblogs.com/leesf456/ 

相關推薦

Zookeeper使用（二）--Java API

一、前言 　　上一篇部落格我們通過命令列來操作Zookeper的客戶端和服務端並進行相應的操作，這篇主要介紹如何通過API（JAVA）來操作Zookeeper。 二、開發環境配置 　　首先開啟Zookeeper服務端（上一篇部落格有具體的方法），方便客戶端連線。 　　配置開發環境環境

ElasticSearch6（二）-- Java API連接es

RR work monitor 才會 你會 art 端口 有用 erro 此ElasticSearch系列基於最新版的6.2.4版本。 一、pom.xml依賴 <dependencies> <dependency>

zookeeper（四）——Java的API、Curator、watcher

一、JavaAPI提供ZooKeeper新增、查詢、修改、刪除節點操作 pom檔案： <dependency> <groupId>org.apache.zookeeper</groupId> <artifact

Zookeeper--（二）配置文件講解、客戶端使用

臨時 結束 java .com per 客戶端使用 .cn 事務 客戶端 Java操作zookeeper 創建節點 創建子節點 CreateMode.EPHEMETAL臨時節點保證本次回話有效，本次會話結束臨時節點消失，可以實現分布式事務鎖

HBase概念學習（四）Java API之掃描和過濾器

影響 產生 str 實例 分享 get 批量 java api track HBase主要的CRUD操作就不多介紹了，無非就是Put，Get。Delete三個類的運用。 本文相當於是閱讀HBase權威指南的總結。 一、掃描（Scan） 如今看一下掃

Java學習筆記（二）---java 修飾符

技術 外部類 blog 訪問權限 定義 log 發生 繼承 指向 一、java 修飾符 Java語言提供了很多修飾符，主要分為以下兩類： 訪問修飾符 非訪問修飾符 1、訪問控制修飾符 Java中，可以使用訪問控制符來保護對類、變量、方法和構造方法的訪問。Javav支持

springcloud實踐（二）之api網關：zuul

actor 控制 patch isp where url 大文件上傳 html base zuul是什麽? front door. API Gateway.Zuul is a JVM based router and server side load balancer b

（二）Java數組的使用

align als 使用 gpo out let 有序數組 return height Java數組 　　無序數組插入刪除查詢操作： public class ArrayList { private static int[] intArray; priv

（二）java集合框架綜述

args 目的 res arraylist 意義 lec 多少 多條 無序集合 一集合框架圖 說明：對於以上的框架圖有如下幾點說明 1.所有集合類都位於java.util包下。Java的集合類主要由兩個接口派生而出：Collection和Map，Colle

JavaSE學習總結（二）——Java語言基礎

默認 ase oid 64位 pac 交互 log 類型 blank 目錄 一、Java程序預覽 1.1、Hello World 1.2、Eclipse使用 1.2.1、項目位置 1.2.2、重置窗口布局 1.2.3、設置字體 1.2.4、修改控制臺字體 1.2.4

MongoDB學習之（二）java連接

代碼 core UC mongo bte 就是 ava jar包 lan 上一章完了下mongodb的安裝和IDE工具，現在開始使用java進行連接。 第一步：使用jar包， 這裏需要三個包，具體為啥我也不清楚，反正因為報錯，我就按照官方文檔一個個的都下載了。 鏈接：htt

AD 域服務簡介（二）- Java 獲取 AD 域用戶

set env red smo 關系數據庫 sign exception ini 性能 博客地址：http://www.moonxy.com 一、前言 先簡單簡單回顧上一篇博文中關於 AD 域和 LDAP目錄訪問協議的基本概念。 AD（Active Directory）活

Java開發筆記（二）Java工程的帝國區劃

頂部 圖片 最有 屬於 聚集地 合格 ext info 主界面 上一篇文章介紹了如何運行了第一個Java程序“Hello World”。然而這個開發環境看起來那麽陌生，一個個名字符號完全不知道它們是幹啥的呀，對於初學者來說，好比天書一般，多看幾眼感覺都要走火入魔了。因此接下

（二）Java集合專題-詳細分析HashtableJDK1.8集合底層實現的思想

（一）詳細分析Hashtable1.8集合底層實現的思路和原始碼 （1）先來看一下Hashtable的結構圖：首先他和HashMap結構都是一樣的，都是由陣列和連結串列進行實現的（在JDK1.8是用陣列和連結串列和紅黑樹的），每一個數組裡面存的是一個Entry節點，節點裡面有Key、

springboot上傳下載檔案（3）--java api 操作HDFS叢集+叢集配置

 只有光頭才能變強! 前一篇文章講了nginx+ftp搭建獨立的檔案伺服器 但這個伺服器宕機了怎麼辦？ 我們用hdfs分散式檔案系統來解決這個問題（同時也為hadoop系列開個頭） 目錄 1、Ubuntu14.04下配置Hadoop(2.8.5)叢集環境詳解(完全分

Java編譯（二） Java前端編譯：Java原始碼編譯成Class檔案的過程

Java編譯（二）Java前端編譯： Java原始碼編譯成Class檔案的過程               在上篇文章《Java三種編譯方式：前端編

從Paxos到ZooKeeper（二）初識ZooKeeper

ZooKeeper是什麼 ZooKeeper是一個典型的分散式資料一致性解決方案 ZooKeeper特性 （一）順序一致性 從一個客戶端發起的請求會嚴格按照發起請求的順序應用到Zookeeper中 （二）原子性 所有事務請求處理的結果在叢集中各個機器上的應用情況是一樣的

Java EE入門教程系列第一章Java EE的概述（二）——Java EE技術框架和開發工具

1.3Java EE的技術框架 從技術的角度劃分，完整的Java EE分成了4個部分：元件技術、服務技術、通訊技術和架構技術。 下面給出的是一個適合初學者的體系結構簡化圖，暫時接觸不到的部分統一用“支援技術”表示，我們暫時只專注於與應用級開發相關的技術即可。 1.元件技術 這是

基於Curator操作ZooKeeper（二）-Watcher操作-補充TreeCache

轉自：https://blog.csdn.net/Leafage_M/article/details/78735485#treecache Java原生API操作ZooKeeper可參看： Java原生API操作Zookeeper（一） Java原生API操作Zookeeper（二）

基於Curator操作ZooKeeper（二）-Watcher操作

Java原生API操作ZooKeeper可參看： Java原生API操作Zookeeper（一） Java原生API操作Zookeeper（二） 相關內容： 基於Curator操作ZooKeeper（一）-基本操作 基於Curator操作ZooKeeper（二）-Watche