ZookeeperTest1java
package com.pa.zookeeper.test1; import java.io.IOException; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooKeeper; public class ZookeeperTest1 { public static void main(String[] args) throws KeeperException, InterruptedException, IOException { // test1(); test2(); } public static void test1() throws IOException, KeeperException, InterruptedException{ //建立一个Zookeeper实例,第一个参数为目标服务器地址和端口,第二个参数为Session超时时间,第三个为节点变化时的回调方法 ZooKeeper zk = new ZooKeeper("192.168.56.103:3000,192.168.56.103:3001,192.168.56.103:3002", 500000,new Watcher() { // 监控全部被触发的事件 public void process(WatchedEvent event) { System.out.println(event.getType() + "---" + event.getState()); } }); //建立一个节点root,数据是mydata,不进行ACL权限控制,节点为永久性的(即客户端shutdown了也不会消失) zk.create("/root", "mydata".getBytes(),Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); //在root下面建立一个childone znode,数据为childone,不进行ACL权限控制,节点为永久性的 zk.create("/root/childone","childone".getBytes(), Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); //取得/root节点下的子节点名称,返回List<String> zk.getChildren("/root",true); //取得/root/childone节点下的数据,返回byte[] zk.getData("/root/childone", true, null); //修改节点/root/childone下的数据,第三个参数为版本,若是是-1,那会无视被修改的数据版本,直接改掉 zk.setData("/root/childone","childonemodify".getBytes(), -1); //删除/root/childone这个节点,第二个参数为版本,-1的话直接删除,无视版本 zk.delete("/root/childone", -1); zk.delete("/root", -1); //关闭session zk.close(); } public static void test2() throws IOException, KeeperException, InterruptedException{ // 建立一个与服务器的链接 ZooKeeper zk = new ZooKeeper("192.168.56.103:3000,192.168.56.103:3001,192.168.56.103:3002", 500000, new Watcher() { // 监控全部被触发的事件 public void process(WatchedEvent event) { System.out.println("已经触发了" + event.getType() + "事件!"); } }); // 建立一个目录节点 zk.create("/testRootPath", "testRootData".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println(new String(zk.getData("/testRootPath",false,null))); // 建立一个子目录节点 zk.create("/testRootPath/testChildPathOne", "testChildDataOne".getBytes(),Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); // 取出子目录节点列表 System.out.println(zk.getChildren("/testRootPath",true)); System.out.println(new String(zk.getData("/testRootPath/testChildPathOne",false,null))); // 修改子目录节点数据 zk.setData("/testRootPath/testChildPathOne","modifyChildDataOne".getBytes(),-1); System.out.println("目录节点状态:["+zk.exists("/testRootPath",true)+"]"); // 建立另一个子目录节点 zk.create("/testRootPath/testChildPathTwo", "testChildDataTwo".getBytes(),Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); System.out.println(zk.getChildren("/testRootPath",true)); System.out.println(new String(zk.getData("/testRootPath/testChildPathTwo",true,null))); // 删除子目录节点 zk.delete("/testRootPath/testChildPathTwo",-1); zk.delete("/testRootPath/testChildPathOne",-1); // 删除父目录节点 zk.delete("/testRootPath",-1); // 关闭链接 zk.close(); } }
ZookeeperTest2node
package com.pa.zookeeper.test1; import java.io.IOException; import java.util.Date; 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.KeeperState; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooKeeper; /** * ZooKeeper Java Api 使用样例<br> * ZK Api Version: 3.4.3 */ public class ZookeeperTest2 implements Watcher { private static final int SESSION_TIMEOUT = 10000; private static final String CONNECTION_STRING = "192.168.56.103:2181"; // private static final String CONNECTION_STRING = "192.168.56.103:3000,192.168.56.103:3001,192.168.56.103:3002"; private static final String ZK_PATH = "/nileader"; private ZooKeeper zk = null; private static int i = 0; private CountDownLatch connectedSemaphore = new CountDownLatch( 1 ); static void p(String msg){ System.out.println(++i + "," + msg); } static ZookeeperTest2 sample = new ZookeeperTest2(); public static void main( String[] args ) { p("main-1"); sample.createConnection( CONNECTION_STRING, SESSION_TIMEOUT ); if ( sample.createPath( ZK_PATH, "我是节点初始内容" ) ) { p( "main:数据内容: " + sample.readData( ZK_PATH ) + "\n" ); sample.writeData( ZK_PATH, "我是更新后的数据" ); p( "main:更新后的数据内容: " + sample.readData( ZK_PATH ) + "\n" ); sample.deleteNode( ZK_PATH ); } sample.releaseConnection(); } /** * 建立ZK链接 * @param connectString ZK服务器地址列表 * @param sessionTimeout Session超时时间 */ public void createConnection( String connectString, int sessionTimeout ) { this.releaseConnection(); try { p("createConnection:开始建立链接"); zk = new ZooKeeper( connectString, sessionTimeout, this ); connectedSemaphore.await(); p("createConnection:链接建立完毕"); } catch ( InterruptedException e ) { p( "链接建立失败,发生 InterruptedException" ); e.printStackTrace(); } catch ( IOException e ) { p( "链接建立失败,发生 IOException" ); e.printStackTrace(); } } /** * 关闭ZK链接 */ public void releaseConnection() { if ( this.zk != null ) { try { p("关闭链接"); this.zk.close(); } catch ( InterruptedException e ) { // ignore e.printStackTrace(); } } } /** * 收到来自Server的Watcher通知后的处理。 */ @Override public void process( WatchedEvent event ) { p("process:收到事件通知:[state=" + event.getState() + ",type=" + event.getType() + ",path=" + event.getPath() + "]"); if ( KeeperState.SyncConnected == event.getState() ) { connectedSemaphore.countDown(); } String ts = new Date(System.currentTimeMillis()).toString(); if ( KeeperState.SyncConnected == event.getState() && !event.getPath().equals("") ) { p( "process:中获取数据内容:(" + ts + ") ========================"); String data = sample.readData( ZK_PATH ); p( "process:中获取数据内容完毕: (" + ts + ")========================" + data); } } /** * 建立节点 * @param path 节点path * @param data 初始数据内容 * @return */ public boolean createPath( String path, String data ) { try { p( "createPath:开始建立节点, Path: " + this.zk.create( path, // data.getBytes(), // Ids.OPEN_ACL_UNSAFE, // CreateMode.EPHEMERAL ) + ", content: " + data ); p("createPath:节点建立完毕"); } catch ( KeeperException e ) { p( "节点建立失败,发生KeeperException" ); e.printStackTrace(); } catch ( InterruptedException e ) { p( "节点建立失败,发生 InterruptedException" ); e.printStackTrace(); } return true; } /** * 读取指定节点数据内容 * @param path 节点path * @return */ public String readData( String path ) { try { p("readData:开始获取数据,path:" + path ); String data = new String( this.zk.getData( path, true, null ) ); p("readData:获取数据完毕:" + data); return data; } catch ( KeeperException e ) { p( "读取数据失败,发生KeeperException,path: " + path ); e.printStackTrace(); return ""; } catch ( InterruptedException e ) { p( "读取数据失败,发生 InterruptedException,path: " + path ); e.printStackTrace(); return ""; } } /** * 更新指定节点数据内容 * @param path 节点path * @param data 数据内容 * @return */ public boolean writeData( String path, String data ) { try { p( "writeData:开始更新数据"); p("writeData:path:" + path + ", stat: " + this.zk.setData( path, data.getBytes(), -1 ) ); p("writeData:数据更新完毕"); } catch ( KeeperException e ) { p( "更新数据失败,发生KeeperException,path: " + path ); e.printStackTrace(); } catch ( InterruptedException e ) { p( "更新数据失败,发生 InterruptedException,path: " + path ); e.printStackTrace(); } return false; } /** * 删除指定节点 * @param path 节点path */ public void deleteNode( String path ) { try { p("deleteNode:开始删除节点"); this.zk.delete( path, -1 ); p( "deleteNode:删除节点成功,path:" + path ); } catch ( KeeperException e ) { p( "删除节点失败,发生KeeperException,path: " + path ); e.printStackTrace(); } catch ( InterruptedException e ) { p( "删除节点失败,发生 InterruptedException,path: " + path ); e.printStackTrace(); } } }