Zookeeper原始API的使用

水深无声 2022-07-13 03:26 79阅读 0赞

1、Zk客户端的简单介绍

1.1 Zk构造函数参数所明

![20161222163513703][]

1.2 Zk创建节点的API讲解

![20161222163604000][]

![20161222163607611][]

![20161222163611172][]

1.3 关于exists  API的讲解

![20161222163707392][]

2、Zk客户端，不使用Watch的简单代码测试

public class ZookeeperBase {
 
 	/** zookeeper地址 */
 	private final String CONNECT_ADDR = "192.168.2.101:2181";
 	/** session超时时间 */
 	private final int SESSION_OUTTIME = 2000;// ms
 	/*** 用于操作Zookeeper */
 	private ZooKeeper zk = null;
 	/** zk父路径设置 */
 	private static final String PARENT_PATH = "/testWatch";
 	/** zk子路径设置 */
 	private static final String CHILDREN_PATH = "/testWatch/children";
 
 	public void connection() {
 		try {
 			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 建立连接");
 						}
 					}
 				}
 			});
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 
 	}
 
 	/**
 	 * 创建节点的路径
 	 * 
 	 * @param path
 	 * 节点的路径
 	 * @param data
 	 * 节点的数据
 	 * @param createMode
 	 * 节点的模式
 	 * @return 创建节点的路径
 	 */
 	public String createNode(String path, String data, CreateMode createMode) {
 		String result = null;
 		try {
 			result = zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE,
 					createMode);
 		} catch (KeeperException e) {
 			e.printStackTrace();
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 		}
 		return result;
 	}
 
 	/**
 	 * 
 	 * @param path
 	 * 要获取数据的路径
 	 * 
 	 * @return 路径节点的数据
 	 */
 	public String getData(String path) {
 		String result = null;
 		try {
 			byte[] data = zk.getData(path, false, null);
 			result = new String(data);
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 		return result;
 	}
 
 	/**
 	 * 设置节点信息
 	 * 
 	 * @param path
 	 * @param data
 	 * @return
 	 */
 	public String setDatd(String path, String data) {
 		String result = null;
 		try {
 			zk.setData(path, data.getBytes(), -1);
 			result = new String(data);
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 		return result;
 	}
 
 	/**
 	 * 同步的方式，删除节点信息
 	 * 
 	 * @param path
 	 */
 	public void deleteNode(String path) {
 		try {
 			zk.delete(path, -1);
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 	}
 
 	/**
 	 * 使用异步的方式，删除节点信息
 	 * 
 	 * @param path
 	 * @param object
 	 */
 	public void deleteNodeCallback(String path, Object object) {
 		try {
 			zk.delete(path, -1, new VoidCallback() {
 
 				@Override
 				public void processResult(int paramInt, String paramString,
 						Object paramObject) {
 					System.out.println("paramString: " + paramString);
 					System.out.println("paramObject: " + paramObject);
 				}
 			}, object);
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 	}
 
 	/**
 	 * 返回父路径下的，孩子节点的名字信息
 	 * 
 	 * @param parentPath
 	 * @return
 	 */
 	public List<String> getChildsNameInfo(String parentPath) {
 		List<String> childrens = null;
 		try {
 			childrens = zk.getChildren(parentPath, false);
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 		return childrens;
 	}
 
 	/**
 	 * 返回父目录下的子节点信息
 	 * 
 	 * @param parentPath
 	 * @return
 	 */
 	public Map<String, String> getChildsDataInfo(String parentPath) {
 		List<String> childrensNameInfo = null;
 		Map<String, String> childrens = new HashMap<String, String>();
 		try {
 			childrensNameInfo = zk.getChildren(parentPath, false);
 			for (int i = 0; i < childrensNameInfo.size(); i++) {
 				String data = getData(parentPath + childrensNameInfo.get(i));
 				childrens.put(parentPath + childrensNameInfo.get(i), data);
 			}
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 		return childrens;
 	}
 
 	public void close() {
 		if (zk != null) {
 			try {
 				zk.close();
 				System.out.println("zk 释放连接");
 			} catch (InterruptedException e) {
 				e.printStackTrace();
 			}
 		}
 	}
 
 	public Stat exists(String path) {
 		Stat stat = null;
 		try {
 			stat = zk.exists(CHILDREN_PATH, false);
 		} catch (KeeperException e) {
 			e.printStackTrace();
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 		}
 		return stat;
 	}
 
 	// 用于测试zk
 	public static void main(String[] args) throws Exception {
 
 		// 1.创建zk客户端
 		ZookeeperBase zookeeperBase = new ZookeeperBase();
 		// 连接服务端
 		zookeeperBase.connection();
 
 		// 2.创建父节点信息
 		zookeeperBase.createNode(PARENT_PATH, "testParent",
 				CreateMode.PERSISTENT);
 		// 3.创建子节点信息
 		zookeeperBase.createNode(CHILDREN_PATH, "testChilren",
 				CreateMode.PERSISTENT);
 
 		// 4.获取父节点的数据信息
 		String parentData = zookeeperBase.getData(PARENT_PATH);
 		System.out.println(parentData);
 		// 5.获取子节点的数据信息
 		String data = zookeeperBase.getData(CHILDREN_PATH);
 		System.out.println(data);
 		// 6.设置父节点的数据信息
 		zookeeperBase.setDatd(PARENT_PATH, "haha");
 		// 7.设置子节点的数据信息
 		zookeeperBase.setDatd(CHILDREN_PATH, "heihei");
 		// 8.同步删除节点信息
 		zookeeperBase.deleteNode(CHILDREN_PATH);
 		// 9.异步的删除节点的信息（带有回调的方式）
 		zookeeperBase.deleteNodeCallback(PARENT_PATH, "aaa");
 
 		// 关闭连接
 		zookeeperBase.close();
 
 	}
 }

3、Watcher的介绍

3.1 Watcher组成

![20161222163830596][]

3.2 Watcher的事件

![20161222163902910][]

![20161222163911784][]

3.3 Watcher的注册

![20161222163934426][]

4、使用Watcher的Zk客户端代码

/**
 * Zookeeper Wathcher 本类就是一个Watcher类（实现了org.apache.zookeeper.Watcher类）
 * 
 */
 public class ZooKeeperWatcher implements Watcher {
 
 	/** 定义原子变量 */
 	AtomicInteger seq = new AtomicInteger();
 	/** 定义session失效时间 */
 	private static final int SESSION_TIMEOUT = 10000;
 	/** zookeeper服务器地址 */
 	private static final String CONNECTION_ADDR = "192.168.2.101:2181";
 	/** zk父路径设置 */
 	private static final String PARENT_PATH = "/testWatch";
 	/** zk子路径设置 */
 	private static final String CHILDREN_PATH = "/testWatch/children";
 	/** 进入标识 */
 	private static final String LOG_PREFIX_OF_MAIN = "【Main】";
 	/** zk变量 */
 	private ZooKeeper zk = null;
 	/** 用于等待zookeeper连接建立之后 通知阻塞程序继续向下执行 */
 	private CountDownLatch connectedCountDownLatch = new CountDownLatch(1);
 
 	/**
 	 * 创建ZK连接
 	 * 
 	 * @param connectAddr
 	 * ZK服务器地址列表
 	 * @param sessionTimeout
 	 * Session超时时间
 	 */
 	public void createConnection(String connectAddr, int sessionTimeout) {
 		this.releaseConnection();
 		try {
 			zk = new ZooKeeper(connectAddr, sessionTimeout, this);
 			System.out.println(LOG_PREFIX_OF_MAIN + "开始连接ZK服务器");
 			connectedCountDownLatch.await();
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 	}
 
 	/**
 	 * 关闭ZK连接
 	 */
 	public void releaseConnection() {
 		if (this.zk != null) {
 			try {
 				this.zk.close();
 			} catch (InterruptedException e) {
 				e.printStackTrace();
 			}
 		}
 	}
 
 	/**
 	 * 创建节点
 	 * 
 	 * @param path
 	 * 节点路径
 	 * @param data
 	 * 数据内容
 	 * @return
 	 */
 	public boolean createPath(String path, String data) {
 		try {
 			// 设置监控(由于zookeeper的监控都是一次性的所以 每次必须设置监控)
 			this.zk.exists(path, true);
 			// 创建节点
 			System.out.println(LOG_PREFIX_OF_MAIN + "节点创建成功, Path: "
 					+ this.zk.create( /** 路径 */
 					path,
 					/** 数据 */
 					data.getBytes(),
 					/** 所有可见 */
 					Ids.OPEN_ACL_UNSAFE,
 					/** 永久存储 */
 					CreateMode.PERSISTENT) + ", content: " + data);
 		} catch (Exception e) {
 			e.printStackTrace();
 			return false;
 		}
 		return true;
 	}
 
 	/**
 	 * 读取指定节点数据内容
 	 * 
 	 * @param path
 	 * 节点路径
 	 * @return
 	 */
 	public String readData(String path, boolean needWatch) {
 		try {
 			return new String(this.zk.getData(path, needWatch, null));
 		} catch (Exception e) {
 			e.printStackTrace();
 			return "";
 		}
 	}
 
 	/**
 	 * 更新指定节点数据内容
 	 * 
 	 * @param path
 	 * 节点路径
 	 * @param data
 	 * 数据内容
 	 * @return
 	 */
 	public boolean writeData(String path, String data) {
 		try {
 			System.out.println(LOG_PREFIX_OF_MAIN + "更新数据成功，path：" + path
 					+ ", stat: " + this.zk.setData(path, data.getBytes(), -1));
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 		return false;
 	}
 
 	/**
 	 * 删除指定节点
 	 * 
 	 * @param path
 	 * 节点path
 	 */
 	public void deleteNode(String path) {
 		try {
 			this.zk.delete(path, -1);
 			System.out.println(LOG_PREFIX_OF_MAIN + "删除节点成功，path：" + path);
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 	}
 
 	/**
 	 * 判断指定节点是否存在,实际是用来watch节点
 	 * 
 	 * @param path
 	 * 节点路径
 	 */
 	public Stat exists(String path, boolean needWatch) {
 		try {
 			return this.zk.exists(path, needWatch);
 		} catch (Exception e) {
 			e.printStackTrace();
 			return null;
 		}
 	}
 
 	/**
 	 * 获取子节点
 	 * 
 	 * @param path
 	 * 节点路径
 	 */
 	private List<String> getChildren(String path, boolean needWatch) {
 		try {
 			return this.zk.getChildren(path, needWatch);
 		} catch (Exception e) {
 			e.printStackTrace();
 			return null;
 		}
 	}
 
 	/**
 	 * 删除所有节点
 	 */
 	public void deleteAllTestPath() {
 		if (this.exists(CHILDREN_PATH, false) != null) {
 			this.deleteNode(CHILDREN_PATH);
 		}
 		if (this.exists(PARENT_PATH, false) != null) {
 			this.deleteNode(PARENT_PATH);
 		}
 	}
 
 	/**
 	 * 收到来自Server的Watcher通知后的处理。
 	 */
 	@Override
 	public void process(WatchedEvent event) {
 		System.out.println("=========================================");
 		System.out.println("进入 process 。。。。。event = " + event);
 
 		if (event == null) {
 			return;
 		}
 
 		// 连接状态
 		KeeperState keeperState = event.getState();
 		// 事件类型
 		EventType eventType = event.getType();
 		// 受影响的path
 		String path = event.getPath();
 
 		String logPrefix = "【Watcher-" + this.seq.incrementAndGet() + "】";
 
 		System.out.println(logPrefix + "收到Watcher通知");
 		System.out.println(logPrefix + "连接状态:\t" + keeperState.toString());
 		System.out.println(logPrefix + "事件类型:\t" + eventType.toString());
 
 		if (KeeperState.SyncConnected == keeperState) {
 			// 成功连接上ZK服务器
 			if (EventType.None == eventType) {
 				System.out.println(logPrefix + "成功连接上ZK服务器");
 				connectedCountDownLatch.countDown();
 			}
 			// 创建节点
 			else if (EventType.NodeCreated == eventType) {
 				System.out.println(logPrefix + "节点创建");
 				try {
 					Thread.sleep(100);
 				} catch (InterruptedException e) {
 					e.printStackTrace();
 				}
 				this.exists(path, true);
 			}
 			// 更新节点
 			else if (EventType.NodeDataChanged == eventType) {
 				System.out.println(logPrefix + "节点数据更新");
 				try {
 					Thread.sleep(100);
 				} catch (InterruptedException e) {
 					e.printStackTrace();
 				}
 				System.out.println(logPrefix + "数据内容: "
 						+ this.readData(PARENT_PATH, true));
 			}
 			// 更新子节点
 			else if (EventType.NodeChildrenChanged == eventType) {
 				System.out.println(logPrefix + "子节点变更");
 				try {
 					Thread.sleep(3000);
 				} catch (InterruptedException e) {
 					e.printStackTrace();
 				}
 				// 查看受影响的子节点信息
 				System.out.println(logPrefix + "子节点列表："
 						+ this.getChildren(PARENT_PATH, true));
 			}
 			// 删除节点
 			else if (EventType.NodeDeleted == eventType) {
 				System.out.println(logPrefix + "节点 " + path + " 被删除");
 			}
 		} else if (KeeperState.Disconnected == keeperState) {
 			System.out.println(logPrefix + "与ZK服务器断开连接");
 		} else if (KeeperState.AuthFailed == keeperState) {
 			System.out.println(logPrefix + "权限检查失败");
 		} else if (KeeperState.Expired == keeperState) {
 			System.out.println(logPrefix + "会话失效");
 		}
 
 		System.out.println("=========================================");
 		System.out.println();
 	}
 
 	/**
 	 * 方法名称：测试zookeeper监控 
 	 * 概要说明：主要测试watch功能 
 	 * 
 	 * @param args
 	 * @throws Exception
 	 */
 	public static void main(String[] args) throws Exception {
 
 		// 建立watcher
 		ZooKeeperWatcher zkWatch = new ZooKeeperWatcher();
 		// 创建连接
 		zkWatch.createConnection(CONNECTION_ADDR, SESSION_TIMEOUT);
 		// System.out.println(zkWatch.zk.toString());
 
 		Thread.sleep(1000);
 
 		// 清理之前的节点信息
 		zkWatch.deleteAllTestPath();
 
 		// 1.创建节点信息，并watch这个节点 ========
 		zkWatch.createPath(PARENT_PATH, System.currentTimeMillis() + "");
 		// 2.对父节点的孩子节点，进行watch(不然父节点的子节点变化，将收不到watch)
 		zkWatch.getChildren(PARENT_PATH, true);
 		Thread.sleep(1000);
 		// 3.创建子节点
 		zkWatch.createPath(CHILDREN_PATH, System.currentTimeMillis() + "");
 
 		Thread.sleep(50000);
 		// 4.删除节点信息
 		zkWatch.deleteAllTestPath();
 		Thread.sleep(1000);
 		zkWatch.releaseConnection();
 	}
 }

5、[源码下载][Link 1]

[20161222163513703]: /images/20220713/bec74d73efc341f6aeca57e25031ce3e.png
[20161222163604000]: /images/20220713/fcc79286f748436e9e826044f28ffbd0.png
[20161222163607611]: /images/20220713/cb1642c3160543d2b383ffe2d89986f6.png
[20161222163611172]: /images/20220713/b3fa70c15b5d49a993597d125c6ca815.png
[20161222163707392]: /images/20220713/8b577df6d69743a9bf452f207eaae38f.png
[20161222163830596]: /images/20220713/3c420408e4764575a08d9833b19168d5.png
[20161222163902910]: /images/20220713/6f206fbd2992465893e951511b46d4c8.png
[20161222163911784]: /images/20220713/9491a4dbc66144aea4d6e6605ccf3059.png
[20161222163934426]: /images/20220713/265cc387f8cf4dc9bca0f3b8287f8ac1.png
[Link 1]: http://pan.baidu.com/s/1pLhM1KJ