Zookeeper开发小结

一.实例化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 once
zooKeeper.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_UNSAFE
zooKeeper.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通知之后,在处理过程中,数据仍然有变更的可能,因此在时间线上,不可能做到完全准确.