ZooKeeper官方文檔之Java客戶端開發(fā)案例翻譯

更新時(shí)間：2022年01月27日 15:08:18 作者：愛碼叔（稀有氣體）

網(wǎng)上有很多ZooKeeper的java客戶端例子，我也看過很多，不過大部分寫的都不好，有各種問題。兜兜轉(zhuǎn)轉(zhuǎn)還是覺得官方給的例子最為經(jīng)典，在學(xué)習(xí)之余翻譯下來，供朋友們參考

一個(gè)簡(jiǎn)單的監(jiān)聽客戶端

通過開發(fā)一個(gè)非常簡(jiǎn)單的監(jiān)聽客戶端，為你介紹ZooKeeper的Java API。此ZooKeeper的客戶端，監(jiān)聽ZooKeeper中node的變化并做出響應(yīng)。

需求

這個(gè)客戶端有如下四個(gè)需求：

1、它接收如下參數(shù)：

ZooKeeper服務(wù)的地址
被監(jiān)控的znode的名稱
可執(zhí)行命令參數(shù)

2、它會(huì)取得znode上關(guān)聯(lián)的數(shù)據(jù)，然后執(zhí)行命令

3、如果znode變化，客戶端重新拉取數(shù)據(jù)，再次執(zhí)行命令

4、如果znode消失了，客戶端殺掉進(jìn)行的執(zhí)行命令。

程序設(shè)計(jì)

一般我們會(huì)這么做，把ZooKeeper的程序分成兩個(gè)單元，一個(gè)維護(hù)連接，另外一個(gè)監(jiān)控?cái)?shù)據(jù)。本程序中Executor類維護(hù)ZooKeeper的連接，DataMonitor監(jiān)控ZooKeeper的數(shù)據(jù)。同時(shí)，Executor維護(hù)主線程以及執(zhí)行邏輯。它負(fù)責(zé)對(duì)用戶的交互做出響應(yīng)，這里的交互既指根據(jù)你傳入?yún)?shù)做出響應(yīng)，也指根據(jù)znode的狀態(tài)，關(guān)閉和重啟。

Executor類

// 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 {
            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的工作是啟停通過命令行傳入的執(zhí)行命令。他通過響應(yīng)ZooKeeper對(duì)象觸發(fā)的事件來實(shí)現(xiàn)。就像上面的代碼，在ZooKeeper的構(gòu)造器中，Executor傳遞自己的引用作為watcher參數(shù)。同時(shí)，他傳遞自己的引用作為DataMonitorLisrener參數(shù)給DataMonitor構(gòu)造器。在Executor定義中，實(shí)現(xiàn)了這些接口。

public class Executor implements Watcher, Runnable, DataMonitor.DataMonitorListener {
...

ZooKeeper的Java API定義了Watcher接口。ZooKeeper用它來反饋給它的持有者。它僅支持一個(gè)方法process()，ZooKeeper用它來反饋主線程感興趣的通用事件，例如ZooKeeper的連接狀態(tài)，或者ZooKeeper session的狀態(tài)。例子中的Executor只是簡(jiǎn)單的把事件傳遞給DataMonitor，由DataMonitor來決定怎么處理。為了方便，Executor或者其他的類似Executor的對(duì)象持有ZooKeeper連接，但是可以很自由的把事件委派給其他對(duì)象。它也用此作為觸發(fā)watch事件的默認(rèn)渠道。

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

DataMonitorListener接口，并不是ZooKeeper提供的API。它是為這個(gè)示例程序設(shè)計(jì)的自定義接口。DataMonitor對(duì)象用它為它的持有者（也是Executor對(duì)象）反饋，

DataMonitorListener接口是下面這個(gè)樣子：

public interface DataMonitorListener {
    /**
    * The existence status of the node has changed.
    */
    void exists(byte data[]); 
    /**
    * The ZooKeeper session is no longer valid.
    * 
    * @param rc
    * the ZooKeeper reason code
    */
    void closing(int rc);
}

這個(gè)接口定義在DataMonitor類中，被Executor類實(shí)現(xiàn)。當(dāng)調(diào)用Executor.exists()，Executor根據(jù)需求決定是否啟動(dòng)還是關(guān)閉。回憶一下，需求提到當(dāng)znode不再存在時(shí)，殺掉進(jìn)行中的執(zhí)行命令。

當(dāng)調(diào)用Executor.closing()，作為對(duì)ZooKeeper連接永久消失的響應(yīng)，Executor決定是否關(guān)閉它自己。

就像你可能猜想的那樣，，作為對(duì)ZooKeeper狀態(tài)變化的響應(yīng)，這些方法的調(diào)用者是DataMonitor。

下面是Exucutor中 DataMonitorListener.exists()和DataMonitorListener.closing()的實(shí)現(xiàn)

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類

ZooKeeper的邏輯都在DataMonitor類中。他是異步和事件驅(qū)動(dòng)的。DataMonitor在構(gòu)造函數(shù)中完成啟動(dòng)。

public DataMonitor(ZooKeeper zk, String znode, Watcher chainedWatcher, DataMonitorListener listener) {
        this.zk = zk; 
        this.znode = znode; 
        this.chainedWatcher = chainedWatcher; 
        this.listener = listener;
    // Get things started by checking if the node exists. We are going 
    // to be completely event driven zk.exists(znode, true, this, null);
    }

對(duì)zk.exists()的調(diào)用，會(huì)檢查znode是否存在，設(shè)置watch，傳遞他自己的引用作為完成后的回調(diào)對(duì)象。這意味著，當(dāng)watch被引發(fā)，真正的處理才開始。

Note

不要把完成回調(diào)和watch回調(diào)搞混。ZooKeeper.exists()完成時(shí)的回調(diào)，發(fā)生在DataMonitor對(duì)象實(shí)現(xiàn)的的StatCallback.processResult()方法中，調(diào)用發(fā)生在server上異步的watch設(shè)置操作（通過zk.exists()）完成時(shí)。

另一邊，watch觸發(fā)時(shí)，給Executor對(duì)象發(fā)送了一個(gè)事件，因?yàn)镋xecutor注冊(cè)成為ZooKeeper對(duì)象的一個(gè)watcher。

你可能注意到DataMonitor也可以注冊(cè)它自己作為這個(gè)特定事件的watcher。這是ZooKeeper 3.0.0中加入的（多watcher的支持）。在這個(gè)例子中，DataMonitor并沒有注冊(cè)為watcher（譯者：這里指zookeeper對(duì)象的watcher）。

當(dāng)ZooKeeper.exists()在server上執(zhí)行完成。ZooKeeper API將在客戶端發(fā)起這個(gè)完成回調(diào)

public void processResult(int rc, String path, Object ctx, Stat stat) {
    boolean exists;
    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:
        // Retry errors
        zk.exists(znode, true, this, null);
        return;
    } 
    byte b[] = null;
    if (exists) {
        try {
            b = zk.getData(znode, false, null);
        } catch (KeeperException e) {
            // We don't need to worry about recovering now. The watch
            // callbacks will kick off any exception handling
            e.printStackTrace();
        } catch (InterruptedException e) {
            return;
        }
    }     
    if ((b == null && b != prevData)
            || (b != null && !Arrays.equals(prevData, b))) {
        listener.exists(b);
        prevData = b;
    }
}

首先檢查了znode存在返回的錯(cuò)誤代碼，致命的錯(cuò)誤及可恢復(fù)的錯(cuò)誤。如果znode存在，將從znode取得數(shù)據(jù)，如果狀態(tài)發(fā)生改變，調(diào)用Executor的exists回調(diào)。不需要為getData做任何異常處理。因?yàn)樗鼮槿魏慰赡芤l(fā)錯(cuò)誤的情況設(shè)置了監(jiān)控：如果在調(diào)用ZooKeeper.getData()前，node被刪除了，通過ZooKeeper.exists設(shè)置的監(jiān)聽事件被觸發(fā)回調(diào)；如果發(fā)生了通信錯(cuò)誤，當(dāng)連接恢復(fù)時(shí)，連接的監(jiān)聽事件被觸發(fā)。

最后，看一下DataMonitor是如何處理監(jiān)聽事件的：

public void process(WatchedEvent event) {
        String path = event.getPath();
        if (event.getType() == Event.EventType.None) {
            // We are are being told that the state of the
            // connection has changed
            switch (event.getState()) {
            case SyncConnected:
                // In this particular example we don't need to do anything
                // here - watches are automatically re-registered with 
                // server and any watches triggered while the client was 
                // disconnected will be delivered (in order of course)
                break;
            case Expired:
                // It's all over
                dead = true;
                listener.closing(KeeperException.Code.SessionExpired);
                break;
            }
        } else {
            if (path != null && path.equals(znode)) {
                // Something has changed on the node, let's find out
                zk.exists(znode, true, this, null);
            }
        }
        if (chainedWatcher != null) {
            chainedWatcher.process(event);
        }
    }

在session過期前，如果客戶端zookeeper類庫(kù)能重新發(fā)布和zookeeper的連接通道（SyncConnected event），session的所有watch將會(huì)重新發(fā)布。（zookeeper 3.0.0開始）。學(xué)習(xí)開發(fā)手冊(cè)中的ZooKeeper Watches。繼續(xù)往下講，當(dāng)DataMonitor從znode收到事件，他將會(huì)調(diào)用zookeeper.exists()，來找出發(fā)生了什么變化。

完整代碼清單

Executor.java

/**
 * A simple example program to use DataMonitor to start and
 * stop executables based on a znode. The program watches the
 * specified znode and saves the data that corresponds to the
 * znode in the filesystem. It also starts the specified program
 * with the specified arguments when the znode exists and kills
 * the program if the znode goes away.
 */
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
public class Executor
    implements Watcher, Runnable, DataMonitor.DataMonitorListener
{
    String znode;
    DataMonitor dm;
    ZooKeeper zk;
    String filename;
    String exec[];
    Process child;
 
    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);
    }
 
    /**
     * @param args
     */
    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 {
            new Executor(hostPort, znode, filename, exec).run();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /***************************************************************************
     * We do process any events ourselves, we just need to forward them on.
     *
     * @see org.apache.zookeeper.Watcher#process(org.apache.zookeeper.proto.WatcherEvent)
     */
    public void process(WatchedEvent event) {
        dm.process(event);
    }
     public void run() {
        try {
            synchronized (this) {
                while (!dm.dead) {
                    wait();
                }
            }
        } catch (InterruptedException e) {
        }
    }
 
    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 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();
            }
        }
    }
}

DataMonitor.java

/**
 * A simple class that monitors the data and existence of a ZooKeeper
 * node. It uses asynchronous ZooKeeper APIs.
 */
import java.util.Arrays;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.Stat; 
public class DataMonitor implements Watcher, StatCallback {
    ZooKeeper zk;
    String znode; 
    Watcher chainedWatcher;
    boolean dead;
    DataMonitorListener listener;
    byte prevData[];
    
    public DataMonitor(ZooKeeper zk, String znode, Watcher chainedWatcher,
            DataMonitorListener listener) {
        this.zk = zk;
        this.znode = znode;
        this.chainedWatcher = chainedWatcher;
        this.listener = listener;
        // Get things started by checking if the node exists. We are going
        // to be completely event driven
        zk.exists(znode, true, this, null);
    }
 
    /**
     * Other classes use the DataMonitor by implementing this method
     */
    public interface DataMonitorListener {
        /**
         * The existence status of the node has changed.
         */
        void exists(byte data[]);
 
        /**
         * The ZooKeeper session is no longer valid.
         *
         * @param rc
         *                the ZooKeeper reason code
         */
        void closing(int rc);
    }
    public void process(WatchedEvent event) {
        String path = event.getPath();
        if (event.getType() == Event.EventType.None) {
            // We are are being told that the state of the
            // connection has changed
            switch (event.getState()) {
            case SyncConnected:
                // In this particular example we don't need to do anything
                // here - watches are automatically re-registered with 
                // server and any watches triggered while the client was 
                // disconnected will be delivered (in order of course)
                break;
            case Expired:
                // It's all over
                dead = true;
                listener.closing(KeeperException.Code.SessionExpired);
                break;
            }
        } else {
            if (path != null && path.equals(znode)) {
                // Something has changed on the node, let's find out
                zk.exists(znode, true, this, null);
            }
        }
        if (chainedWatcher != null) {
            chainedWatcher.process(event);
        }
    }
    public void processResult(int rc, String path, Object ctx, Stat stat) {
        boolean exists;
        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:
            // Retry errors
            zk.exists(znode, true, this, null);
            return;
        }
        byte b[] = null;
        if (exists) {
            try {
                b = zk.getData(znode, false, null);
            } catch (KeeperException e) {
                // We don't need to worry about recovering now. The watch
                // callbacks will kick off any exception handling
                e.printStackTrace();
            } catch (InterruptedException e) {
                return;
            }
        }
        if ((b == null && b != prevData)
                || (b != null && !Arrays.equals(prevData, b))) {
            listener.exists(b);
            prevData = b;
        }
    }
}

以上就是Java客戶端開發(fā)案例ZooKeeper官方文檔翻譯的詳細(xì)內(nèi)容，更多關(guān)于java開發(fā)案例ooKeeper文檔翻譯的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: