Apache > Hadoop > ZooKeeper
 

ZooKeeper の Java での使用例

簡単なウォッチクライアント

ZooKeeper Java API を利用するための手引きとして、ここでは非常に簡単なウォッチクライアントを作成します。作成する ZooKeeper クライアントは、ZooKeeper ノードに変更があったかどうかをウォッチし、プログラムを起動または停止して応答します。

要件

クライアントの要件は次の 4 項目です。

  • 以下の引数を取ります。

    • ZooKeeper サービスのアドレス

    • ウォッチする znode の名前

    • プログラムとその引数

  • znode に関連付けられたデータを取得し、プログラムを起動します。

  • znode が変更されたら、クライアントはその内容を取得し、プログラムを再起動します。

  • znode が消滅したら、クライアントはプログラムを kill します。

プログラムの設計

慣習として、ZooKeeper アプリケーションは 2 のユニットに分けます。一方は接続を維持します。もう一方はデータを監視します。ここで取り上げるアプリケーションでは、Executor という名前のクラスが ZooKeeper 接続を管理し、DataMonitor という名前のクラスが ZooKeeper ツリー上のデータを監視します。なお、Executor には、メインスレッドと実行ロジックも収められています。Executor は、ごくわずかなユーザーとの対話を受け持つほか、引数で渡されたプログラムを znode の状態に応じて (上述の要件に従い) 停止または再起動します。

Executor クラス

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 {
            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 の役割は、コマンドラインで指定されたプログラムの起動と停止です。Executor はこの処理を、ZooKeeper オブジェクトで発生したイベントに応答する形で行います。上のコードをみるとわかるように、Executor は自身への参照を、ZooKeeper コンストラクタの Watcher 引数として渡します。また、DataMonitor コンストラクタに対しても、自身への参照を DataMonitorListener 引数として渡します。Executor の定義では、これらのインタフェースを実装しています。

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

Watcher インタフェースは、ZooKeeper Java API で定義されています。ZooKeeper は Watcher インタフェースを使って、このインタフェースのコンテナと通信します。Watcher インタフェースは process() という 1 つのメソッドだけをサポートしており、ZooKeeper はこのメソッドを使って、ZooKeeper 接続の状態や ZooKeeper セッションといった、メインスレッドの処理対象となる一般的なイベントを伝達します。サンプルアプリケーションの Executor は、こうしたイベントを DataMonitor に転送するだけで、イベントをどう処理するかは DataMonitor に任せています。これは、慣習上、Executor または類似のオブジェクトは ZooKeeper 接続を「所有」するものの、イベントの処理はほかのオブジェクトに任せることができることを示すためです。DataMonitor は、ウォッチイベントが発生するデフォルトチャンネルとしても使われています。(あとでもう少し詳しく説明します。)

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

一方、DataMonitorListener インタフェースは、ZooKeeper API には含まれていません。DataMonitorListener は、このサンプルアプリケーションのために設計された、完全に独自のインタフェースです。DataMonitor オブジェクトは DataMonitorListener インタフェースを使って、このインタフェースのコンテナと通信します。このコンテナも Executor オブジェクトです。DataMonitorListener インタフェースは、次のようになっています。

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);
}

DataMonitorListener インタフェースは DataMonitor クラスで定義されており、Executor クラスで実装されています。Executor は、Executor.exists() が呼び出されると、要件に従ってプログラムを起動すべきか停止すべきかを決定します。要件では、znode が存在しなくなったら、プログラムを kill することになっていることを思い起こしてください。

Executor は、Executor.closing() が呼び出されると、ZooKeeper 接続が永久に失われたかどうかに応じて、自身をシャットダウンすべきかどうかを決定します。

すでに想像がつくと思いますが、ZooKeeper の状態の変化に応じて、これらのメソッドを呼び出すオブジェクトは、DataMonitor です。

DataMonitorListener.exists()DataMonitorListener.closing の Executor での実装は次のようになっています。

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 ロジックの本体に相当し、ほとんどが非同期でイベント駆動です。DataMonitor は次のようなコンストラクタで動作を開始します。

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);
}

ZooKeeper.exists() の呼び出しでは、対象となる znode の存在をチェックし、ウォッチを設定し、完了コールバックオブジェクトとして、自身への参照 (this) を渡しています。その意味では、DataMonitor がサンプルアプリケーションの動作の出発点となります。実際の処理はウォッチがトリガされたときに行われるからです。

完了コールバックとウォッチコールバックを混同しないようにしてください。ZooKeeper.exists() 完了コールバックは、このサンプルアプリケーションでは DataMonitor オブジェクトで実装されているメソッド StatCallback.processResult() ですが、このコールバックは、 (ZooKeeper.exists() による) 非同期でのウォッチの設定操作がサーバーで完了したときに呼び出されます。

一方、ウォッチがトリガされると、Executor オブジェクトにイベントが送信されます。これは、Executor が ZooKeeper オブジェクトの Watcher として登録されているからです。

余談になりますが、DataMonitor も、特定のウォッチイベントの Watcher として自身を登録できることに読者は気付いているかもしれません。これは、ZooKeeper 3.0.0 の新機能です (複数の Watcher のサポート)。ただし、ここで取り上げるサンプルアプリケーションでは、DataMonitor を Watcher として登録することはしていません。

サーバー上で ZooKeeper.exists() 操作が完了すると、ZooKeeper API はクライアント上で次のコールバックを呼び出します。

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 の存在、致命的なエラー、および回復可能なエラーのどれに該当するかを調べます。ファイル (すなわち znode) が存在する場合には、znode からデータを取得し、状態が変更されている場合には exists() コールバックを呼び出します。getData は、エラーを引き起こす可能性のあるすべてのものを対象にウォッチを保留中にしているので、getData の呼び出しでは Exception の処理を行う必要がないことに注意してください。ZooKeeper.getData() を呼び出す前にノードが削除されている場合、ZooKeeper.exists() によって設定されたウォッチイベントはコールバックをトリガします。通信エラーが存在する場合は、接続が回復したときに、接続ウォッチイベントが発生します。

次に示すのは DataMonitor がウォッチイベントを処理する部分です。

    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);
        }
    }

クライアント側の ZooKeeper ライブラリが、セッションの有効期限が切れる (Expired イベント) 前に、ZooKeeper への通信チャンネルを再確立できる (SyncConnected イベント) 場合には、セッションのすべてのウォッチがサーバーとの間で自動的に再確立されます (ウォッチの自動再設定は ZooKeeper 3.0.0 の新機能です)。詳細については、プログラマーズガイドの「ZooKeeper のウォッチ」を参照してください。上の関数のやや後半よりの部分では、DataMonitor が znode のイベントを受け取ったときに、何が変更されたかを調べるために ZooKeeper.exists() を呼び出しています。

全ソースリスト

Executor.java
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
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;
        }
    }
}