2. 程式人生 > >Zookeeper使用--Java API(轉)

Zookeeper使用--Java API(轉)

阿新 • • 發佈：2017-09-29
包括 acl logs 新的 ack max false pen strong

[轉載自作者leesf，原文鏈接]

操作示例

　　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

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

　　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 子節點獲取

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

　　① 同步方式　　

技術分享 技術分享 
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]
技術分享

　　結果表示通過異步的方式可以獲取子節點信息。

　　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

　　結果表明采用異步方式同樣可方便獲取節點的數據。

　　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，表明成功更新節點數據。

　　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。

　　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

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

Zookeeper使用--Java API(轉)

相關推薦

Zookeeper使用--Java API(轉)

包括 acl logs 新的 ack max false pen strong [轉載自作者leesf，原文鏈接] 操作示例 　　1 創建節點 　　創建節點有異步和同步兩種方式。無論是異步或者同步，Zookeeper都不支持遞歸調用，即無法在父節點不存在的情況下

ZooKeeper Java API

state exce 處理 art throws 枚舉類 event exception 類型 org.apache.zookeeper.ZookeeperZookeeper 是在 Java 中客戶端主類，負責建立與 zookeeper 集群的會話， 並提供方法進行操作。o

zookeeper java api（2）

    這裡介紹其他的API對zookeeper的操作。 同步方式獲取子節點資料 public static void getChildrenSync() throws KeeperException, InterruptedException {

zookeeper java api（1）

1 Zookeeper安裝以及啟動     這裡我已經進行了安裝，並且啟動了Zookeeper。埠是2182 2 Zookeeper config tickTime=2000 initLimit=10 syncLimit=5 dataDir=D://zook

Zookeeper --- Java API簡單例項

1、簡單使用 import java.util.List; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeep

使用ZooKeeper Java API程式設計

https://www.cnblogs.com/IcanFixIt/p/7882107.htmlhttps://blog.csdn.net/qiushisoftware/article/details/79043379https://blog.csdn.net/wo54107

Zookeeper Java API (二) zk節點內容的操作和節點的刪除

獲取節點內容 獲取節點內容的方法定義 public byte[] getData(String path, boolean watch, Stat stat) 其中: - path:要獲取內容的節點路徑 - watch:是否設定監控,也就是是否需要

Zookeeper Java API (三) zk節點的通知(Watcher)和回撥(CallBack)

Zk中的通知(Watcher) ZooKeeper中實現對接點的監控,需要實現Watcher介面類,實現其中的process方法 public class WatcherDemo implements Watcher{ public void p

Zookeeper Java API呼叫

public class ZooKeeperTest implements Watcher{ //public final static String zkServerPath = "12.45.67.80:2181"; //叢集，使用逗號分隔 public final

【Kafka】使用非自帶zookeeper Java API 例子

這裡沒有使用kafka自帶的zk。1.啟動zk：zk下載解壓至任意資料夾。新建kafka-zk-csdn資料夾。這裡啟動包含三個節點的zk偽叢集，進入kafka-zk-csdn資料夾，新建zk1，zk2和zk3資料夾。kafka-zk-csdn:--zk1    --data

使用ZooKeeper提供的Java API操作ZooKeeper

zookeeper 服務協調框架 分布式 集群 Java API 建立客戶端與zk服務端的連接 我們先來創建一個普通的maven工程，然後在pom.xml文件中配置zookeeper依賴： <dependencies> <dependency>

使用Java API操作zookeeper的acl權限

zookeeper Java API 分布式 服務協調框架 ACL權限 默認匿名權限 ZooKeeper提供了如下幾種驗證模式（scheme）： digest：Client端由用戶名和密碼驗證，譬如user:password，digest的密碼生成方式是Sha1摘要的base64形式 a

zookeeper 基本命令和java api

1.Zookeeper命令工具 啟動zookeeper服務後，連線到zookeeper服務： zkServer -server localhost:2181 執行結果如下： Connecting to localhost:2181 2018-06-10 03:5

第6章 使用ZooKeeper原生Java API進行客戶端開發

使用ZooKeeper原生Java API進行客戶端開發 6-1 建立客戶端與zk服務端的連線 6-2 zk會話重連機制 6-3 同步非同步建立zk節點 6-4 修改zk節點資料 6-5 同步非同步刪除zk節點 6-6 CountDownL

Java API 操作 zookeeper

1.需要的jar包: 1.1 zookeeper的核心包:zookeeper-3.4.5.jar 1.2 log4j的jar包:log4j-1.2.15.jar 1.3 netty的jar包:netty-3.2.2.Final.jar 1.4 slf4j的jar

Zookeeper使用（二）--Java API

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

jdk8環境下，用eclipse indigo進行Zookeeper叢集java api客戶端連線，因版本報錯

Unresolved compilation problem: The type java.lang.CharSequence cannot be resolved. It is indirectly referenced from required .class files S

zookeeper框架——Java API開發

一、基本運用          org.apache.zookeeper.Zookeeper是客戶端入口主類，負責建立與server的會話。它提供了表 1 所示幾類主要方法 ： 功能 描述 create

18 大資料zookeeper --使用java api操作zookeeper

ZooKeeper服務命令: 在準備好相應的配置之後，可以直接通過zkServer.sh 這個指令碼進行服務的相關操作 1. 啟動ZK服務: sh bin/zkServer.sh start 2. 檢視ZK服務狀態: sh bin/zkServer.sh status 3. 停止

使用Java API、Curator操作zookeeper的acl許可權

zk原生api操作acl許可權 預設匿名許可權 ZooKeeper提供瞭如下幾種驗證模式（scheme）： digest：Client端由使用者名稱和密碼驗證，譬如user:password，digest的密碼生成方式是Sha1摘要的base64形式 auth：不使用任何id