zookeeper的java操作方式详解

雨点打透心脏的1/2处 2022-06-17 11:49 195阅读 0赞

1.创建会话:

//zk创建会话
    public class ZkSessionDemo1 implements Watcher{
    	
    	private static CountDownLatch latch = new CountDownLatch(1); //计数器为1
    	
    
    	@Override
    	public void process(WatchedEvent event) {
    		System.out.println("watch event触发");
    		if(KeeperState.SyncConnected == event.getState()) {
    			latch.countDown();//闭锁计数为0
    		}
    	}
    	
    	public static void main(String[] args) throws IOException {
    		ZooKeeper zk = new ZooKeeper("23.105.195.9:2181", 5000, new ZkSessionDemo1());
    		System.out.println(zk.getState());
    		
    		try {
    			latch.await(); //闭锁在计数器为0前等待
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		System.out.println("zk session已经创建好");
    	}
    
    }

2.创建一个复用sessionId和sessionPasswd的ZooKeeper会话实例

import java.util.concurrent.CountDownLatch;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    
    //创建一个复用sessionId和sessionPasswd的ZooKeeper实例
    public class ZkSessionDemo2 implements Watcher {
    	
    private static CountDownLatch latch = new CountDownLatch(1); //计数器为1
    	
    
    	@Override
    	public void process(WatchedEvent event) {
    		System.out.println("watch event触发:" + event);
    		if(KeeperState.SyncConnected == event.getState()) {
    			latch.countDown();//闭锁计数为0
    		}
    	}
    	
    	public static void main(String[] args) throws Exception {
    		ZooKeeper zk = new ZooKeeper("23.105.195.9:2181", 5000, new ZkSessionDemo1());
    		latch.await(); //闭锁在计数器为0前等待
    		long sessionId = zk.getSessionId();
    		byte[] passwd = zk.getSessionPasswd();
    		
    		//使用错误的sessionPwd
    		zk = new ZooKeeper("23.105.195.9:2181", 5000, new ZkSessionDemo1(), sessionId, "test".getBytes());
    		
    		//使用正确的sessionPwd
    		zk = new ZooKeeper("23.105.195.9:2181", 5000, new ZkSessionDemo1(), sessionId, passwd);
    		
    		Thread.sleep(Integer.MAX_VALUE);
    	}
    	
    }

3.同步创建节点

import java.io.IOException;
    import java.util.concurrent.CountDownLatch;
    
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.ZooDefs.Ids;
    
    //同步创建接口
    public class ZkCreateDemo implements Watcher {
    	
    	private static CountDownLatch latch = new CountDownLatch(1); //计数器为1
    	
    	@Override
    	public void process(WatchedEvent event) {
    		System.out.println("watch event触发");
    		if(KeeperState.SyncConnected == event.getState()) {
    			System.out.println("准备好");
    			latch.countDown();//闭锁计数为0
    		}
    	}
    	
    	public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
    		ZooKeeper zk = new ZooKeeper("23.105.195.9:2181", 5000, new ZkCreateDemo());
    		System.out.println(zk.getState());
    		
    		latch.await(); //闭锁在计数器为0前等待
    		
    		//Ids.OPEN_ACL_UNSAFE任何节点都不受权限控制
    		String path1 = zk.create("/zk-test-ephemeral3-", "".getBytes(), 
    				Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); //临时节点
    		System.out.println("成功创建znode：" + path1);
    		
    		String path2 = zk.create("/zk-test-ephemeral3-", "".getBytes(), 
    				Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); //临时顺序节点
    		System.out.println("成功创建znode：" + path2);
    		System.out.println("zk session已经创建好");
    	}
    	
    }

4.异步创建节点

import java.io.IOException;
    import java.util.concurrent.CountDownLatch;
    
    import org.apache.zookeeper.AsyncCallback;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.ZooDefs.Ids;
    
    public class ZkAsyncCreateDemo implements Watcher {
    	
    private static CountDownLatch latch = new CountDownLatch(1); //计数器为1
    	
    	@Override
    	public void process(WatchedEvent event) {
    		System.out.println("watch event触发");
    		if(KeeperState.SyncConnected == event.getState()) {
    			System.out.println("准备好");
    			latch.countDown();//闭锁计数为0
    		}
    	}
    	
    	public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
    		ZooKeeper zk = new ZooKeeper("23.105.195.9:2181", 5000, new ZkAsyncCreateDemo());
    		
    		latch.await(); //闭锁在计数器为0前等待
    		
    		//Ids.OPEN_ACL_UNSAFE任何节点都不受权限控制,异步已经没有返回值了
    		zk.create("/zk-test-ephemeral6-", "".getBytes(), 
    				Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
    				new ISStringCallback(), "i am context"); //临时节点
    		
    		zk.create("/zk-test-ephemeral6-", "".getBytes(), 
    				Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,
    				new ISStringCallback(), "i am context"); //临时顺序节点
    		
    		Thread.sleep(Integer.MAX_VALUE);
    	}
    	
    	
    }
    
    class ISStringCallback implements AsyncCallback.StringCallback {
    	
    	@Override
    	public void processResult(int rc, String path, Object ctx, String name) {
    		System.out.println("rc=" + rc + "; path=" + path + "; ctx=" + ctx + "; name=" + name);
    	}
    	
    }

使用异步接口，用户仅仅需要实现AsyncCallback.StringCallback接口即可。  
AsyncCallback包含StatCallback, DataCallBack, ACLCallback,  
ChildrenCallback,Children2Callback,StringCallback和VoidCallback  
七种不同的回调接口，用户可以在不同的异步接口中实现不同的接口。  
和同步接口方法最大的区别在于，节点的创建过程(包括网络通信和服务端的节点创建过程)是  
异步的。并且，在同步接口调用过程中，我们需要关注接口抛出异常的可能；但是在异步接口中，  
接口本身是不会抛出异常的，所有异常都会在回调函数中通过Result Code(响应码)来体现  
  
  
来看看void processResult(int rc, String path, Object ctx, String name)的几个参数如表

![Center][]

5.删除节点

//删除节点
    public class ZkDeleteDemo implements Watcher {
    
    	private static CountDownLatch latch = new CountDownLatch(1);
    	private static ZooKeeper zk;
    
    	public static void main(String[] args) throws Exception {
    
    		String path = "/zk-book";
    		zk = new ZooKeeper("23.105.195.9:2181", 5000, //
    				new ZkDeleteDemo());
    		latch.await();
    
    		zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    		zk.delete(path, -1);
    
    		Thread.sleep(Integer.MAX_VALUE);
    	}
    
    	@Override
    	public void process(WatchedEvent event) {
    		if (KeeperState.SyncConnected == event.getState()) {
    			if (EventType.None == event.getType() && null == event.getPath()) {
    				latch.countDown();
    			}
    		}
    	}
    
    }

删除api常用方法:

public void delete(final String path, int version)  
pulbic void delete(final String path, int version, VoidCallback cb, Object ctx)

![Center 1][]  
  
  
6.查询节点

客户端可以通过ZooKeeper的api来获取一个节点所有子节点，有如下8个方法可以使用:  
 List<String> getChildren(final String path, Watcher watcher)  
 List<String> getChildren(String path, boolean watch)  
 void getChildren(final String path, Watcher watcher, ChildrenCallback cb, Object ctx)  
 void getChildren(String path,boolean watch, Stat stat)  
 List<String> getChildren(final String path, Watcher watcher, Stat stat)  
 List<String> getChildren(String path, boolean watch, Stat stat)  
 void getChildren(final String path, Watcher watcher, Children2Callback cb, Object ctx)  
 void getChildren(String path, boolean watch, Children2Callback cb, Object ctx)  
   
 参数解释:  
 \-----------------------------|------------------------------------------------------------------------------------------------------------------------  
 参数名                 |                                                说明

\------------------------------|-----------------------------------------------------------------------------------------------------------------------  
 path    |指定数据节点的节点路径，即api调用的目的是获取该节点的子节点列表

|                     客户端发送通知。该参数允许传入null

\-----------------------------|\------------------------------------------------------------------------------------------------------------------------  
 stat                             |                  指定数据节点的状态信息。用法是在接口中传入一个旧的stat变量，该变量会在方                                      |                 法执行过程中，被来自服务端响应的新stat对象替换。

\------------------------------|\-----------------------------------------------------------------------------------------------------------------------  
      
     
首先，我们来看看注册watcher，如果zookeeper客户端在获取到指定节点的子节点列表后，还需要订阅这个子节点列表的变化通知，  
那么就可以通过注册一个watcher来实现。当有子节点被添加或是删除时，服务端就会向客户端发送一个NodeChildrenChanged（EventType.NodeChildrenChanged）  
类型的事件通知。需要注意的是，在服务端发送给客户端的事件通知中，是不包含最小的节点列表的，客户端必须主动重新进行获取。通常客户端在  
收到这个事件通知后，就可以再次获取最新子节点的列表。  
再来看看用于描述节点状态信息的对象:stat。stat对象中记录了一个节点的基本属性信息，例如节点创建时的事务ID(cZxiid),最后一次修改的  
事务ID(mZxid)和节点的数据内容的长度(dataLength)等。有时候，我们不仅需要获取节点的最新子节点列表，还要获取这个节点最新的节点状态  
信息。对于这种情况，我们可以将一个旧的stat变量传入api接口，该stat变量会在方法执行过程中，被来自服务端响应的新stat对象替换

import java.util.List;
    import java.util.concurrent.CountDownLatch;
    
    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.Watcher.Event.KeeperState;
    import org.apache.zookeeper.ZooDefs.Ids;
    import org.apache.zookeeper.ZooKeeper;
    
    //使用同步api获取子节点列表
    public class ZkReadDemo1 implements Watcher {
    	
    	private static CountDownLatch latch = new CountDownLatch(1);
    	private static ZooKeeper zk = null;
    
    	//关于watcher，zookeeper服务端在向客户端发送watcher的“NodeChildrenChanged”事件
    	//通知的时候，仅仅只会发出一个通知，而不会把节点的变化情况发送给客户端，需要客户端重新获取。
    	//另外，由于watcher通知是一次性的，即一旦触发一次通知后，该watcher就失效了，因此客户端
    	//需要反复注册watcher
    	@Override
    	public void process(WatchedEvent event) {
    		if(KeeperState.SyncConnected == event.getState()) {
    			if(EventType.None == event.getType() && null == event.getPath()) {
    				latch.countDown();
    			} else if(event.getType() == EventType.NodeChildrenChanged) {
    				try {
    					//我们可以看到，获取的都是子节点的相对路径
    					System.out.println("获取子节点数据:" + zk.getChildren(event.getPath(), true));
    				} catch (KeeperException | InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    	
    	public static void main(String[] args) throws Exception {
    		String path = "/zk-book";
    		zk = new ZooKeeper("23.105.195.9:2181", 5000, new ZkReadDemo1());
    		latch.await();
    		zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    		zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    		
    		List<String> childrenList = zk.getChildren(path, true);
    		System.out.println(childrenList);
    		
    		//一旦此时有子节点创建，服务器就向客户端发送一个子节点变更的事件通知，于是客户端在这个事件通知之后就可以
    		//再次调用getChildren方法来获取新的子节点列表
    		zk.create(path + "/c2", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    		
    		Thread.sleep(Integer.MAX_VALUE);
    	}
    
    }

//使用同步api获取子节点列表
    //我们将子节点列表获取逻辑进行了异步化。异步接口通常会应用在这样的使用场景中：应用启动的时候，会获取一些配置信息，
    //例如机器列表，这些配置通常比较大，并且不希望配置的获取影响应用的主流程
    public class ZkReadDemo2 implements Watcher {
    	
    	private static CountDownLatch latch = new CountDownLatch(1);
    	private static ZooKeeper zk = null;
    
    	//关于watcher，zookeeper服务端在向客户端发送watcher的“NodeChildrenChanged”事件
    	//通知的时候，仅仅只会发出一个通知，而不会把节点的变化情况发送给客户端，需要客户端重新获取。
    	//另外，由于watcher通知是一次性的，即一旦触发一次通知后，该watcher就失效了，因此客户端
    	//需要反复注册watcher
    	@Override
    	public void process(WatchedEvent event) {
    		if(KeeperState.SyncConnected == event.getState()) {
    			if(EventType.None == event.getType() && null == event.getPath()) {
    				latch.countDown();
    			} else if(event.getType() == EventType.NodeChildrenChanged) {
    				try {
    					//我们可以看到，获取的都是子节点的相对路径
    					System.out.println("获取子节点数据:" + zk.getChildren(event.getPath(), true));
    				} catch (KeeperException | InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    	
    	public static void main(String[] args) throws Exception {
    		String path = "/zk-book1";
    		zk = new ZooKeeper("23.105.195.9:2181", 5000, new ZkReadDemo2());
    		latch.await();
    		zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    		zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    		
    		zk.getChildren(path, true, new IChildren2Callback(), null);
    		
    		//一旦此时有子节点创建，服务器就向客户端发送一个子节点变更的事件通知，于是客户端在这个事件通知之后就可以
    		//再次调用getChildren方法来获取新的子节点列表
    		zk.create(path + "/c2", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    		
    		Thread.sleep(Integer.MAX_VALUE);
    	}
    
    }
    
    class IChildren2Callback implements AsyncCallback.Children2Callback {
    
    	@Override
    	public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
    		System.out.println(rc + "; " + path + "; " + ctx + "; " + children + "; " + stat);
    	}
    	
    }

getData:

byte\[\] getData(final String path, Watcher watcher, Stat stat)  
 byte\[\] getData(String path, boolean watch, Stat stat)

void getData(final String path, Wathcer watcher, DataCallback cb, Object ctx)  
 void getData(String path, boolean watch, DataCallback cb, Object ctx)

![Center 2][]

//使用同步api获取数据
    public class ZkGetData1 implements Watcher {
    	
    	private static CountDownLatch latch = new CountDownLatch(1);
    	private static ZooKeeper zk = null;
    	private static Stat stat = new Stat();
    	
    	public static void main(String[] args) throws Exception {
    		String path = "/zk-book3";
    		zk = new ZooKeeper("23.105.195.9:2181", 5000, new ZkGetData1());
    		latch.await();
    		zk.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    		System.out.println(new String(zk.getData(path, true, stat)));
    		
    		System.out.println(stat.getCzxid() + "," + stat.getMzxid() + "," + stat.getVersion());
    		
    		zk.setData(path, "123".getBytes(), -1); //修改数据
    		
    		Thread.sleep(Integer.MAX_VALUE);
    	}
    	
    	
    	@Override
    	public void process(WatchedEvent event) {
    		if(KeeperState.SyncConnected == event.getState()) {
    			if(EventType.None == event.getType() && null == event.getPath()) {
    				latch.countDown();
    			} else if(event.getType() == EventType.NodeDataChanged) { //修改数据触发了NodeDataChanged
    				try {
    					System.out.println(new String(zk.getData(event.getPath(), true, stat)));
    					System.out.println(stat.getCzxid() + "," + stat.getMzxid() + "," + stat.getVersion());
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    
    	
    	
    }

import java.io.IOException;
    import java.util.concurrent.CountDownLatch;
    
    import org.apache.zookeeper.AsyncCallback;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.Watcher.Event.EventType;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.ZooDefs.Ids;
    import org.apache.zookeeper.data.Stat;
    
    //使用异步api获取数据
    public class ZkGetData2 implements Watcher {
    	
    	private static CountDownLatch latch = new CountDownLatch(1);
    	private static ZooKeeper zk = null;
    	private static Stat stat = new Stat();
    	
    	public static void main(String[] args) throws Exception {
    		String path = "/zk-book4";
    		zk = new ZooKeeper("23.105.195.9:2181", 5000, new ZkGetData2());
    		latch.await();
    		zk.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    		
    		zk.getData(path, true, new IDataCallback(), null);
    		
    		zk.setData(path, "123".getBytes(), -1); //修改数据
    		
    		Thread.sleep(Integer.MAX_VALUE);
    	}
    	
    	
    	@Override
    	public void process(WatchedEvent event) {
    		if(KeeperState.SyncConnected == event.getState()) {
    			if(EventType.None == event.getType() && null == event.getPath()) {
    				latch.countDown();
    			} else if(event.getType() == EventType.NodeDataChanged) { //修改数据触发了NodeDataChanged
    				try {
    					System.out.println(new String(zk.getData(event.getPath(), true, stat)));
    					System.out.println(stat.getCzxid() + "," + stat.getMzxid() + "," + stat.getVersion());
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    	
    }
    
    class IDataCallback implements AsyncCallback.DataCallback {
    
    	@Override
    	public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
    		System.out.println(rc + ", " + path + ", " + new String(data));
    		System.out.println(stat.getCzxid() + "," + stat.getMzxid() + "," + stat.getVersion());
    	}
    	
    }

7.检查节点是否存在

public Stat exists(final String path, Watcher watcher)  
public Stat exists(String path, boolean watch)  
public void exists(final String path, Watcher watcher, StatCallback cb, Object ctx)  
public void exists(String path, boolean watch, StatCallback cb, Object ctx)

![Center 3][]

import java.util.concurrent.CountDownLatch;
    
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.Watcher.Event.EventType;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.ZooDefs.Ids;
    import org.apache.zookeeper.data.Stat;
    
    //使用同步接口检测节点是否存在
    public class ZkExistsDemo implements Watcher {
    
    
        private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
        private static ZooKeeper zk;
        public static void main(String[] args) throws Exception {
    
        	String path = "/zk-book8";
        	zk = new ZooKeeper("23.105.195.9:2181", 5000, new ZkExistsDemo());
        	connectedSemaphore.await();
    
        	zk.exists( path, true );
        	
        	zk.create( path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT );
        	
        	zk.setData( path, "123".getBytes(), -1 );
        	
        	zk.create( path+"/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT );
        	
        	zk.delete( path+"/c1", -1 );
        	
        	zk.delete( path, -1 );
        	
            Thread.sleep( Integer.MAX_VALUE );
        }
    
        @Override
        public void process(WatchedEvent event) {
            try {
                if (KeeperState.SyncConnected == event.getState()) {
                    if (EventType.None == event.getType() && null == event.getPath()) {
                        connectedSemaphore.countDown();
                    } else if (EventType.NodeCreated == event.getType()) {
                        System.out.println("Node(" + event.getPath() + ")Created");
                        zk.exists( event.getPath(), true );
                    } else if (EventType.NodeDeleted == event.getType()) {
                        System.out.println("Node(" + event.getPath() + ")Deleted");
                        zk.exists( event.getPath(), true );
                    } else if (EventType.NodeDataChanged == event.getType()) {
                        System.out.println("Node(" + event.getPath() + ")DataChanged");
                        zk.exists( event.getPath(), true );
                    }
                }
            } catch (Exception e) {}
        }
    
    }

[Center]: /images/20220617/1df0067bb73a400b900b51e95107fc04.png
[Center 1]: /images/20220617/8c95efae38e4421b98d8fe8e28883baf.png
[Center 2]: /images/20220617/6507ce1ae2294c16b9cebf343e60ee7c.png
[Center 3]: /images/20220617/c088ba5381ff498a9d6641db51d13f70.png