ZOOKEEPER（三）：java使用zookeeper

Z时代
2024-01-10
分类：综合

java

通过创建一个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();
            }
        }
    }
}

创建同步节点

 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("\n");
            sb.append("path"+path).append("\n");
            sb.append("ctx="+ctx).append("\n");
            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("\n");
            sb.append("path"+path).append("\n");
            sb.append("ctx="+ctx).append("\n");
            sb.append("Stat="+stat).append("\n");
            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("\n");
            sb.append("path="+path).append("\n");
            sb.append("ctx="+ctx).append("\n");
            sb.append("children="+children).append("\n");
            sb.append("stat="+stat).append("\n");
            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();
    }
}