碎碎念：啟動成功了一半，可以啟動，可以debug，但是有些方法無法訪問，而且create在哪里，我還不清楚，那個DataMonitor，不能完全按照官網寫，要像我一樣改一下，不然會報werror，因為有些過時了

ZooKeeper的Java實體

一個簡單的watch客戶端
- 要求
- 程式設計
- Executor
- DataMonitor
完整代碼
- Executor
- DataMonitor
啟動
參考鏈接

一個簡單的watch客戶端

作用：監視ZooKeeper節點的更改，對程式的啟動或停止做出回應，

要求

1 它需要四個引數：
zk服務器的地址
被監視節點的名字
輸出要寫入的檔案名
帶引數的可執行檔案

是這樣理解嗎？

2 獲取與znode關聯的資料并啟動可執行檔案
3 若被監視的znode發生更改，客戶機將重新獲取內容并重啟啟動可執行程式
4 若被監視的znode消失，客戶端將殺死可執行程式

程式設計

ZooKeeper應用程式分為兩部分：維護與服務器連接和監視節點資料，Executor類負責維護連接部分，DataMonitor負責監視zookeeper樹中的資料，Executor包含主執行緒和執行邏輯，它負責少量的用戶互動，以及與可執行程式（根據被監視的znode節點的狀態停止或重啟）的互動，

Executor

Executor物件是這個簡單示例中的基本容器，它包含了ZooKeeper物件和DataMonitor，

// from the Executor class...

public static void main(String[] args) {
    if (args.length < 4) {
        System.err
                .println("USAGE: Executor hostPort znode filename program [args ...]");
        System.exit(2);
    }
    String hostPort = args[0];
    String znode = args[1];
    String filename = args[2];
    String exec[] = new String[args.length - 3];
    System.arraycopy(args, 3, exec, 0, exec.length);
    try {
    //Executor的任務是根據命令列中輸入去啟動和停止的可執行程式，以回應zk物件觸發的時間，(前面的args）
        new Executor(hostPort, znode, filename, exec).run();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

public Executor(String hostPort, String znode, String filename,
        String exec[]) throws KeeperException, IOException {
    this.filename = filename;
    this.exec = exec;
    
    zk = new ZooKeeper(hostPort, 3000, this);
    dm = new DataMonitor(zk, znode, null, this);
}

public void run() {
    try {
        synchronized (this) {
            while (!dm.dead) {
                wait();
            }
        }
    } catch (InterruptedException e) {
    }
}

Executor實作了這些

public class Executor implements Watcher, Runnable, DataMonitor.DataMonitorListener {
...
//DataMonitor.DataMonitorListener這個是啥？


ZooKeeper的java api定義了watcher 介面，zk用watcher與容器（如Executor）進行通信，watcher僅包含process()方法，zk利用它去傳遞主執行緒感興趣的時間，例如zk連接或會話的狀態，本例中的Executor僅將事件下發給DataMonitor ，然后由DataMonitor 決定如何處理，
>  就說Executor接收了，但不想做就交給DataMonitor 處理了？是不是也可以交由其他的呢？

```java
public void process(WatchedEvent event) {
    dm.process(event);
}

下面的DataMonitorListener介面是本案例設計的，不屬于zkAPI，DataMonitor 物件使用它來與其容器(如Executor)進行通信（DataMonitor.DataMonitorListener），

public interface DataMonitorListener {

    void exists(byte data[]);

    /**
    *
    * @param rc
    * the ZooKeeper reason code
    */
    void closing(int rc);

DataMonitor.DataMonitorListener這個是啥？
就是DataMonitor里面定義了DataMonitorListener介面，并由Executor實作了，

package example;

import org.apache.zookeeper.WatchedEvent;

public class DataMonitor {
    DataMonitor dm;
    public void process(WatchedEvent event) {
        dm.process(event);
    }
    public interface DataMonitorListener {
        /**
         * 節點存活與否判斷
         */
        void exists(byte data[]);

        /**
         * ZooKeeper會話失效
         * @param src
         * ZooKeeper的原因碼(reason code)
         */
        void closing(int src);
    }
}

下面是Executor 對 DataMonitorListener.exists ()和 DataMonitorListener.closing 的實作:

```java
public void exists( byte[] data ) {
    if (data == null) {
        if (child != null) {
            System.out.println("Killing process");
            child.destroy();
            try {
                child.waitFor();
            } catch (InterruptedException e) {
           }
        }
        child = null;
    } else {
        if (child != null) {
            System.out.println("Stopping child");
            child.destroy();
            try {
               child.waitFor();
            } catch (InterruptedException e) {
            e.printStackTrace();
            }
        }
        try {
            FileOutputStream fos = new FileOutputStream(filename);
            fos.write(data);
            fos.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            System.out.println("Starting child");
            child = Runtime.getRuntime().exec(exec);
            new StreamWriter(child.getInputStream(), System.out);
            new StreamWriter(child.getErrorStream(), System.err);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

public void closing(int rc) {
    synchronized (this) {
        notifyAll();
    }
}

DataMonitor

DataMonitor是本程式ZooKeeper邏輯的核心，它是異步和事件驅動的，

public DataMonitor(ZooKeeper zk, String znode, Watcher chainedWatcher,
        DataMonitorListener listener) {
    this.zk = zk;
    this.znode = znode;
    this.chainedWatcher = chainedWatcher;
    this.listener = listener;
    //檢查znode是否存在，并設定監視，
    //傳遞自身作為回呼物件，watcher觸發時就會引起真實的處理流程
    //exists在服務器端完成，但其回呼在客戶端完成
    zk.exists(znode, true, this, null);

Note

1 不要將完成回呼和watch的回呼搞混，exists()的完成回呼——（客戶機端）processResult ()是在服務器上的watch（ZooKeeper.exists()的）操作之后執行，
2 Executor注冊為了zk物件的一個watcher，所以watch觸發時會向Executor物件發送一個事件
3 zk3.0后DataMonitor也可以注冊為特定事件的watcher，本例不支持，

public void processResult(int rc, String path, Object ctx, Stat stat) {
        boolean exists;
        //1 檢查節點是否存在
        switch (rc) {
            //節點存在
            case Code.OK:
                exists = true;
                break;
            //節點不存在
            case Code.NoNode:
                exists = false;
                break;
            //會話被服務器終止（致命錯誤）
            case Code.SESSIONEXPIRED:
            //未認證（致命錯誤）
            case Code.NoAuth:
                dead = true;
                listener.closing(rc);
                return;
            default:
                //可恢復的服務
                zk.exists(znode,true,this,null);
                return;
        }

        byte b[] = null;
        // 2 存在則從znode獲取資料
        if (exists) {
            try{
                //這兩句不是太理解欸
                //如果節點在呼叫zookeeper.getData之前被洗掉，
                //zookeeper.exists()設定的watch將會觸發一個回呼

                //如果由通信錯誤，連接的watch事件會在連接恢復時觸發
                b = zk.getData(znode, false,null);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                return;
            }
        }
        //如果狀態發生變化，呼叫Executor 的 exists() 回呼函式
        //???這里不是太理解
        if ((b == null && b!= prevData)
              || b != null && !Arrays.equals(prevData, b)) {
            listener.exists(b);
            prevData = b;
        }
    }

完整代碼

Executor

package example;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

public class Executor implements Watcher, Runnable, DataMonitor.DataMonitorListener {
    String filename;
    String exec[];
    ZooKeeper zk;
    DataMonitor dm;
    Process child;

    public static void main(String[] args) {
        if (args.length < 4) {
            //（標準錯誤輸出流）實時輸出錯誤，顯示為紅色，out要累計到一定程度才輸出
            //https://blog.csdn.net/weixin_42153410/article/details/94618943
            System.err.
                    println("USAGE: Executor hostPort znode filename program [args ...]");
            System.exit(2);
        }
        String hostPort = args[0];
        String znode = args[1];
        String filename = args[2];
        String exec[] = new String[args.length - 3];
        /**
         *  Object src : 原陣列
         *    int srcPos : 從元資料的起始位置開始
         * 　　Object dest : 目標陣列
         * 　　int destPos : 目標陣列的開始起始位置
         * 　　int length  : 要copy的陣列的長度
         */
        System.arraycopy(args, 3, exec, 0, exec.length);
        try {
            new Executor(hostPort, znode, filename, exec).run();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public Executor(String hostPort, String znode, String filename,
                    String exec[]) throws KeeperException, IOException {
        this.filename = filename;
        this.exec = exec;
        //public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)這個？
        //this是Executor對自身的參考
        zk = new ZooKeeper(hostPort, 3000, this);
        dm = new DataMonitor(zk, znode, null, this);
    }

    public void run() {
        try {
            //同一時間只能有一個執行緒訪問
            synchronized (this) {
                while (!dm.dead) {
                    wait();
                }
            }
        }catch (InterruptedException e) {
        }
    }
    public void exists(byte[] data) {
        //沒有傳資料
        if (data == null) {
            //如果行程不為空
            if (child != null) {
                System.out.println("Killing process");
                //殺死子行程
                child.destroy();
                try {
                    //讓執行緒等待到終止為止
                    child.waitFor();
                } catch (InterruptedException e) {
                    //執行緒在等待時中斷
                    e.printStackTrace();
                }
            }
            //行程置空
            child = null;
        } else {
            //data有資料（znode存在，或發生變化？）
            if (child != null) {
                //但是行程不為空
                System.out.println("Stopping child");
                //先終止行程
                child.destroy();
                try {
                    child.waitFor();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            try {
                //將znode資料存入檔案
                FileOutputStream fos = new FileOutputStream(filename);
                fos.write(data);
                fos.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                //啟動行程
                System.out.println("Starting child");
                //getRuntime 回傳與當前Java應用程式關聯的運行時物件(Runtime)
                //exec 在單獨的行程中執行指定的字串命令
                // 即，執行緒執行用戶的命令
                child = Runtime.getRuntime().exec(exec);
                //兩個執行緒輸出執行結果及日志
                new StreamWriter(child.getInputStream(), System.out);
                new StreamWriter(child.getErrorStream(), System.err);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    public void closing(int rc) {
        synchronized (this) {
            //喚醒物件的等待池中的所有執行緒，進入鎖池
            // 和會話失效有啥關系？
            notifyAll();
        }
    }

    static class StreamWriter extends Thread {
        OutputStream os;
        InputStream is;
        StreamWriter(InputStream is,OutputStream os) {
            this.is = is;
            this.os = os;
            start();
        }
        public void run() {
            byte b[] = new byte[80];
            int rc;
            try {
                while ((rc = is.read(b)) > 0) {
                    os.write(b,0,rc);
                }
            } catch (IOException e) {}
        }
    }
    public void process(WatchedEvent event) {
        dm.process(event);
    }
}

DataMonitor

package example;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.KeeperException.Code;

import java.util.Arrays;

public class DataMonitor implements Watcher, AsyncCallback.StatCallback {

    ZooKeeper zk;
    String znode;
    Watcher chainedWatcher;
    DataMonitorListener listener;
    boolean dead;
    byte prevData[];
    public DataMonitor(ZooKeeper zk, String znode, Watcher chainedWatcher,
                       DataMonitorListener listener) {
        this.zk = zk;
        this.znode = znode;
        this.listener = listener;
        //檢查znode是否存在，并設定監視，
        //傳遞自身作為回呼物件，watcher觸發時就會引起真實的處理流程
        //exists在服務器端完成，但其回呼在客戶端完成
        zk.exists(znode,true,this,null);
    }

    public interface DataMonitorListener {
        /**
         * 節點存活與否判斷
         */
        void exists(byte data[]);

        /**
         * ZooKeeper會話失效
         * @param src
         * ZooKeeper的原因碼(reason code)
         */
        void closing(int src);
    }

    /**
     * 檢查節點存在與否，存在且狀態變化的呼叫 Executor 的 exists() 回呼函式
     * @param rc
     * @param path
     * @param ctx
     * @param stat
     */
    public void processResult(int rc, String path, Object ctx, Stat stat) {
        boolean exists = true;
        //1 檢查節點是否存在
        Code code = Code.get(rc);
        switch (code) {
            //節點存在
            case OK:
                exists = true;
                break;
            //節點不存在
            case NONODE:
                exists = false;
                break;
            //會話被服務器終止（致命錯誤）
            case SESSIONEXPIRED:
            //未認證（致命錯誤）
            case NOAUTH:
                dead = true;
                listener.closing(rc);
                return;
            default:
                //可恢復的服務
                zk.exists(znode,true,this,null);
                return;
        }

        byte b[] = null;
        // 2 存在則從znode獲取資料
        if (exists) {
            try{
                //這兩句不是太理解欸
                //如果節點在呼叫zookeeper.getData之前被洗掉，
                //zookeeper.exists()設定的watch將會觸發一個回呼

                //如果由通信錯誤，連接的watch事件會在連接恢復時觸發
                b = zk.getData(znode, false,null);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                return;
            }
        }
        //如果狀態發生變化，呼叫Executor 的 exists() 回呼函式
        //???這里不是太理解
        if ((b == null && b!= prevData)
                || b != null && !Arrays.equals(prevData, b)) {
            listener.exists(b);
            prevData = b;
        }
    }

    /**
     * 處理watch事件
     */
    public void process(WatchedEvent event) {
        String path = event.getPath();
        if (event.getType() == Watcher.Event.EventType.None) {
            //我們被告知連接狀態已經變化
            switch(event.getState()) {
                case SyncConnected:
                    //在這個例子中，不需要做任何事情-watch自動和客戶端重連和注冊；
                    //客戶端斷連時watch依次觸發
                    break;
                case Expired:
                    dead = true;
                    listener.closing(Code.SESSIONEXPIRED.intValue());
                    break;
            }
        } else {
            if (path != null && path.equals(znode)) {
                zk.exists(znode,true,this,null);
            }
        }
        if (chainedWatcher != null) {
            chainedWatcher.process(event);
        }
    }
}

啟動

ZookeeperServerMain先啟動，可參考

-Dlog4j.configuration=file:conf/log4j.properties
conf/zoo.cfg

在這里插入圖片描述

-Dlog4j.configuration=file:conf/log4j.properties
//以空格作為分割，第一個引數是地址，第二個是監視node，第三個是輸出的檔案地址 第四個是命令

127.0.0.1:2181 /zk_test E:/soft/kj/ZooKeeper/zookeeper-3.4.13/1.txt create

在這里插入圖片描述