ZOOKEEPER(三):java使用zookeeper

通过创建一个zookeeper实例来链接zookeeper服务器。

注意:zookeeper客户端和服务器端会话的建立是一个异步的过程,也就是说在程序中,我们程序方法在处理完客户端初始化后,立即返回(程序往下执行代码,这样,大多数情况下我们并没有真正构建好一个可用会话,在会话的声明周期处于"CONNECTING"时才算真正建立完毕,所以我们需要使用多线程中的一个工具类CountDownLatch) 

构造函数如下所示:

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, 
boolean canBeReadOnly, HostProvider aHostProvider, ZKClientConfig clientConfig)
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, long sessionId,
     byte[] sessionPasswd, boolean canBeReadOnly, HostProvider aHostProvider,
         ZKClientConfig clientConfig)

监听一个事件

public class CreateSession implements Watcher { 

    private static ZooKeeper zookeeper;
    public static void main(String[] args) throws IOException, InterruptedException {
        zookeeper = new ZooKeeper("127.0.0.0:2181",5000,new CreateSession());
        System.out.println(zookeeper.getState());//状态
        Thread.sleep(Integer.MAX_VALUE);
    }
    
    private void doSomething(){
        System.out.println("业务处理。。。");
    }
    
    @Override
    public void process(WatchedEvent event) {
        System.out.println("event = "+event);
        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething();
            }
        }
    }
    
}

创建同步节点

public class CreateNodeSync implements Watcher { 

    private static ZooKeeper zookeeper;
    
    public static void main(String[] args) throws IOException, InterruptedException {
        zookeeper = new ZooKeeper("192.168.1.105:2181",5000,new CreateNodeSync());
        System.out.println(zookeeper.getState());
        
        Thread.sleep(Integer.MAX_VALUE);
    }
    
    private void doSomething(){
        try {
String path
= zookeeper.create("/node_4", "张三".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println("return path:"+path); } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("业务处理。。。"); } @Override public void process(WatchedEvent event) { System.out.println("event="+event); if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(); } } } }

其中 ,zookeeper.create() 方法有四个参数:

       参数1,节点路径(名称):/nodeName (不允许递归创建节点,也就是说在父节点不存在的情况下,不允许创建子节点)

  参数2,节点内容:要求类型是字节数组 (不支持序列化方式,如果需要实现序列化,可使用java相关的序列化框架,如Hession,Kryo框架)

  参数3,节点权限:使用Ids.OPEN_ACL_UNSAFE开发权限即可(一般在权限没有太高要求的场景下,没必要关注),也可以使用如下方法

ACL aclIp = new ACL(Perms.READ,new Id("ip","127.0.0.0"));
ACL aclDigest = new ACL(Perms.READ|Perms.WRITE,new Id("user1",DigestAuthenticationProvider.generateDigest("user1:123456")));
ArrayList<ACL> acls = new ArrayList<ACL>();
acls.add(aclDigest);
acls.add(aclIp);

  参数4,节点类型 创建节点的类型:CreateMode.* 提供四种节点类型

    PERSISTENT (持久节点)

    PERSISTENT_SEQUENTIAL(持久顺序节点)

    EPHEMRAL(临时节点)

    EPHEMRAL_SEQUENTIAL(临时顺序节点)

异步创建节点

public class CreateNodeASync implements Watcher { 
    private static ZooKeeper zookeeper;
    public static void main(String[] args) throws IOException, InterruptedException {
        zookeeper = new ZooKeeper("127.0.0.0:2181",5000,new CreateNodeASync());
        System.out.println(zookeeper.getState());
        Thread.sleep(Integer.MAX_VALUE);
    }
    
    private void doSomething(){
        // 异步时,无返回值,IStringCallback 在这里是异步回调接口
        zookeeper.create("/node_5", "张三".getBytes(), Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT,new IStringCallback(),"AAAAA");
    }
    
    @Override
    public void process(WatchedEvent event) {
        System.out.println("event="+event);
        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething();
            }
        }
    }
    
    static class IStringCallback implements AsyncCallback.StringCallback{

        @Override
        public void processResult(int rc, String path, Object ctx, String name) {
            //rc:为服务端响应码, 0表示调用成功,-4表示端口连接,-110表示指定节点存在,-112表示会话已经过期
            //path:接口调用时传入API的数据节点的路径参数
            //ctx:为调用接口传入API的ctx的值
            //name:实际在服务器端创建节点的名称,有序时,会在节点名字后面加类似 000001
 } } }

检测节点是都存在

    exists方法

    参数1 path:路径

    参数2 watcher :注册的watcher对象。一旦之后节点内容有变更,则会向客户端发送通知,该参数允许为null (用于三类事件监听:节点的创建,删除,更新)

    参数3 watch :是否使用watcher,如果true 则使用默认上文中的watcher,false则不使用watcher

    参数4 cb:回调函数

    参数5 ctx:用于传递的下文信息对象

  注意:exists 方法意义在与无论节点是都存在,都可以进行注册watcher,能够对节点的创建,删除和修改进行监听,但是其子节点发生各种变化,都不会通知客户端。

同步判断节点是否存在

public class NodeExistsSync implements Watcher{
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new NodeExistsSync());
        System.out.println(zooKeeper.getState().toString());
        Thread.sleep(Integer.MAX_VALUE);
    }
    
    private void doSomething(ZooKeeper zooKeeper){
        try{
            Stat stat = zooKeeper.exists("/node_1", true);    
            System.out.println(stat);  
        }catch(Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething(zooKeeper);
            }else{
                try {
                    if (event.getType()==EventType.NodeCreated){
                        System.out.println(event.getPath()+" created");
                        System.out.println(zooKeeper.exists(event.getPath(), true));
                    }
                    else if (event.getType()==EventType.NodeDataChanged){
                        System.out.println(event.getPath()+" updated");
                        System.out.println(zooKeeper.exists(event.getPath(), true));
                    }
                    else if (event.getType()==EventType.NodeDeleted){
                        System.out.println(event.getPath()+" deleted");
                        System.out.println(zooKeeper.exists(event.getPath(), true));
                    }
                    
                } catch (Exception e) {
                    e.printStackTrace();
                }
                
            }
        
        }
    }

}

异步判断节点是否存在

public class NodeExistsASync implements Watcher{
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {   
        zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new NodeExistsASync());
        System.out.println(zooKeeper.getState().toString());         
        Thread.sleep(Integer.MAX_VALUE);
    }
    
    private void doSomething(ZooKeeper zookeeper){
        zooKeeper.exists("/node_1", true, new IStateCallback(), null);
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething(zooKeeper);
            }else{
                try {
                    if (event.getType()==EventType.NodeCreated){
                        System.out.println(event.getPath()+" created");
                        zooKeeper.exists(event.getPath(), true, new IStateCallback(), null);
                    }
                    else if (event.getType()==EventType.NodeDataChanged){
                        System.out.println(event.getPath()+" updated");
                        zooKeeper.exists(event.getPath(), true, new IStateCallback(), null);
                    }
                    else if (event.getType()==EventType.NodeDeleted){
                        System.out.println(event.getPath()+" deleted");
                        zooKeeper.exists(event.getPath(), true, new IStateCallback(), null);
                    }
                    
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
    
    static class  IStateCallback implements AsyncCallback.StatCallback{

        @Override
        public void processResult(int rc, String path, Object ctx, Stat stat) {
            System.out.println("rc:"+rc); 
        }
    }

}

删除节点

delete方法(api提供了两个接口,同步删除和异步删除方式)

    同步方式:

      参数1,节点名称 /deletePath

      参数2,版本号,即表示本次删除操作是针对该数据的某个版本进行操作。

    异步方式:

      参数3:一个异步回调函数

      参数4:用于传递上下文信息的对象。

    注意:在zookeeper中,只允许删除叶子节点信息,也就是说如果当前节点不是叶子节点则无法删除,或必须先删除其下所有的子节点,

同步删除节点

public class DeleteNodeSync implements Watcher{  
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws IOException, InterruptedException, 
                KeeperException {
        zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new DeleteNodeSync());
        System.out.println(zooKeeper.getState().toString());           
        Thread.sleep(Integer.MAX_VALUE);
    }
    
    private void doSomething(ZooKeeper zooKeeper){
        try {
            zooKeeper.delete("/node_7", -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething(zooKeeper);
            }
        }
    }

}

异步删除节点

public class DeleteNodeASync implements Watcher{
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws IOException, InterruptedException,  KeeperException {
        zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new DeleteNodeASync());
        System.out.println(zooKeeper.getState().toString());
        Thread.sleep(Integer.MAX_VALUE);
    }
    
    private void doSomething(WatchedEvent event){

        zooKeeper.delete("/node_6", -1, new IVoidCallback(),null);
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething(event);
            }
        }
    }
    
    static class IVoidCallback implements AsyncCallback.VoidCallback{

        @Override
        public void processResult(int rc, String path, Object ctx) {
            StringBuilder sb = new StringBuilder();
            sb.append("rc="+rc).append("
");
            sb.append("path"+path).append("
");
            sb.append("ctx="+ctx).append("
");
            System.out.println(sb.toString());
        }        
    }
}

同步修改节点

public class UpdateNodeSync implements Watcher{
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws IOException, InterruptedException, 
            KeeperException {
        zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new UpdateNodeSync());
        System.out.println(zooKeeper.getState().toString());      
        Thread.sleep(Integer.MAX_VALUE);
    }
    
    private void doSomething(ZooKeeper zooKeeper){
        try {
            Stat stat = zooKeeper.setData("/node_6", "123".getBytes(), -1);
            System.out.println("stat:"+stat);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void process(WatchedEvent event) {

        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething(zooKeeper);
            }
        }
    }

}

异步修改节点

public class UpdateNodeASync implements Watcher{
    
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new UpdateNodeASync());
        System.out.println(zooKeeper.getState().toString());
        
        Thread.sleep(Integer.MAX_VALUE);
        
    }
    
    private void doSomething(WatchedEvent event){

        zooKeeper.setData("/node_6", "234".getBytes(), -1, new IStatCallback(),null);
    
    }

    @Override
    public void process(WatchedEvent event) {
        
        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething(event);
            }
        }
    }
    
    static class IStatCallback implements AsyncCallback.StatCallback{

        @Override
        public void processResult(int rc, String path, Object ctx, Stat stat) {
            StringBuilder sb = new StringBuilder();
            sb.append("rc="+rc).append("
");
            sb.append("path"+path).append("
");
            sb.append("ctx="+ctx).append("
");
            sb.append("Stat="+stat).append("
");
            System.out.println(sb.toString()); 
        }        
    }

}

同步获取权限

public class GetDataSyncAuth implements Watcher{  
    private static ZooKeeper zooKeeper;
    private static Stat stat = new Stat();
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {     
        zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new GetDataSyncAuth());
        System.out.println(zooKeeper.getState().toString());  
        Thread.sleep(Integer.MAX_VALUE);      
    }
    
    private void doSomething(ZooKeeper zookeeper){
        zooKeeper.addAuthInfo("user2", "user2:1234".getBytes());
        try {
            System.out.println(new String(zooKeeper.getData("/node_4", true, stat)));
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } 
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething(zooKeeper);
            }else{                
                if (event.getType()==EventType.NodeDataChanged){
                    try {
                        System.out.println(new String(zooKeeper.getData(event.getPath(), true, stat)));
                        System.out.println("stat:"+stat);
                    } catch (KeeperException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }    
            }
        }
    }

}

同步获取节点数据

public class GetDataSync implements Watcher{
    private static ZooKeeper zooKeeper;
    private static Stat stat = new Stat();
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {  
        zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new GetDataSync());
        System.out.println(zooKeeper.getState().toString());  
        Thread.sleep(Integer.MAX_VALUE);
    }
    
    private void doSomething(ZooKeeper zookeeper){         
        zookeeper.addAuthInfo("user2", "user2:123456".getBytes());
            try {
                System.out.println(new String(zooKeeper.getData("/node_4", true, stat)));
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething(zooKeeper);
            }else{                
                if (event.getType()==EventType.NodeDataChanged){
                    try {
                        System.out.println(new String(zooKeeper.getData(event.getPath(), true, stat)));
                        System.out.println("stat:"+stat);
                    } catch (KeeperException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }               
                }    
            }
        }
    }
}

异步获取节点数据

public class GetDataASync implements Watcher{
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws IOException, InterruptedException,KeeperException {
        zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new GetDataASync());
        System.out.println(zooKeeper.getState().toString());    
        Thread.sleep(Integer.MAX_VALUE);  
    }
    
    private void doSomething(ZooKeeper zookeeper){
        zooKeeper.getData("/node_1", true, new IDataCallback(), null);    
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething(zooKeeper);
            }else{
                if (event.getType()==EventType.NodeDataChanged){
                    try {
                        zooKeeper.getData(event.getPath(), true, new IDataCallback(), null);
                    } catch (Exception e) {
                        e.printStackTrace();
                    } 
                }
            }
        }
    }
    
    static class IDataCallback implements AsyncCallback.DataCallback{
        @Override
        public void processResult(int rc, String path, Object ctx, byte[] data,
                Stat stat) {
            try {
                System.out.println(new String(zooKeeper.getData(path, true, stat)));
                System.out.println("stat:"+stat);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

获取子节点列表和数据

    getChildren读取数据方法:包括子节点列表获取和子节点数据的获取。

    参数1 path:获取指定节点下的数据(获取子节点列表)

    参数2 watcher: 注册的watcher ,一旦在本次子节点获取后,子节点列表发生变更的话,那么就会向客户端发送通知,该参数允许为null

    参数3 wath:表明是否需要注册一个watcher:如果为true,则会使用到zookeeper客户端上下文中提到的那个默认watcher,如果false,则表明不需要注册Watcher,

    参数4 cb: 回调函数

    参数5 ctx:上下文信息

    参数6 stat :指定数据节点的节点状态信息

  注意:当我们获取指定节点的子节点列表后,还需要订阅这个子节点列表的变化通知,这时候就可以通过注册一个watcher来实现,当子节点被添加或删除时,服务器端就会触发一个“NodeChildrenChanged”类型的时间通知,需要注意的是服务端发送给客户端的事件通知中,是不包含最新的节点列表的,客户端必须主动重新进行获取,通常在客户端收到这个事件通知后,就可以再次主动获取最新的子节点列表,也就是说,zookeeper服务端在想客户端发送watcher“NodeChildrenChanged”事件通知的时候,仅仅只发了一个通知,不会把节点的变化情况发给客户端,需要客户端自己重新获取,另外Watcher通知是一次性的,即触发后失效,因此客户端需要反复注册Watcher才行。

同步获取子节点数据

public class GetChildrenSync implements Watcher{ 
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {         
        zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new GetChildrenSync());
        System.out.println(zooKeeper.getState().toString());
        Thread.sleep(Integer.MAX_VALUE);
    }
    
    private void doSomething(ZooKeeper zooKeeper){
        try {
            List<String> children =  zooKeeper.getChildren("/", true);
            System.out.println(children);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething(zooKeeper);
            }else{
                if (event.getType()==EventType.NodeChildrenChanged){
                    try {
                        System.out.println(zooKeeper.getChildren(event.getPath(), true));
                    } catch (KeeperException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }        
        }
    }
}

异步获取子节点数据

public class GetChildrenASync implements Watcher{
    
    private static ZooKeeper zooKeeper;
    
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        
        
        zooKeeper = new ZooKeeper("192.168.1.105:2181",5000,new GetChildrenASync());
        System.out.println(zooKeeper.getState().toString());
                
        Thread.sleep(Integer.MAX_VALUE);
        

    }
    
    private void doSomething(ZooKeeper zookeeper){
        try {
            zooKeeper.getChildren("/", true, new IChildren2Callback(), null);        
        } catch (Exception e) {
            e.printStackTrace();
        }

        
    }

    @Override
    public void process(WatchedEvent event) {

        if (event.getState()==KeeperState.SyncConnected){
            if (event.getType()==EventType.None && null==event.getPath()){
                doSomething(zooKeeper);
            }else{            
                if (event.getType()==EventType.NodeChildrenChanged){
                    zooKeeper.getChildren(event.getPath(), true, new IChildren2Callback(), null);
                }            
            }
        }
    }
    
    static class IChildren2Callback implements AsyncCallback.Children2Callback{

        @Override
        public void processResult(int rc, String path, Object ctx,
                List<String> children, Stat stat) {
            
            StringBuilder sb = new StringBuilder();
            sb.append("rc="+rc).append("
");
            sb.append("path="+path).append("
");
            sb.append("ctx="+ctx).append("
");
            sb.append("children="+children).append("
");
            sb.append("stat="+stat).append("
");
            System.out.println(sb.toString());
        }
    }

}
package bjsxt.zookeeper.base;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;

/**
 * Zookeeper base学习笔记
 */
public class ZookeeperBase {

    /** zookeeper地址 */
    static final String CONNECT_ADDR = "192.168.2.2:2181";// 集群可以写多个,英文逗号隔开
    /** session超时时间 */
    static final int SESSION_OUTTIME = 2000;// ms
    /** 信号量,阻塞程序执行,用于等待zookeeper连接成功,发送成功信号 */
    static final CountDownLatch connectedSemaphore = new CountDownLatch(1);

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

        ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME,
                new Watcher() {
                    @Override
                    public void process(WatchedEvent event) {
                        // 获取事件的状态
                        KeeperState keeperState = event.getState();
                        EventType eventType = event.getType();
                        // 如果是建立连接
                        if (KeeperState.SyncConnected == keeperState) {
                            if (EventType.None == eventType) {
                                // 如果建立连接成功,则发送信号量,让后续阻塞程序向下执行
                                System.out.println("zk 建立连接");
                                connectedSemaphore.countDown();
                            }
                        }
                    }
                });

        // 进行阻塞
        connectedSemaphore.await();

        System.out.println("..");
        // 创建父节点
        // zk.create("/testRoot", "testRoot".getBytes(), Ids.OPEN_ACL_UNSAFE,
        // CreateMode.PERSISTENT);

        // 创建子节点
        // zk.create("/testRoot/children", "children data".getBytes(),
        // Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        // 获取节点洗信息
        // byte[] data = zk.getData("/testRoot", false, null);
        // System.out.println(new String(data));
        // System.out.println(zk.getChildren("/testRoot", false));

        // 修改节点的值
        // zk.setData("/testRoot", "modify data root".getBytes(), -1);
        // byte[] data = zk.getData("/testRoot", false, null);
        // System.out.println(new String(data));

        // 判断节点是否存在
        // System.out.println(zk.exists("/testRoot/children", false));
        // 删除节点
        // zk.delete("/testRoot/children", -1);
        // System.out.println(zk.exists("/testRoot/children", false));

        zk.close();

    }

}

 参考: https://www.cnblogs.com/shamo89/p/9786323.html

原文地址:https://www.cnblogs.com/yrjns/p/12454630.html