java連接zookeeper的實現(xiàn)示例

更新時間：2023年11月01日 11:44:09 作者：曹朋羽

ZooKeeper官方提供了Java API,可以通過Java代碼來連接zookeeper服務進行操作,本文就來介紹一下java連接zookeeper的實現(xiàn)示例,具有一定的參考價值,感興趣的可以了解一下

API

ZooKeeper官方提供了Java API，可以通過Java代碼來連接zookeeper服務進行操作?？梢赃B接、創(chuàng)建節(jié)點、獲取節(jié)點數(shù)據(jù)、監(jiān)聽節(jié)點變化等操作，具體有以下幾個重要的類：

ZooKeeper：ZooKeeper類是Java API的核心類，用于與ZooKeeper服務器建立連接，并提供了一系列方法來操作ZooKeeper的節(jié)點。
Watcher：Watcher是ZooKeeper的一個回調(diào)接口，當節(jié)點發(fā)生變化時會調(diào)用相應的方法進行通知。
CreateMode：CreateMode枚舉類定義了節(jié)點的類型，包括永久節(jié)點、臨時節(jié)點、順序節(jié)點和臨時順序節(jié)點。
Stat：Stat類表示節(jié)點的元數(shù)據(jù)信息，比如修改版本、數(shù)據(jù)長度、子節(jié)點數(shù)量等。

添加依賴

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

操作例子

String host = "localhost:2181";
//建立連接
zooKeeper = new ZooKeeper(host, 2000, null);
String path = "/test";
Watcher watcher = new Watcher() {
  @Override
  public void process(WatchedEvent watchedEvent) {
    System.out.println("Node changed: " + watchedEvent.getPath());
    System.out.println(watchedEvent);
  }
};
//獲取節(jié)點狀態(tài) 如果不存在返回null
Stat stat = zooKeeper.exists(path, false);
if(null != stat){
	System.out.println(stat.getCzxid()+"-"+stat.getAversion());
}

//創(chuàng)建節(jié)點 包含版本、時間、數(shù)據(jù)長度等信息
zooKeeper.create(path,"123".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
//添加watcher
zooKeeper.addWatch(path, watcher, AddWatchMode.PERSISTENT);

//獲取節(jié)點數(shù)據(jù)
byte[] data = zooKeeper.getData(path, false, null);
System.out.println(new String(data));

zooKeeper.create(path+"/1","child1".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
//獲取子節(jié)點
List<String> children = zooKeeper.getChildren(path, false);
System.out.println("childs size:"+children.size());
//刪除子節(jié)點
zooKeeper.delete(path+"/1",-1);
zooKeeper.close();

zkClient

zkClient封裝了zookeeper的官方api，簡化了一些繁瑣的操作，并提供了一些額外的功能，提高了開發(fā)效.

添加依賴

<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.11</version>
</dependency>

zkclient對節(jié)點數(shù)據(jù)的操作進行了序列化，這里先準備一個string類型的序列化類。需要實現(xiàn)ZkSerializer接口

public class ZkStringSerializer implements ZkSerializer {
    @Override
    public byte[] serialize(Object o) throws ZkMarshallingError {
        return String.valueOf(o).getBytes();
    }

    @Override
    public Object deserialize(byte[] bytes) throws ZkMarshallingError {
        return new String(bytes);
    }
}

基本操作

ZkClient zkClient = new ZkClient("localhost:2181");
//自定義序列化 否則報錯
zkClient.setZkSerializer(new ZkStringSerializer());
String path = "/test";
//判斷節(jié)點是否存在
boolean exist = zkClient.exists(path);
System.out.println(exist);
if(!exist){//創(chuàng)建節(jié)點
    zkClient.create(path,"123", CreateMode.PERSISTENT);
}
//讀取節(jié)點數(shù)據(jù)
System.out.println((String) zkClient.readData(path));
zkClient.writeData(path,"456");//設置節(jié)點數(shù)據(jù)
System.out.println((String) zkClient.readData(path));
zkClient.delete(path);//刪除節(jié)點

zkClient.close();

節(jié)點變化事件

String path = "/test";
/**
 * 節(jié)點變化事件
 * 只監(jiān)聽節(jié)點增減，不監(jiān)聽數(shù)據(jù)變化事件
 */
zkClient.subscribeChildChanges(path, new IZkChildListener() {
    @Override
    public void handleChildChange(String parentPath, List<String> children) throws Exception {
        System.out.println("節(jié)點"+parentPath+"發(fā)生變化");
        System.out.println(children);
    }
});
//節(jié)點操作，觀察handleChildChange接收到對應事件
Thread.sleep(2000);
zkClient.createPersistent(path);
Thread.sleep(2000);
zkClient.createPersistent(path+"/child1");
Thread.sleep(2000);
zkClient.writeData(path+"/child1","123");
Thread.sleep(2000);
zkClient.delete(path+"/child1");
Thread.sleep(2000);
zkClient.delete(path);
Thread.sleep(100000);

節(jié)點數(shù)據(jù)變化事件

    String path = "/test";
    /**
     * 節(jié)點變化事件,只檢測當前節(jié)點，感知不到其子節(jié)點
     * 節(jié)點被刪除或節(jié)點數(shù)據(jù)變化
     */
    zkClient.subscribeDataChanges(path, new IZkDataListener() {
        @Override
        public void handleDataChange(String s, Object o) throws Exception {
            System.out.println("節(jié)點："+s+"數(shù)據(jù)變?yōu)?"+o);
        }

        @Override
        public void handleDataDeleted(String s) throws Exception {
            System.out.println("節(jié)點："+s+"刪除");
        }
    });

    Thread.sleep(2000);
    zkClient.createPersistent(path);
    Thread.sleep(2000);
    zkClient.createPersistent(path+"/child1");
    Thread.sleep(2000);
    zkClient.delete(path+"/child1");
    Thread.sleep(2000);
    zkClient.writeData(path,"123");
    Thread.sleep(2000);
    zkClient.delete(path);
    Thread.sleep(100000);
}

Curator

curator是另一個java連接zookeeper類庫。功能更加強大。提供了連接重試、分布式鎖、選舉、隊列等多種實際場景的用例。這里先簡單搞個使用例子。

添加依賴

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>5.1.0</version>
</dependency>

curator-framework是基礎的依賴，一些特定的使用方式需要添加不同的依賴，有curator-recipes、curator-x-discovery、curator-x-async等。

基本操作

//創(chuàng)建連接
CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181", new ExponentialBackoffRetry(1000, 3));
client.start();
String path = "/test";
client.checkExists().forPath(path);//判斷是否存在
client.create().forPath(path, "123".getBytes());//創(chuàng)建節(jié)點
byte[] data = client.getData().forPath(path);//獲取數(shù)據(jù)
System.out.println(new String(data));
client.setData().forPath(path, "456".getBytes());//設置數(shù)據(jù)
client.delete().forPath(path);//刪除節(jié)點

client.close();

節(jié)點監(jiān)聽

CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181", new ExponentialBackoffRetry(1000, 3));
client.start();
String path = "/test";
NodeCache  nodeCache = new NodeCache(client,path);
//添加監(jiān)聽
nodeCache.getListenable().addListener(new NodeCacheListener() {
    @Override
    public void nodeChanged() throws Exception {
        ChildData data = nodeCache.getCurrentData();
        if (data != null) {
            System.out.println("Node changed: " + data.getPath() + ", value: " + new String(data.getData()));
        } else {
            System.out.println("Node deleted: " + nodeCache.getPath());
        }
    }
});
nodeCache.start();
client.create().forPath(path);
client.setData().forPath(path, "123".getBytes());
client.delete().forPath(path);
client.close();

這里NodeCache被標識@Deprecated，也不知道被什么方式代替了，后面再研究。先簡單使用。

到此這篇關(guān)于java連接zookeeper的實現(xiàn)示例的文章就介紹到這了,更多相關(guān)java連接zookeeper內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: