Zookeeper的操作

1.用戶端腳本

  進入Zookeeper的bin目錄之後，直接執行下面的指令：

  出現下面的輸出資訊，就表示已經成功連接配接上本地的zookeeper伺服器了。

  預設連接配接本地的zookeeper伺服器，要是想連接配接指定的zookeeper伺服器，通過下面的指令實作。

zkCli.sh -server ip:port

(1)建立節點

  使用create指令建立節點，用法如下：

create [-s] [-e] path data acl

  其中，-s或-e分别指定節點特性：順序或臨時節點，預設情況下建立的是持久節點。path為其在zookeeper根節點下的路徑，data該節點資料内容，acl為其權限。

例：在zookeeper根節點下建立一個叫作/zk-book，節點資料内容為“123”的節點。

(2)讀取

  使用ls指令列出指定節點下的所有子結點，與linux的ls指令是類似的。用法如下：

ls path [watch]

  使用get指令擷取指定節點的資料内容，用法如下：

get path [watch]

  使用stat指令檢視節點的屬性資訊，用法如下：

stat path [watch]

(3)更新

  使用set指令更新指定節點的資料内容，用法如下：

set path data [version]

  可以看到這裡節點/zk-book的資料内容更新為了"456"，dataVersion也變成了1.

(4)删除

  使用delete指令删除指定節點，用法如下：

delete path [version]

  對于包含子節點的節點，該指令無法成功删除

(5)退出

  使用quit指令即可退出

2.Java用戶端API使用

導入jar包

  在使用Java操作zookeeper之前，需要在Idea中導入zookeeper的jar包。

  首先打開Idea，依次點選File - Project Structure…

  點選Modules - Dependencies - JARs or Directories… - +号

  找到自己安裝zookeeper的路徑下的lib包，導入即可。

(1)建立會話

  用戶端可以通過建立一個zookeeper執行個體來連接配接zookeeper伺服器。zookeeper的4種構造方法如下：

ZooKeeper(String connectString, int sessionTimeout, Watcher watcher);

#connectString：值zookeeper伺服器清單，形如host:port

#sessionTimeout：會話的逾時時間

#watcher：Watcher事件通知處理器

ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly);

#canBeReadOnly：辨別目前會話是否支援隻讀模式

ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, long sessionId, byte[] sessionPasswd);

#sessionId：會話ID

#sessionPasswd：會話密鑰

ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, long sessionId, byte[] sessionPasswd, boolean canBeReadOnly);

  下面用Java來建立一個基本的zookeeper會話執行個體。

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import java.util.concurrent.CountDownLatch;

public class Simple_session implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    
    public static void main(String[] args) throws Exception{
    	//使用上述第一種構造方法執行個體化一個ZooKeeper對象
        ZooKeeper zooKeeper = new ZooKeeper("192.168.1.12:2181", 5000, new Simple_session());
        System.out.println(zooKeeper.getState());
        try{
            connectedSemaphore.await();
        }catch(InterruptedException e){}
        System.out.println("ZooKeeper session established.");
    }
    
    //重寫Watcher接口中的process()方法
    //該方法負責處理來自ZooKeeper服務端的Watcher通知，
    //收到服務端發來的SyncConnected事件之後，解除主程式在CountDownLatch上的等待阻塞
    public void process(WatchedEvent event){
        System.out.println("Receive watched event:" + event);
        if(Event.KeeperState.SyncConnected == event.getState()){
            connectedSemaphore.countDown();
        }
    }
}
           

  成功建立了一個簡單的會話，程式的輸出結果如下：

(2)建立節點

  用戶端可以通過ZooKeeper的API來建立一個資料節點，有下面兩個接口：

同步方式建立節點

String create(final String path, byte data[ ], List< ACL> acl, CreateMode createMode);

#path:需要建立的資料節點的結點路徑

#data[ ]:節點建立後的初始内容

#acl:權限

#createMode:節點類型

異步方式建立節點

void create(final String path, byte data[ ], List< ACL> acl, CreateMode createMode, StringCallback cb, Object ctx);

#cb:異步回調函數

#ctx:用于傳遞一個對象，可以在回調方法執行的時候使用，通常放一個上下文資訊

  上面兩個接口中，有一個CreateMode，其是一個枚舉類型。

enum createMode{
	PERSISTENT,				//持久
	PERSISTENT_SEQUENTIAL,	//持久順序
	EPHEMERAL,				//臨時
	EPHEMERAL_SEQUENTIAL;	//臨時順序
}
           

  下面使用同步API來建立一個節點。

import org.apache.zookeeper.*;
import java.util.concurrent.CountDownLatch;

public class Create_node implements Watcher {

    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);

    public static void main(String[] args) throws Exception{
        ZooKeeper zooKeeper = new ZooKeeper("192.168.1.12:2181", 5000, new Create_node());
        connectedSemaphore.await();
        
        //這裡使用的是同步的節點建立接口
        //臨時節點
        String path1 = zooKeeper.create("/zk-test-ephemeral-", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("Success create znode: " + path1);
		
		//臨時順序節點
        String path2 = zooKeeper.create("/zk-test-ephemeral-", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println("Success create znode: " + path2);
    }

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

(3)删除節點

  用戶端可以通過ZooKeeper的API來删除一個節點，有下面兩個接口：

同步删除

public void delete(final String path, int version);

異步删除

public void delete(final String path, int version, VoidCallback cb, Object ctx);

(4)擷取資料及更新資料

  讀取資料包括了子節點清單的擷取和節點資料的擷取，ZooKeeper提供了不同API來擷取資料。

1.擷取子結點

  使用getChildren來擷取所有子結點。具體的有8個接口，這裡隻列一個最常用的。

List< String > getChildren(final String path, Watcher watcher)

  接口中的參數有一個很重要的東西是，注冊Watcher。如果ZooKeeper用戶端在擷取到指定節點的子節點清單後，還需要訂閱這個子節點清單的變化通知，那麼就可以通過注冊一個Watcher來實作。當有子節點被添加或者删除時，服務端就會向用戶端發送一個NodeChildrenChanged類型的事件通知。在服務端發送給用戶端的事件通知中，不包含最新的節點清單，用戶端必須主動重新進行擷取。

  使用同步API擷取子節點清單。

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

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        String path = "/zk-book";
        zk = new ZooKeeper("192.168.1.8:2181", 5000, new GetChildren_test());
        connectedSemaphore.await();
        zk.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zk.create(path + "/c1", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

        List<String> childrenList = zk.getChildren(path, true);
        System.out.println(childrenList);

        zk.create(path + "/c2", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        Thread.sleep(Integer.MAX_VALUE);
    }

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

2.擷取節點的資料内容

  使用getData來擷取節點的資料内容。

byte[] getData(final String path, Watcher watcher, Stat stat)

  這裡的Stat參數，表示指定資料節點的結點狀态資訊。

  既然可以擷取節點的資料内容，那麼肯定也可以修改節點的資料内容。更新資料使用setData。

Stat setData(final String path, byte data[], int version)

  這裡的version參數，用于指定節點的資料版本，表面本次跟新操作是針對指定的資料版本進行的。

public class GetData_test 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("192.168.1.8:2181", 5000, new GetData_test());
        connectedSemaphore.await();
        zk.create(path, "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println(new String(zk.getData(path, true, stat)));
        System.out.println(stat.getCzxid() + "," + stat.getMzxid() + "" + stat.getVersion());
        zk.setData(path, "456".getBytes(), -1);
        Thread.sleep(10000);
    }

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

  這裡需要注意一下，因為我們之前的擷取子節點操作已經建立了/zk-book，在這裡進行操作時候會再次建立/zk-book，會報錯。

  這裡隻需要将之前建立的/zk-book節點删除再運作該程式即可，删除的話用上面第一種用戶端操作的方法區删除會比較簡單。