apache zookeeper使用方法實(shí)例詳解

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

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

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

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

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

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

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

/**
 * 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();
      }
    }
  }
}

/**
 * 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;
    }
  }
}