zookeeperJavaapi-创新互联

maven依赖：

  4.0.0
  per.ym
  zk
  0.0.1-SNAPSHOT

  
    
      org.apache.zookeeper
      zookeeper
      3.4.12
    

    
        junit
        junit
        4.12
        test
    
  

测试类：

package per.ym.zookeeper;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.AsyncCallback.ChildrenCallback;
import org.apache.zookeeper.AsyncCallback.DataCallback;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.AsyncCallback.StringCallback;
import org.apache.zookeeper.AsyncCallback.VoidCallback;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

public class ZkTest implements Watcher{

    private ZooKeeper zk;

    private CountDownLatch cdl;

    private String path = "/test";

    private String rootPath = "/";

    private int sessionTimeOut = 15000;

    private byte[] data = "data".getBytes();

    private byte[] newData = "newData".getBytes();

    @Before
    public void connect() throws IOException {
        zk = new ZooKeeper("192.168.61.131:2184", sessionTimeOut, this);
        cdl = new CountDownLatch(1);
    }

    //同步调用
    @Test
    public void testSync( ) throws Exception {
        //等待与zookeeper服务端连接完成
        cdl.await();

        //创建一个持久节点/test，并为其赋值为data
        zk.create("/test", data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        //获取根节点下的子节点
        List children = zk.getChildren("/", this);
        System.out.println("根节点下的子节点有: " + children);

        //查询节点/test保存的数据
        Stat stat = new Stat();
        byte[] tempData = zk.getData(path, this, stat);
        System.out.println("test节点数据为: " + new String(tempData));

        //设置节点/test的数据为newData
        zk.setData(path, newData, -1);
        tempData = zk.getData(path, this, stat);
        System.out.println("test节点新数据为: " + new String(tempData));

        //删除节点/test
        zk.delete(path, -1);

        //判断节点/test是否存在
        stat = zk.exists(path, this);
        System.out.println(stat);
    }

    //创建节点后进行回调
    private StringCallback cb = new StringCallback() {

        @Override
        public void proce***esult(int rc, String path, Object ctx, String name) {
            switch (Code.get(rc)) { 
                case CONNECTIONLOSS:
                    //如果失去连接，我们无法保证是在创建前还是创建后丢失的，因此重试
                    create();
                    break;
                case OK:
                    System.out.println("节点/test创建成功");
                    break;
                case NODEEXISTS:
                    System.out.println("节点/test已经存在");
                    break;
                default:
                    System.out.println(KeeperException.create(Code.get(rc), path));
                }

        }
    };

    //获取子节点进行回调
    private ChildrenCallback ccb = new ChildrenCallback() {

        @Override
        public void proce***esult(int rc, String path, Object ctx, List children) {
            switch (Code.get(rc)) {
                case CONNECTIONLOSS:
                    getChildren();
                    break;
                case OK:
                    System.out.println("根节点下的子节点有: " + children);
                    break;
                default:
                    System.out.println(KeeperException.create(Code.get(rc), path));
            }
        }

    };

    //获取数据进行回调
    private DataCallback dcb = new DataCallback() {

        @Override
        public void proce***esult(int rc, String path, Object ctx, byte[] data, Stat stat) {
            switch (Code.get(rc)) {
                case CONNECTIONLOSS:
                    getData();
                    break;
                case OK:
                    System.out.println("test节点数据为: " + new String(data));
                    break;
                default:
                    System.out.println(KeeperException.create(Code.get(rc), path));
            }
        }

    };

    //设置数据进行回调
    private StatCallback scb = new StatCallback() {

        @Override
        public void proce***esult(int rc, String path, Object ctx, Stat stat) {
            switch (Code.get(rc)) {
                case CONNECTIONLOSS:
                    setData();
                    break;
                case OK:
                    //这个ctx就是我们调用zk.setData时传入的最后一个参数
                    System.out.println("test节点设置新数 " + new String((byte[])ctx) + "成功");
                    break;
                default:
                    System.out.println(KeeperException.create(Code.get(rc), path));
            }
        }

    };

    //删除节点进行回调
    private VoidCallback vcb = new VoidCallback() {

        @Override
        public void proce***esult(int rc, String path, Object ctx) {
            switch (Code.get(rc)) {
                case CONNECTIONLOSS:
                    delete();
                    break;
                case OK:
                    System.out.println("删除节点/test成功");
                    break;
                default:
                    System.out.println(KeeperException.create(Code.get(rc), path));
            }
        }

    };

    //节点是否存在进行回调
    private StatCallback scb2 = new StatCallback() {

        @Override
        public void proce***esult(int rc, String path, Object ctx, Stat stat) {
            switch (Code.get(rc)) { 
                case CONNECTIONLOSS:
                    exists();
                    break;
                case OK:
                    System.out.println("节点/test存在");
                    break;
                case NONODE:
                    System.out.println("节点/test不存在");
                    break;
                default:
                    break;
            }
        }

    };

    //异步调用
    @Test
    public void testAsync() throws Exception {
        cdl.await();

        create();

        getChildren();

        setData();

        getData();

        exists();

        delete();

        exists();
    }

    private void create() {
        zk.create(path, data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, cb, data);
    }

    private void getChildren() {
        zk.getChildren(rootPath, this, ccb, null);
    }

    private void getData() {
        zk.getData(path, this, dcb, null);
    }

    private void setData() {
        zk.setData(path, newData, -1, scb, newData);
    }

    private void delete() {
        zk.delete(path, -1, vcb, null);
    }

    private void exists() {
        zk.exists(path, this, scb2, null);
    }

    @Override
    public void process(WatchedEvent event) {
        //如果连接成功，放行
        if (event.getType().equals(EventType.None)) {
            cdl.countDown();
        }

        System.out.println(event);
    }

    @After
    public void close() throws InterruptedException {
        //测试完成后关闭连接
        zk.close();
    }
}

另外有需要云服务器可以了解下创新互联scvps.cn，海内外云服务器15元起步，三天无理由+7*72小时售后在线，公司持有idc许可证，提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案，具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势，专为企业上云打造定制，能够满足用户丰富、多元化的应用场景需求。