zkClient操作zookeeper集群

1、导入依赖

   

    com.101tec

    zkclient

    0.10

2.编写序列化类（解决获取数据时候无法序列化而报错）

import org.I0Itec.zkclient.exception.ZkMarshallingError;

public class ZkSerializer implements org.I0Itec.zkclient.serialize.ZkSerializer {

        //序列化，数据--》byte[]

        public byte[] serialize(Object o) throws ZkMarshallingError {

            return String.valueOf(o).getBytes();

        }

        //反序列化，byte[]--->数据

        public Object deserialize(byte[] bytes) throws ZkMarshallingError {

            return new String(bytes);

        }

}

3、常规操作（增删查改）

import org.I0Itec.zkclient.ZkClient;

import org.I0Itec.zkclient.ZkConnection;

import org.apache.zookeeper.CreateMode;

import java.util.List;

public class ZkTest {

    //zookeeper集群地址

    public static final String  URL = "192.168.112.101:2181,192.168.112.102:2181,192.168.112.103:2181";

    //连接超时时间

    public static final int TIME_OUT = 5000;

    public static void main(String[] args) throws InterruptedException {

        ZkClient client = new ZkClient(new ZkConnection(URL),TIME_OUT);

        //设置序列化

        client.setZkSerializer(new ZkSerializer());

        //增加临时节点

        client.createEphemeral("/a1",123);

        //增加永久节点

        client.create("/a2",123, CreateMode.PERSISTENT);

        //Thread.sleep(10000);

        //删除节点

        client.delete("/a2");

        //查询节点有哪些子节点

        List children = client.getChildren("/")

 

        System.out.println(children);

        //获取节点数据

     Object data = client.readData("/a3");

        System.out.println(data);

        //修改

        client.writeData("/a3",456);

        client.close();

    }

}

4.监听操

importorg.I0Itec.zkclient.IZkChildListener

import org.I0Itec.zkclient.IZkDataListener;

import org.I0Itec.zkclient.ZkClient;

import org.I0Itec.zkclient.ZkConnection;

import org.I0Itec.zkclient.serialize.SerializableSerializer;

import org.apache.zookeeper.CreateMode;

import java.util.List;

public class ZkWatch {

  //zookeeper集群地址

   public static final String  URL = "192.168.112.101:2181,192.168.112.102:2181,192.168.112.103:2181";

//连接超时时间

 public static final int TIME_OUT = 5000;

public static void main(String[] args) throws InterruptedException {

 ZkClient client = new ZkClient(new ZkConnection(URL),TIME_OUT);

//设置序列化

client.setZkSerializer(new SerializableSerializer());

//开始监听节点变化

client.subscribeChildChanges("/", new IZkChildListener() {

@Override

public void handleChildChange(String parentPath, List currentChilds) throws Exception {

 System.out.println(parentPath);

System.out.println(currentChilds);

}

 });

//开始监听节点数据变化

 client.subscribeDataChanges("/a4", new IZkDataListener() {

 @Override

 public void handleDataChange(String dataPath, Object data) throws Exception {

System.out.println("dataPath:" + dataPath);

System.out.println("data:" + data);

 }

@Override

 public void handleDataDeleted(String dataPath) throws Exception {

System.out.println("dataPath:" + dataPath);

}

});

 Thread.sleep(2000);

new Thread(new Runnable() {

@Override