Zookeeper开发小结
public class ZkClient { private ZooKeeper zooKeeper; private String connectString; private Integer sessionTimeout; private Object waiter = new Object();//simple object-lock //"zk-client.properties";//classpath下 public ZkClient(String configLocation) throws Exception{ Properties config = new Properties(); config.load(ClassLoader.getSystemResourceAsStream(configLocation)); connectString = config.getProperty("zk.connectString"); if(connectString == null){ throw new NullPointerException("'zk.connectString' cant be empty.. "); } sessionTimeout = Integer.parseInt(config.getProperty("zk.sessionTimeout","-1")); connectZK(); } /** * core method,启动zk服务 本实例基于自动重连策略,如果zk连接没有建立成功或者在运行时断开,将会自动重连. */ private void connectZK() { synchronized (waiter) { try { SessionWatcher watcher = new SessionWatcher(); // session的构建是异步的 this.zooKeeper = new ZooKeeper(connectString, sessionTimeout, watcher, false); } catch (Exception e) { e.printStackTrace(); } waiter.notifyAll(); } } class SessionWatcher implements Watcher { public void process(WatchedEvent event) { // 如果是“数据变更”事件 if (event.getType() != Event.EventType.None) { return; } // 如果是链接状态迁移 // 参见keeperState synchronized (waiter) { switch (event.getState()) { // zk连接建立成功,或者重连成功 case SyncConnected: System.out.println("Connected..."); waiter.notifyAll(); break; // session过期,这是个非常严重的问题,有可能client端出现了问题,也有可能zk环境故障 // 此处仅仅是重新实例化zk client case Expired: System.out.println("Expired..."); // 重连 connectZK(); break; // session过期 case Disconnected: // 链接断开，或session迁移 System.out.println("Connecting...."); break; case AuthFailed: close(); throw new RuntimeException("ZK Connection auth failed..."); default: break; } } } } private void close(){ try { synchronized (waiter) { if (this.zooKeeper != null) { zooKeeper.close(); } waiter.notifyAll(); } } catch (Exception e) { e.printStackTrace(); } }}

二.授权与验证

String auth = "admin:admin";//anywhere,but before znode operation//can addauth more than oncezooKeeper.addAuthInfo("digest", auth.getBytes("UTF-8"));

?

Id id = new Id("digest", DigestAuthenticationProvider.generateDigest(auth));ACL acl = new ACL(ZooDefs.Perms.ALL, id);List<ACL> acls = Collections.singletonList(acl);//如果不需要访问控制,可以使用acls = ZooDefs.Ids.OPEN_ACL_UNSAFEzooKeeper.create("/singleWorker", null, acls, CreateMode.PERSISTENT);

三.创建znode节点

try{Id id = new Id("digest", DigestAuthenticationProvider.generateDigest(auth));ACL acl = new ACL(ZooDefs.Perms.ALL, id);List<ACL> acls = Collections.singletonList(acl);zooKeeper.create("/workers", null, acls, CreateMode.PERSISTENT);}catch(KeeperException.NodeExistsException e){//在并发环境中,节点创建有可能已经被创建,即使使用exist方法检测也不能确保.} catch (Exception e){e.printStackTrace();}

四.删除节点

//zookeeper不允许直接删除含有子节点的节点;//如果你需要删除当前节点以及其所有子节点,需要递归来做private void deletePath(String path,ZooKeeper zooKeeper) throws Exception{List<String> children = zooKeeper.getChildren(path,false);for(String child : children){String childPath = path + "/" + child;deletePath(childPath,zooKeeper);}try{ zooKeeper.delete(path,-1);}catch(KeeperException.NoNodeException e){//ignore}}

//删除节点,删除时比较version,避免删除时被其他client修改public boolean delete(String path){try{Stat stat = zooKeeper.exists(path,false);//如果节点已经存在if(stat != null){zooKeeper.delete(path,stat.getVersion());}}catch(KeeperException.NoNodeException e){//igore}catch (Exception e){e.printStackTrace();                return false;}return true;}

五.修改数据

public boolean update(String path,byte[] data){  try{  Stat stat = zooKeeper.exists(path,false);  //如果节点已经存在  if(stat != null){  zooKeeper.setData(path,data,stat.getVersion());  return true;  }  }catch (KeeperException.NoNodeException e){   //ignore  }catch (Exception e){  e.printStackTrace();  }return false;}

五.事务

public boolean create(String name){try{Transaction tx = zooKeeper.transaction();tx.create("/workers/servers/" + name,null, ZooDefs.Ids.OPEN_ACL_UNSAFE,null);tx.create("/workers/schedule/" + name,null, ZooDefs.Ids.OPEN_ACL_UNSAFE,null);tx.commit();return true;} catch (Exception e){e.printStackTrace();}return false;}

? ? 备注:zookeeper中的watcher机制非常好,但是重要的数据变更,不能完全依赖watcher通知,因为对于zkClient而言,网络异常都将会导致watcher有丢失的潜在风险,而且watcher是"即发即失",当你接收到watcher通知之后,在处理过程中,数据仍然有变更的可能,因此在时间线上,不可能做到完全准确.???????

?