【ZooKeeper系列】2.用Java實現ZooKeeper API的調用

溫馨提示：在這里我再次提個小要求，希望大家能習慣看官方文檔，文檔雖然是英文但用詞都比較簡單，基本都能看懂文檔表達的意思。授之以魚不如授之以漁的道理相信大家都明白，也希望通過猿人谷的這個ZooKeeper系列，讓大家入門、到熟悉，舉一反三后能精通ZooKeeper。

在前一篇我們介紹了ZooKeeper單機版、偽集群和集群環境搭建，通過命令行的方式做了節點的創建、刪除、更新、獲取節點信息的測試。Zookeeper 的目的是為客戶端構建復雜的協調功能提供簡單、高效的核心 API，這一篇我們用Java通過ZooKeeper提供的API接口來實現這些增刪改查的功能。

1 簡介

org.apache.zookeeper.Zookeeper是ZooKeeper客戶端的主類，在官方文檔（該系列文章以v3.5.5為主，v3.6.6的API Docs還沒有）中已明確說明（This is the main class of ZooKeeper client library.）。

This is the main class of ZooKeeper client library. To use a ZooKeeper service, an application must first instantiate an object of ZooKeeper class. All the iterations will be done by calling the methods of ZooKeeper class. The methods of this class are thread-safe unless otherwise noted.
Once a connection to a server is established, a session ID is assigned to the client. The client will send heart beats to the server periodically to keep the session valid.

創建一個ZooKeeper的實例來使用org.apache.zookeeper.Zookeeper里的方法，官方文檔已經指出沒有特別聲明的話，ZooKeeper類里的方法是線程安全的。客戶端連接到ZooKeeper服務的時候，會給客戶端分配一個會話ID（session ID），客戶端與服務端會通過心跳來保持會話有效。

org.apache.zookeeper.Zookeeper里的方法非常多，就不一一列舉了，只列幾個增刪改查的。

2 測試環境搭建

這里新建一個Spring Boot的項目來進行測試，新建Spring Boot項目的過程很簡單，也不是這里的重點，就不做介紹了。

項目里會需要額外引入兩個包來進行測試：

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

        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter-api</artifactId>
            <version>5.5.2</version>
        </dependency>

3 API測試

完整測試代碼如下：

/**
 *  簡單測試示例
 * @author 猿人谷
 * @date 2019/12/16
 */
public class ZooKeeperDemo {

    private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperDemo.class);

    private static final int SESSION_TIME_OUT = 10000;
    // ZooKeeper服務的地址，如為集群，多個地址用逗號分隔
    private static final String CONNECT_STRING = "127.0.0.1:2181";
    private static final String ZNODE_PATH = "/zk_demo";
    private static final String ZNODE_PATH_PARENT = "/app1";
    private static final String ZNODE_PATH_CHILDREN = "/app1/app1_1";

    private ZooKeeper zk = null;

    @Before
    public void init() throws IOException {
        zk = new ZooKeeper(CONNECT_STRING, SESSION_TIME_OUT, new Watcher(){
            @Override
            public void process(WatchedEvent event) {
                System.out.println("已經觸發了" + event.getType() + "事件！");
            }
        });

    }

    @Test
    public void testCreate() throws KeeperException, InterruptedException {
        zk.create(ZNODE_PATH, "anna2019".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    @Test
    public void testCreateParentZnode() throws KeeperException, InterruptedException {
        zk.create(ZNODE_PATH_PARENT, "anna2019".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    @Test
    public void testCreateChildrenZnode() throws KeeperException, InterruptedException {
        zk.create(ZNODE_PATH_CHILDREN, "anna2020".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    @Test
    public void testGet() throws KeeperException, InterruptedException {
        byte[] data1 = zk.getData(ZNODE_PATH, false, null);
        byte[] data2 = zk.getData(ZNODE_PATH_PARENT, false, null);
        byte[] data3 = zk.getData(ZNODE_PATH_CHILDREN, false, null);
        LOGGER.info("{}的信息：{}", ZNODE_PATH, new String(data1) );
        LOGGER.info("{}的信息：{}", ZNODE_PATH_PARENT, new String(data2) );
        LOGGER.info("{}的信息：{}", ZNODE_PATH_CHILDREN, new String(data3) );
    }

    /**
     *  刪除
     * @throws KeeperException
     * @throws InterruptedException
     */
    @Test
    public void testDelete() throws KeeperException, InterruptedException {
        // 指定要刪除的版本，-1表示刪除所有版本
        zk.delete(ZNODE_PATH, -1);
    }

    /**
     *  刪除含有子節點
     * @throws KeeperException
     * @throws InterruptedException
     */
    @Test
    public void testDeleteHasChildrenZnode() throws KeeperException, InterruptedException {
        // 指定要刪除的版本，-1表示刪除所有版本
        zk.delete(ZNODE_PATH_PARENT, -1);
    }

    @Test
    public void testSet() throws KeeperException, InterruptedException {
        Stat stat = zk.setData(ZNODE_PATH, "yuanrengu".getBytes(), -1);
        LOGGER.info(stat.toString());
    }

}

上面有用到@Before，簡單說明下：

@BeforeClass – 表示在類中的任意public static void方法執行之前執行
@AfterClass – 表示在類中的任意public static void方法執行之后執行
@Before – 表示在任意使用@Test注解標注的public void方法執行之前執行
@After – 表示在任意使用@Test注解標注的public void方法執行之后執行
@Test – 使用該注解標注的public void方法會表示為一個測試方法

如果將SESSION_TIME_OUT設置的時間太短，會報API客戶端異常：org.apache.zookeeper.KeeperException$ConnectionLossException: KeeperErrorCode = ConnectionLoss for /zk_demo 。完整的報錯信息如下：

09:33:52.139 [main-SendThread(106.12.111.172:2181)] DEBUG org.apache.zookeeper.ClientCnxnSocketNIO - Ignoring exception during shutdown input
java.net.SocketException: Socket is not connected
	at sun.nio.ch.Net.translateToSocketException(Net.java:123)
	at sun.nio.ch.Net.translateException(Net.java:157)
	at sun.nio.ch.Net.translateException(Net.java:163)
	at sun.nio.ch.SocketAdaptor.shutdownInput(SocketAdaptor.java:401)
	at org.apache.zookeeper.ClientCnxnSocketNIO.cleanup(ClientCnxnSocketNIO.java:198)
	at org.apache.zookeeper.ClientCnxn$SendThread.cleanup(ClientCnxn.java:1338)
	at org.apache.zookeeper.ClientCnxn$SendThread.cleanAndNotifyState(ClientCnxn.java:1276)
	at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1254)
Caused by: java.nio.channels.NotYetConnectedException: null
	at sun.nio.ch.SocketChannelImpl.shutdownInput(SocketChannelImpl.java:782)
	at sun.nio.ch.SocketAdaptor.shutdownInput(SocketAdaptor.java:399)
	... 4 common frames omitted
09:33:52.140 [main-SendThread(106.12.111.172:2181)] DEBUG org.apache.zookeeper.ClientCnxnSocketNIO - Ignoring exception during shutdown output
java.net.SocketException: Socket is not connected
	at sun.nio.ch.Net.translateToSocketException(Net.java:123)
	at sun.nio.ch.Net.translateException(Net.java:157)
	at sun.nio.ch.Net.translateException(Net.java:163)
	at sun.nio.ch.SocketAdaptor.shutdownOutput(SocketAdaptor.java:409)
	at org.apache.zookeeper.ClientCnxnSocketNIO.cleanup(ClientCnxnSocketNIO.java:205)
	at org.apache.zookeeper.ClientCnxn$SendThread.cleanup(ClientCnxn.java:1338)
	at org.apache.zookeeper.ClientCnxn$SendThread.cleanAndNotifyState(ClientCnxn.java:1276)
	at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1254)
Caused by: java.nio.channels.NotYetConnectedException: null
	at sun.nio.ch.SocketChannelImpl.shutdownOutput(SocketChannelImpl.java:799)
	at sun.nio.ch.SocketAdaptor.shutdownOutput(SocketAdaptor.java:407)
	... 4 common frames omitted

org.apache.zookeeper.KeeperException$ConnectionLossException: KeeperErrorCode = ConnectionLoss for /zk_demo

	at org.apache.zookeeper.KeeperException.create(KeeperException.java:102)
	at org.apache.zookeeper.KeeperException.create(KeeperException.java:54)
	at org.apache.zookeeper.ZooKeeper.getData(ZooKeeper.java:2131)
	at org.apache.zookeeper.ZooKeeper.getData(ZooKeeper.java:2160)
	at com.yuanrengu.demo.ZooKeeperDemo.testGet(ZooKeeperDemo.java:48)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
	at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
	at org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
	at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
	at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
	at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
	at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
	at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
	at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)
	at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
	at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)

Disconnected from the target VM, address: '127.0.0.1:60454', transport: 'socket'

Process finished with exit code -1

起初以為是ZooKeeper服務部署有問題或服務沒啟動，經檢查確認無誤后，debug調試發現，是SESSION_TIME_OUT = 2000;設置的值太小，改為10000后，不再報錯。

SESSION_TIME_OUT 是會話超時時間，也就是當一個zookeeper超過該時間沒有心跳，則認為該節點故障。所以，如果此值小于zookeeper的創建時間，則當zookeeper還未來得及創建連接，會話時間已到，因此拋出異常認為該節點故障。

3.1 創建會話

通過創建一個ZooKeeper實例來連接ZooKeeper服務器（詳見ZooKeeper單機版、偽集群和集群環境搭建）。

官方提供了10種ZooKeeper構造方法和描述如下：

關于每種構造方法的英文描述用詞都很簡單，基本都能看的很明白，根據實際的應用場景選取相應的構造方法。

有傳入參數中包括sessionId和sessionPasswd的構造方法，分別代表會話ID和會話密鑰。這兩個參數能夠唯一確定一個會話，同時客戶端使用這兩個參數可以實現客戶端會話復用，從而達到恢復會話的效果。具體使用方法是第一次連接上ZooKeeper服務器時，通過調用ZooKeeper對象實例的以下兩個接口，即可獲取當前會話的ID和密鑰：long getSessionId(); byte[] getSessionPasswd();獲取到這兩個參數值之后，就可以在下次創建ZooKeeper對象實例的時候傳入構造方法了。

選取幾個典型的構造方法來帶領大家解讀下文檔。

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

public ZooKeeper(String connectString,
                 int sessionTimeout,
                 Watcher watcher)
          throws IOException
          
To create a ZooKeeper client object, the application needs to pass a connection string containing a comma separated list of host:port pairs, each corresponding to a ZooKeeper server.
Session establishment is asynchronous. This constructor will initiate connection to the server and return immediately - potentially (usually) before the session is fully established. The watcher argument specifies the watcher that will be notified of any changes in state. This notification can come at any point before or after the constructor call has returned.

The instantiated ZooKeeper client object will pick an arbitrary server from the connectString and attempt to connect to it. If establishment of the connection fails, another server in the connect string will be tried (the order is non-deterministic, as we random shuffle the list), until a connection is established. The client will continue attempts until the session is explicitly closed.

Added in 3.2.0: An optional "chroot" suffix may also be appended to the connection string. This will run the client commands while interpreting all paths relative to this root (similar to the unix chroot command).

Parameters:
connectString - comma separated host:port pairs, each corresponding to a zk server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If the optional chroot suffix is used the example would look like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a" where the client would be rooted at "/app/a" and all paths would be relative to this root - ie getting/setting/etc... "/foo/bar" would result in operations being run on "/app/a/foo/bar" (from the server perspective).
sessionTimeout - session timeout in milliseconds
watcher - a watcher object which will be notified of state changes, may also be notified for node events
Throws:
IOException - in cases of network failure
IllegalArgumentException - if an invalid chroot path is specified

有一點需要特別說明下，文檔說客戶端和服務端建立會話是異步的。構造方法會在處理完客戶端初始化工作后立即返回，在通常情況下，此時并沒有真正建立好一個可用的會話，此時在會話的生命周期中處于“CONNECTING”的狀態。當該會話真正創建完畢后，ZooKeeper服務端會向會話對應的客戶端發送一個事件通知，以告知客戶端，客戶端只有在獲取這個通知后，才算真正建立了會話。

實例化的ZooKeeper客戶端對象將從connectString列舉的服務器中隨機選擇一個服務器，并嘗試連接到該服務器。如果建立連接失敗，將嘗試連接另一個服務器（順序是不確定的，因為列舉的服務器是隨機洗牌的），直到建立連接。即客戶端連接一個服務器失敗，將繼續嘗試，直到會話顯式關閉。

從3.2.0版本開始添加了可選的"chroot"后綴，意思就是可將“chroot”加在connectString中列舉的服務器后面，即客戶端連上ZooKeeper服務器后，所有對ZooKeeper的操作都會基于這個根目錄。

對參數做下簡要說明：

參數	描述
connectString	指定ZooKeeper服務器列表，有英文逗號分隔的host：port字符串組成，如"127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002"。可以指定客戶端連上connectString中服務器后的根目錄，如 "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a" ，`對ZooKeeper的操作都會基于/app/a這個根目錄，即創建路徑為"/foo/bar"的節點，實際該節點的路徑為"/app/a/foo/bar"` 。
sessionTimeout	會話的超時事件，以“毫秒”為單位的整型值。在一個會話周期內，ZooKeeper客戶端和服務器之間會通過心跳檢測機制來維持會話的有效性，一旦在sessionTimeout時間內沒有進行有效的心跳檢測，會話就會失效。
watcher	ZooKeeper允許客戶端在構造方法中傳入一個接口Watcher（org.apache.zookeeper.Watcher）的實現類對象來作為默認的Watch事件通知器。該參數也可以設置為`null`，表明不需要設置默認的Watch處理器。

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

public ZooKeeper(String connectString,
                 int sessionTimeout,
                 Watcher watcher,
                 boolean canBeReadOnly)
          throws IOException
          
To create a ZooKeeper client object, the application needs to pass a connection string containing a comma separated list of host:port pairs, each corresponding to a ZooKeeper server.
Session establishment is asynchronous. This constructor will initiate connection to the server and return immediately - potentially (usually) before the session is fully established. The watcher argument specifies the watcher that will be notified of any changes in state. This notification can come at any point before or after the constructor call has returned.

The instantiated ZooKeeper client object will pick an arbitrary server from the connectString and attempt to connect to it. If establishment of the connection fails, another server in the connect string will be tried (the order is non-deterministic, as we random shuffle the list), until a connection is established. The client will continue attempts until the session is explicitly closed.

Added in 3.2.0: An optional "chroot" suffix may also be appended to the connection string. This will run the client commands while interpreting all paths relative to this root (similar to the unix chroot command).

Parameters:
connectString - comma separated host:port pairs, each corresponding to a zk server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If the optional chroot suffix is used the example would look like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a" where the client would be rooted at "/app/a" and all paths would be relative to this root - ie getting/setting/etc... "/foo/bar" would result in operations being run on "/app/a/foo/bar" (from the server perspective).
sessionTimeout - session timeout in milliseconds
watcher - a watcher object which will be notified of state changes, may also be notified for node events
canBeReadOnly - (added in 3.4) whether the created client is allowed to go to read-only mode in case of partitioning. Read-only mode basically means that if the client can't find any majority servers but there's partitioned server it could reach, it connects to one in read-only mode, i.e. read requests are allowed while write requests are not. It continues seeking for majority in the background.
Throws:
IOException - in cases of network failure
IllegalArgumentException - if an invalid chroot path is specified

這個構造方法跟上個方法非常相似，只是從3.4版本開始多了一個canBeReadOnly參數，用于標識當前會話是否支持“read-only”模式（只讀模式）。默認情況下，在ZooKeeper集群中，一個節點如果和集群中過半及以上節點失去網絡連接（建立不了連接），那么這個機器將不再處理客戶端請求（包括讀寫請求）。但是在某些使用場景下，當ZooKeeper服務器發生此類故障的時候，還是希望ZooKeeper服務器能夠提供讀服務（寫服務肯定無法提供），這就是ZooKeeper的“read-only”模式。但客戶端可以連接某一分區的服務器，它將以只讀模式連接到其中一個服務器，允許讀取請求，而不允許寫入請求，然后繼續在后臺尋找更多數的服務器(這一句我描述的不夠簡練精準)。

3.2 新增

public String create(String path,
                     byte[] data,
                     List<ACL> acl,
                     CreateMode createMode)
              throws KeeperException,
                     InterruptedException
                     
Create a node with the given path. The node data will be the given data, and node acl will be the given acl.
The flags argument specifies whether the created node will be ephemeral or not.

An ephemeral node will be removed by the ZooKeeper automatically when the session associated with the creation of the node expires.

The flags argument can also specify to create a sequential node. The actual path name of a sequential node will be the given path plus a suffix "i" where i is the current sequential number of the node. The sequence number is always fixed length of 10 digits, 0 padded. Once such a node is created, the sequential number will be incremented by one.

If a node with the same actual path already exists in the ZooKeeper, a KeeperException with error code KeeperException.NodeExists will be thrown. Note that since a different actual path is used for each invocation of creating sequential node with the same path argument, the call will never throw "file exists" KeeperException.

If the parent node does not exist in the ZooKeeper, a KeeperException with error code KeeperException.NoNode will be thrown.

An ephemeral node cannot have children. If the parent node of the given path is ephemeral, a KeeperException with error code KeeperException.NoChildrenForEphemerals will be thrown.

This operation, if successful, will trigger all the watches left on the node of the given path by exists and getData API calls, and the watches left on the parent node by getChildren API calls.

If a node is created successfully, the ZooKeeper server will trigger the watches on the path left by exists calls, and the watches on the parent of the node by getChildren calls.

The maximum allowable size of the data array is 1 MB (1,048,576 bytes). Arrays larger than this will cause a KeeperExecption to be thrown.

Parameters:
path - the path for the node
data - the initial data for the node
acl - the acl for the node
createMode - specifying whether the node to be created is ephemeral and/or sequential
Returns:
the actual path of the created node
Throws:
KeeperException - if the server returns a non-zero error code
KeeperException.InvalidACLException - if the ACL is invalid, null, or empty
InterruptedException - if the transaction is interrupted
IllegalArgumentException - if an invalid path is specified

Talk is cheap. Show me the code.這里我們不瞎BB，直接上官方文檔。官方文檔是不是很容易看懂，而且解釋的非常清楚（而且稍顯啰嗦的感覺）？

這里簡單列下文檔中的幾個關鍵點：

按指定路徑和節點形式創建，可指定節點為持久節點、臨時節點等。
這里要說下CreateMode,大家可能都說ZooKeeper只有4種形式的節點（持久、臨時、持久順序、臨時順序），看文檔的話，其實是有7種形式的。

public enum CreateMode {
   PERSISTENT(0, false, false, false, false),
   PERSISTENT_SEQUENTIAL(2, false, true, false, false),
   EPHEMERAL(1, true, false, false, false),
   EPHEMERAL_SEQUENTIAL(3, true, true, false, false),
   CONTAINER(4, false, false, true, false),
   PERSISTENT_WITH_TTL(5, false, false, false, true),
   PERSISTENT_SEQUENTIAL_WITH_TTL(6, false, true, false, true);
}

PERSISTENT：持久節點（也有叫永久節點的），不會隨著會話的結束而自動刪除。
PERSISTENT_SEQUENTIAL：帶單調遞增序號的持久節點，不會隨著會話的結束而自動刪除。
EPHEMERAL：臨時節點，會隨著會話的結束而自動刪除。
EPHEMERAL_SEQUENTIAL：帶單調遞增序號的臨時節點，會隨著會話的結束而自動刪除。
CONTAINER：容器節點，用于Leader、Lock等特殊用途，當容器節點不存在任何子節點時，容器將成為服務器在將來某個時候刪除的候選節點。
PERSISTENT_WITH_TTL：帶TTL（time-to-live，存活時間）的持久節點，節點在TTL時間之內沒有得到更新并且沒有子節點，就會被自動刪除。
PERSISTENT_SEQUENTIAL_WITH_TTL：帶TTL（time-to-live，存活時間）和單調遞增序號的持久節點，節點在TTL時間之內沒有得到更新并且沒有子節點，就會被自動刪除。

如果指令路徑和版本的節點已經存在，則會拋出一個KeeperException異常。
臨時節點不能有子節點。如果給臨時節點創建子節點會拋KeeperException異常。
臨時節點的生命周期與客戶端會話綁定。一旦客戶端會話失效（客戶端與 Zookeeper連接斷開不一定會話失效），那么這個客戶端創建的所有臨時節點都會被移除。
byte[] data允許的最大數據量為1MB（1,048,576 bytes）。如果超過，會拋KeeperExecption。

運行創建節點的代碼：

    @Test
    public void testCreate() throws KeeperException, InterruptedException {
        zk.create(ZNODE_PATH, "anna2019".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

可以通過日志信息得到節點創建成功：

DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x101402626bb000b, packet:: clientPath:null serverPath:null finished:false header:: 1,1  replyHeader:: 1,12884901937,0  request:: '/zk_demo,#616e6e6132303139,v{s{31,s{'world,'anyone}}},0  response:: '/zk_demo

在服務端查看,/zk_demo節點創建成功：

[zk: 127.0.0.1:2181(CONNECTED) 21] ls /
[zookeeper, zk_demo]
[zk: 127.0.0.1:2181(CONNECTED) 22] stat /zk_demo
cZxid = 0x300000031
ctime = Tue Dec 17 12:52:50 CST 2019
mZxid = 0x300000031
mtime = Tue Dec 17 12:52:50 CST 2019
pZxid = 0x300000031
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 8
numChildren = 0

3.3 獲取

public byte[] getData(String path,
                      boolean watch,
                      Stat stat)
               throws KeeperException,
                      InterruptedException

Return the data and the stat of the node of the given path.
If the watch is true and the call is successful (no exception is thrown), a watch will be left on the node with the given path. The watch will be triggered by a successful operation that sets data on the node, or deletes the node.

A KeeperException with error code KeeperException.NoNode will be thrown if no node with the given path exists.

Parameters:
path - the given path
watch - whether need to watch this node
stat - the stat of the node
Returns:
the data of the node
Throws:
KeeperException - If the server signals an error with a non-zero error code
InterruptedException - If the server transaction is interrupted.

指定路徑的節點不存在時就拋KeeperException.NoNode異常。

運行：

    @Test
    public void testGet() throws KeeperException, InterruptedException {
        byte[] data1 = zk.getData(ZNODE_PATH, false, null);
        byte[] data2 = zk.getData(ZNODE_PATH_PARENT, false, null);
        byte[] data3 = zk.getData(ZNODE_PATH_CHILDREN, false, null);
        LOGGER.info("{}的信息：{}", ZNODE_PATH, new String(data1) );
        LOGGER.info("{}的信息：{}", ZNODE_PATH_PARENT, new String(data2) );
        LOGGER.info("{}的信息：{}", ZNODE_PATH_CHILDREN, new String(data3) );
    }

結果：

13:51:00.288 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - /zk_demo的信息：anna2019
13:51:00.288 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - /app1的信息：anna2019
13:51:00.289 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - /app1/app1_1的信息：anna2020

3.4 更新

public Stat setData(String path,
                    byte[] data,
                    int version)
             throws KeeperException,
                    InterruptedException
                    
Set the data for the node of the given path if such a node exists and the given version matches the version of the node (if the given version is -1, it matches any node's versions). Return the stat of the node.
This operation, if successful, will trigger all the watches on the node of the given path left by getData calls.

A KeeperException with error code KeeperException.NoNode will be thrown if no node with the given path exists.

A KeeperException with error code KeeperException.BadVersion will be thrown if the given version does not match the node's version.

The maximum allowable size of the data array is 1 MB (1,048,576 bytes). Arrays larger than this will cause a KeeperException to be thrown.

Parameters:
path - the path of the node
data - the data to set
version - the expected matching version
Returns:
the state of the node
Throws:
InterruptedException - If the server transaction is interrupted.
KeeperException - If the server signals an error with a non-zero error code.
IllegalArgumentException - if an invalid path is specified

主要注意以下幾點：

版本為-1時，即代表適配指定路徑節點的所有版本。
如果指定路徑的節點不存在會拋KeeperException.NoNode異常，該節點沒有傳入的版本，會拋KeeperException.BadVersion異常。
byte[] data允許的最大數據量為1MB（1,048,576 bytes）。如果超過，會拋KeeperExecption。

運行：

    @Test
    public void testSet() throws KeeperException, InterruptedException {
        Stat stat = zk.setData(ZNODE_PATH, "yuanrengu".getBytes(), -1);
        byte[] data = zk.getData(ZNODE_PATH, false, null);
        LOGGER.info(new String(data));
    }

可以看到數據已經更新：

15:46:16.472 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - yuanrengu

更新的接口提到了版本的概念，上面提到版本為-1時，即代表適配指定路徑節點的所有版本。節點每次setData時版本會加1，更新時指定的版本不存在會報KeeperException.BadVersion異常。我們做個測試：

    @Test
    public void testSetForVersion() throws KeeperException, InterruptedException {
        String pathVersion = "/versionDemo";
        zk.create(pathVersion, "yuanrengu2019".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        Stat stat = zk.setData(pathVersion, "yuanrengu2020".getBytes(), -1);
        LOGGER.info("============1111111  start======================");
        LOGGER.info(String.valueOf(stat));
        LOGGER.info("version:{}", stat.getVersion());
        byte[] data1 = zk.getData(pathVersion, false, null);
        LOGGER.info("data1:{}", new String(data1));
        LOGGER.info("============1111111  end======================");

        Stat stat2 = zk.setData(pathVersion, "yuanrengu2021".getBytes(), stat.getVersion());
        LOGGER.info("============222222  start======================");
        LOGGER.info(String.valueOf(stat2));
        LOGGER.info("version2:{}", stat2.getVersion());
        byte[] data2 = zk.getData(pathVersion, false, null);
        LOGGER.info("data2:{}", new String(data2));
        LOGGER.info("============222222  end======================");

        Stat stat3 = zk.setData(pathVersion, "yuanrengu2022".getBytes(), stat.getVersion());
        LOGGER.info("============3333333  start======================");
        LOGGER.info(String.valueOf(stat3));
        LOGGER.info("version3:{}", stat3.getVersion());
        byte[] data3 = zk.getData(pathVersion, false, null);
        LOGGER.info("data3:{}", new String(data3));
        LOGGER.info("============3333333  end======================");
    }

運行結果如下：

09:56:00.931 [main-SendThread(106.12.111.172:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x1014dce26220008, packet:: clientPath:null serverPath:null finished:false header:: 1,5  replyHeader:: 1,12884902005,0  request:: '/versionDemo,#7975616e72656e677532303230,-1  response:: s{12884901996,12884902005,1576720362715,1576720560918,1,0,0,0,13,0,12884901996} 
09:56:00.940 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - ============1111111  start======================
09:56:00.941 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - 12884901996,12884902005,1576720362715,1576720560918,1,0,0,0,13,0,12884901996

09:56:00.942 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - version:1
09:56:00.971 [main-SendThread(106.12.111.172:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x1014dce26220008, packet:: clientPath:null serverPath:null finished:false header:: 2,4  replyHeader:: 2,12884902005,0  request:: '/versionDemo,F  response:: #7975616e72656e677532303230,s{12884901996,12884902005,1576720362715,1576720560918,1,0,0,0,13,0,12884901996} 
09:56:00.971 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - data1:yuanrengu2020
09:56:00.971 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - ============1111111  end======================
09:56:00.988 [main-SendThread(106.12.111.172:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x1014dce26220008, packet:: clientPath:null serverPath:null finished:false header:: 3,5  replyHeader:: 3,12884902006,0  request:: '/versionDemo,#7975616e72656e677532303231,1  response:: s{12884901996,12884902006,1576720362715,1576720561002,2,0,0,0,13,0,12884901996} 
09:56:00.988 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - ============222222  start======================
09:56:00.988 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - 12884901996,12884902006,1576720362715,1576720561002,2,0,0,0,13,0,12884901996

09:56:00.990 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - version2:2
09:56:01.017 [main-SendThread(106.12.111.172:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x1014dce26220008, packet:: clientPath:null serverPath:null finished:false header:: 4,4  replyHeader:: 4,12884902006,0  request:: '/versionDemo,F  response:: #7975616e72656e677532303231,s{12884901996,12884902006,1576720362715,1576720561002,2,0,0,0,13,0,12884901996} 
09:56:01.017 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - data2:yuanrengu2021
09:56:01.017 [main] INFO com.yuanrengu.demo.ZooKeeperDemo - ============222222  end======================
09:56:01.037 [main-SendThread(106.12.111.172:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x1014dce26220008, packet:: clientPath:null serverPath:null finished:false header:: 5,5  replyHeader:: 5,12884902007,-103  request:: '/versionDemo,#7975616e72656e677532303232,1  response::  

org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode = BadVersion for /versionDemo

	at org.apache.zookeeper.KeeperException.create(KeeperException.java:122)
	at org.apache.zookeeper.KeeperException.create(KeeperException.java:54)
	at org.apache.zookeeper.ZooKeeper.setData(ZooKeeper.java:2384)

測試代碼進行了3次setData操作，第一次setData時傳入的版本為-1，成功后version變為1；第二次setData時傳入的version為1，成功后version變為2；第三次setData時傳入的版本為1，此時就拋了KeeperException.BadVersion異常。如果第三次setData傳入的版本為-1，能更新成功。

3.5 刪除

public void delete(String path,
                   int version)
            throws InterruptedException,
                   KeeperException
                   
Delete the node with the given path. The call will succeed if such a node exists, and the given version matches the node's version (if the given version is -1, it matches any node's versions).
A KeeperException with error code KeeperException.NoNode will be thrown if the nodes does not exist.

A KeeperException with error code KeeperException.BadVersion will be thrown if the given version does not match the node's version.

A KeeperException with error code KeeperException.NotEmpty will be thrown if the node has children.

This operation, if successful, will trigger all the watches on the node of the given path left by exists API calls, and the watches on the parent node left by getChildren API calls.

Parameters:
path - the path of the node to be deleted.
version - the expected node version.
Throws:
InterruptedException - IF the server transaction is interrupted
KeeperException - If the server signals an error with a non-zero return code.
IllegalArgumentException - if an invalid path is specified

節點可能含有子節點，刪除節點的操作有幾點需要特別注意：

版本為-1時，即代表適配指定路徑節點的所有版本。
如果指定路徑的節點不存在會拋KeeperException.NoNode異常，該節點沒有傳入的版本，會拋KeeperException.BadVersion異常。
如果節點含有子節點，刪除父節點(parent node)時會拋KeeperException.NotEmpty異常。

在ZooKeeper中，只允許刪子節點。如果一個節點存在一個或多個子節點，該節點就無法被直接刪除，必須先刪除所有子節點。

/app1有子節點，我們做下刪除操作：

    /**
     *  刪除含有子節點的父節點
     * @throws KeeperException
     * @throws InterruptedException
     */
    @Test
    public void testDeleteHasChildrenZnode() throws KeeperException, InterruptedException {
        // 指定要刪除的版本，-1表示刪除所有版本
        zk.delete(ZNODE_PATH_PARENT, -1);
    }

可以看到日志：

org.apache.zookeeper.KeeperException$NotEmptyException: KeeperErrorCode = Directory not empty for /app1

	at org.apache.zookeeper.KeeperException.create(KeeperException.java:132)
	at org.apache.zookeeper.KeeperException.create(KeeperException.java:54)
	at org.apache.zookeeper.ZooKeeper.delete(ZooKeeper.java:1793)
	at com.yuanrengu.demo.ZooKeeperDemo.testDeleteHasChildrenZnode(ZooKeeperDemo.java:89)

4 總結

上面我們實現了節點的增、刪、改、查的測試，后面的篇章會有更多好玩的用法，如實現分布式鎖、配置中心等。

基于上面的分析，總結幾個注意的點：

節點有7種形式：

PERSISTENT：持久節點（也有叫永久節點的），不會隨著會話的結束而自動刪除。
PERSISTENT_SEQUENTIAL：帶單調遞增序號的持久節點，不會隨著會話的結束而自動刪除。
EPHEMERAL：臨時節點，會隨著會話的結束而自動刪除。
EPHEMERAL_SEQUENTIAL：帶單調遞增序號的臨時節點，會隨著會話的結束而自動刪除。
CONTAINER：容器節點，用于Leader、Lock等特殊用途，當容器節點不存在任何子節點時，容器將成為服務器在將來某個時候刪除的候選節點。
PERSISTENT_WITH_TTL：帶TTL（time-to-live，存活時間）的持久節點，節點在TTL時間之內沒有得到更新并且沒有子節點，就會被自動刪除。
PERSISTENT_SEQUENTIAL_WITH_TTL：帶TTL（time-to-live，存活時間）和單調遞增序號的持久節點，節點在TTL時間之內沒有得到更新并且沒有子節點，就會被自動刪除。

臨時節點不能有子節點。如果給臨時節點創建子節點會拋KeeperException異常。
臨時節點的生命周期與客戶端會話綁定。一旦客戶端會話失效（客戶端與 Zookeeper連接斷開不一定會話失效），那么這個客戶端創建的所有臨時節點都會被移除。
byte[] data允許的最大數據量為1MB（1,048,576 bytes）。

posted on 2019-12-18 10:00 猿人谷閱讀(3312) 評論(0) 收藏舉報