Java調(diào)用Zookeeper的實(shí)現(xiàn)步驟

更新時(shí)間：2021年08月19日 08:39:30 作者：blue星空

本文主要介紹了Java調(diào)用Zookeeper的實(shí)現(xiàn)步驟，文中通過示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

watch機(jī)制

Zookeeper watch是一種監(jiān)聽通知機(jī)制，可以隨時(shí)監(jiān)聽一些數(shù)據(jù)的變化，從而實(shí)現(xiàn)數(shù)據(jù)的及時(shí)性。

Zookeeper所有的讀操作getData(), getChildren()和 exists()都可以設(shè)置監(jiān)聽(watch)?！緦懖僮鲃t是不能設(shè)置監(jiān)視點(diǎn)的?！?br />

Watch的三個(gè)關(guān)鍵點(diǎn)：

一次有效：當(dāng)設(shè)置監(jiān)視的數(shù)據(jù)發(fā)生改變時(shí)，該監(jiān)視事件會(huì)被發(fā)送到客戶端，并且該監(jiān)聽將會(huì)停止，除非重啟注冊(cè)監(jiān)聽；
順序保證：網(wǎng)絡(luò)延遲或者其他因素可能導(dǎo)致不同的客戶端在不同的時(shí)刻感知某一監(jiān)視事件，但是不同的客戶端所看到的一切具有一致的順序；
可以監(jiān)聽數(shù)據(jù)和子節(jié)點(diǎn)：getData()和 exists()可以設(shè)置監(jiān)聽數(shù)據(jù)變化；getChildren 可以設(shè)置監(jiān)聽子節(jié)點(diǎn)變化；

常用API

/**
*　構(gòu)造器
* @param connectString 集群的IP：端口號(hào)；多個(gè)服務(wù)器時(shí)，中間用逗號(hào)分割
* @param sessionTimeout 超時(shí)時(shí)間，單位：毫秒
* @param watcher  監(jiān)聽器，監(jiān)聽節(jié)點(diǎn)變化
* @throws IOException
*/
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException

/**
*
* @param path 節(jié)點(diǎn)路徑
* @param data 數(shù)據(jù)
* @param acl 訪問控制列表
* @param createMode 節(jié)點(diǎn)類型
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public String create(String path, byte[] data, List<ACL> acl, CreateMode createMode) throws KeeperException, InterruptedException

/**
*
* @param path 節(jié)點(diǎn)路徑
* @param watch 監(jiān)聽器
* @return 所有的子節(jié)點(diǎn)的名稱
* @throws KeeperException
* @throws InterruptedException
*/
public List<String> getChildren(String path, boolean watch) throws KeeperException, InterruptedException

/**
*
* @param path 節(jié)點(diǎn)路徑
* @param watcher 監(jiān)聽器
* @param stat  狀態(tài)信息【可以為null】
* @return 節(jié)點(diǎn)數(shù)據(jù)的二進(jìn)制數(shù)組【可以通過new String()轉(zhuǎn)換成字符串信息】
* @throws KeeperException
* @throws InterruptedException
*/
public byte[] getData(String path, Watcher watcher, Stat stat) throws KeeperException, InterruptedException

/**
*
* @param path 節(jié)點(diǎn)路徑
* @param watch 監(jiān)聽器
* @param cb 回調(diào)函數(shù)
* @param ctx  上下文參數(shù) ？【該參數(shù)不太理解，望知道的留言講解，謝謝】
*/
public void getData(String path, boolean watch, AsyncCallback.DataCallback cb, Object ctx)

/**
*
* @param path 節(jié)點(diǎn)路徑
* @param data 數(shù)據(jù)
* @param version 版本號(hào)【初始通常賦值為-1，每次更新會(huì)自動(dòng)+1】
* @return  狀態(tài)信息
* @throws KeeperException
* @throws InterruptedException
*/
public Stat setData(String path, byte[] data, int version) throws KeeperException, InterruptedException

/**
*如果Stat為null，則節(jié)點(diǎn)不存在
* @param path 節(jié)點(diǎn)路徑
* @param watch 監(jiān)聽器
* @return 狀態(tài)信息
* @throws KeeperException
* @throws InterruptedException
*/
public Stat exists(String path, boolean watch) throws KeeperException, InterruptedException

/**
* 如果要?jiǎng)h除的節(jié)點(diǎn)有子節(jié)點(diǎn)，會(huì)報(bào)錯(cuò)：KeeperException$NotEmptyException: KeeperErrorCode = Directory not empty for
* 如果節(jié)點(diǎn)不存在，會(huì)報(bào)錯(cuò)：KeeperException$NoNodeException: KeeperErrorCode = NoNode for
* @param path 節(jié)點(diǎn)路徑
* @param version 版本號(hào)[version = -1 : 匹配所有的版本]
* @throws InterruptedException
* @throws KeeperException
*/
public void delete(String path, int version) throws InterruptedException, KeeperException

JAVA調(diào)用

初始化

try {
  ZooKeeper　zooKeeper = new ZooKeeper("172.23.34.13:2181", 15000, event -> {
        if (event.getType() == Watcher.Event.EventType.None && event.getState() == Watcher.Event.KeeperState.SyncConnected) {
            System.out.println("Connectted successful.");
        }
    });
} catch (IOException e) {
    e.printStackTrace();
}

創(chuàng)建節(jié)點(diǎn): create

@Test
public void create() throws KeeperException, InterruptedException {
    //參數(shù)：1，節(jié)點(diǎn)路徑； 2，要存儲(chǔ)的數(shù)據(jù)； 3，節(jié)點(diǎn)的權(quán)限； 4，節(jié)點(diǎn)的類型
    String nodePath = zooKeeper.create("/java/2183", "This is Java Node 2183.".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    System.out.println(nodePath);
}

獲取子節(jié)點(diǎn)： ls

public void getChildren() throws KeeperException, InterruptedException {
    List<String> children = zooKeeper.getChildren("/", true);
    for (String child : children) {
        System.out.println("child: "+child);
    }
}

同步獲取節(jié)點(diǎn)內(nèi)容： get

@Test
public void getData() throws KeeperException, InterruptedException {
    String path = "/java";
    byte[] bytes = zooKeeper.getData(path, event -> {
        if (event.getType() == Watcher.Event.EventType.NodeDataChanged && path.equals(event.getPath())) {
            System.out.println("Date changed.");
        }
    }, null);
    System.out.printf("The data of %s is : %s \n",path, new String(bytes));
}

異步獲取節(jié)點(diǎn)內(nèi)容： get

@Test
public void getDataAsync() {
    String path = "/java";
    zooKeeper.getData(path, false, new AsyncCallback.DataCallback() {
        @Override
        public void processResult(int i, String s, Object o, byte[] bytes, Stat stat) {
            System.out.printf("The data of %s is : %s \n",path, new String(bytes));
        }
    },"1000");

    //休眠20秒，查看響應(yīng)結(jié)果
    try {
        Thread.sleep(20000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

指定版本號(hào)更新數(shù)據(jù)：set

@Test
public void setData() throws KeeperException, InterruptedException {
    Stat stat = zooKeeper.setData("/java", "This is from java.".getBytes(), -1);
    //更新節(jié)點(diǎn)后，version會(huì)自動(dòng)+1。故，返回值為0
    System.out.println(stat.getAversion());
}

多線程下更新數(shù)據(jù)：set

@Test
public void setDataThread() throws KeeperException, InterruptedException {
    String path = "/java";
    Stat stat = new Stat();
    //1,先獲取節(jié)點(diǎn)的當(dāng)前版本
    zooKeeper.getData(path,false,stat);
    //2，在當(dāng)前版本的基礎(chǔ)上修改節(jié)點(diǎn)內(nèi)容
    zooKeeper.setData(path, "This is from java.".getBytes(), stat.getVersion());
}

判斷節(jié)點(diǎn)是否存在

@Test
public void exists() throws KeeperException, InterruptedException {
    Stat stat = zooKeeper.exists("/java", false);
    if (stat == null) {
        System.out.println("Not Exists.");
    }else {
        System.out.println("Exists.");
    }
}

刪除節(jié)點(diǎn)

@Test
public void delete() throws KeeperException, InterruptedException {
    //version = -1 : 匹配所有的版本
    zooKeeper.delete("/java/2182", -1);
}

到此這篇關(guān)于Java調(diào)用Zookeeper的實(shí)現(xiàn)步驟的文章就介紹到這了,更多相關(guān)Java調(diào)用Zookeeper內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: