public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)
        throws IOException {
	this(connectString, sessionTimeout, watcher, false);	// boolean canBeReadOnly
}

public class FirstWatcher implements Watcher {
    public void process(WatchedEvent event) {
        System.out.println("Watcher: " + event);
    }
}

public static void main(String[] args) throws Exception {
	String hostPort = "127.0.0.1:2181";
	int timeout = 15000;
	FirstWatcher watcher = new FirstWatcher();
	ZooKeeper zk = new ZooKeeper(hostPort, timeout, watcher);
	// 等待接受Event后再关闭
	Thread.sleep(5 * 1000);
	// zk.close();
}

private String serverId = "abc";
private boolean isLeader = false;

public void runForMaster(ZooKeeper zk) throws InterruptedException{
	while(true) {
		try {
			// 尝试创建/master临时znode，成功则自己为主节点
			zk.create("/master", serverId.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
			isLeader = true;
			break;
		// 如果节点已经存在，那么自己就是从节点
		} catch (KeeperException.NodeExistsException e) {
			isLeader = false;
			break;
		// 多种可能，在不确认前不希望确定主节点
		} catch (KeeperException.ConnectionLossException e) {
			// ----> 去尝试判断自己是否是主节点
			if (checkMaster(zk)) {
				break;
			}
		} catch (KeeperException e) {
			// LOGGER错误日志，中止
			break;
		}
	}
}

private boolean checkMaster(ZooKeeper zk) throws InterruptedException{
	while(true) {
		try {
			// 获取/master节点的数据，来判断自己是否是群首
			byte data[] = zk.getData("/master", false, new Stat());
			isLeader = new String(data).equals(serverId);
			return true;
		} catch (KeeperException.NodeExistsException e) {
			// 没有节点，继续尝试创建
			return false;
		} catch (KeeperException.ConnectionLossException e) {
			// 不处理，可能响应信息丢失
		} catch (KeeperException e) {
			// LOGGER错误日志，中止
			return true;
		}
	}
}

private AsyncCallback.StringCallback masterCreateCallback = new AsyncCallback.StringCallback() {
	public void processResult(int rc, String path, Object ctx, String name) {
		switch (KeeperException.Code.get(rc)) {
			case CONNECTIONLOSS:
				checkMaster();
				return;
			case OK:
				isLeader = true;
				break;
			default:
				isLeader = false;
		}
	}
};

public void runForMaster() {
	zk.create("/master", serverId.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
			masterCreateCallback, null);
}

private AsyncCallback.DataCallback masterCheckCallback = new AsyncCallback.DataCallback() {
	public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
		switch (KeeperException.Code.get(rc)) {
			case CONNECTIONLOSS:
				// 继续尝试确认
				checkMaster();
			case NONODE:
				// 通过创建来判断
				runForMaster();
				return;
		}
	}
};

private void checkMaster() {
	zk.getData("/master", false, masterCheckCallback, null);
}

private AsyncCallback.StringCallback createNodeCallback = new AsyncCallback.StringCallback() {
	public void processResult(int rc, String path, Object ctx, String name) {
		switch (KeeperException.Code.get(rc)) {
			case CONNECTIONLOSS:
				// 尝试再创建
				createNode(path, (byte[])ctx);
				return;
			case OK:
				// 成功，doSomething...
				break;
			case NODEEXISTS:
				// 已经存在
				break;
			default:
				// LOGGER日志
		}
	}
};

// 传入data，第二个参数表示要保存到znode的数据，第四个参数可以在回调函数中继续使用
zk.create("/master", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
		createNodeCallback, data);

public void setStatus(String status) {
	this.status = status;
	updateStatus(status);
}

synchronized private void updateStatus(String status) {
	// 重新处理异步请求连接丢失时有个小问题：
	// 处理流程可能变得无序，因为ZK对请求和响应都会很好保持顺序，但如果连接丢失，再发起一个新的请求，会导致整个时序出现空隙
	// 因此进行一个状态更新请求前，先获得当前状态，否则就要放弃更新
	if (status == this.status) {
		// 执行无条件更新，-1代表禁止版本号检查
		zk.setData("/workers/work1", status.getBytes(), -1, statusUpdateCallback, status);
	}
}

private AsyncCallback.StatCallback statusUpdateCallback = new AsyncCallback.StatCallback() {
	public void processResult(int rc, String path, Object ctx, Stat stat) {
		switch (KeeperException.Code.get(rc)) {
			// 丢失事件时，重新更新，由于在updateStatus方法中做了竞态条件的检查，这里就不需要再次检查
			case CONNECTIONLOSS:
				updateStatus((String) ctx);
				return;
		}
	}
};

String name = zk.create("/tasks/task-", task.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL)

Stat stat = new Stat();
// 由于不关心节点变化，第二个参数使用false
byte[] taskData = zk.getData("/tasks", false, stat);
// 通过Stat结构，可以获得当前节点的信息，这里获取了当前节点创建时间
Date startDate = new Date(stat.getCtime());
System.out.println("task: " + new String(taskData) + " ,since:" + startDate);

System.out.println("tasks:");
for (String name : zk.getChildren("/tasks", false)) {
	byte[] task = zk.getData("/tasks/" + name, false, null);
	System.out.println(name + ": " + new String(task));
}

abstract public void process(WatchedEvent event)

public class WatchedEvent {
	// ZK会话状态：Unknown、Disconnected、SyncConnected、AuthFailed、ConnectedReadOnly、SaslAuthenticated和Expired
    final private KeeperState keeperState;
	// 事件类型：NodeCreated、NodeDeleted、NodeDataChanged、NodeChildrenChanged和None
    // 前面三种事件类型只涉及单个znode节点，第四种涉及监视点的znode节点的子节点。None表示无事件发生，而是ZK的会话状态发生了改变
	final private EventType eventType;
	// 如果事件类型不是None时，返回znode路径
    private String path;
}

// 同步
public byte[] getData(String path, boolean watch, Stat stat)
public byte[] getData(final String path, Watcher watcher, Stat stat);
// 异步
public void getData(String path, boolean watch, DataCallback cb, Object ctx)
public void getData(final String path, Watcher watcher, DataCallback cb, Object ctx)

private AsyncCallback.StringCallback masterCreateCallback = new AsyncCallback.StringCallback() {
	public void processResult(int rc, String path, Object ctx, String name) {
		switch (KeeperException.Code.get(rc)) {
			case CONNECTIONLOSS:
				// 连接丢失时，检查/master节点是否存在
				checkMaster();
				return;
			case OK:
				state = States.ELECTED;
				// do Leader something 行使领导权
				break;
			case NODEEXISTS:
				state = States.NOT_ELECTED;
				// 其他进程已经创建了节点，----> 监视该节点
				masterExists();
				break;
			default:
				state = States.NOT_ELECTED;
				// LOGGGER.error
		}
	}
};

private void masterExists() {
	// 调用exists设置监视点 
	zk.exists("/master", masterExistsWatcher, masterExistsCallback, null);
}

private Watcher masterExistsWatcher = new Watcher() {
	public void process(WatchedEvent event) {
		// 如果/master被删除了，那就重新创建，竞争主节点
		if (event.getType() == Event.EventType.NodeDeleted) {
			assert "/master".equals(event.getPath());
			runForMaster();
		}
	}
};

private AsyncCallback.StatCallback masterExistsCallback = new AsyncCallback.StatCallback() {
	public void processResult(int rc, String path, Object ctx, Stat stat) {
		switch (KeeperException.Code.get(rc)) {
			case CONNECTIONLOSS:
				// 连接丢失时，重新设置监视点
				masterExists();
				break;
			case OK:
				// 在create的回调方法执行和exists操作执行之间发生了/master节点被删除的情况
				// 因此在exists返回操作成功后，需要检查返回的stat对象是否为空，因为当节点不存在时stat为null
				if (stat == null) {
					state = States.RUNNING;
					// 就要重新去创建，竞争主节点
					runForMaster();
				}
				break;
			default:
				// 通过获取节点来检查/master节点是否存在
				checkMaster();
		}
	}
};

public void getWorkers() {
	zk.getChildren("/workers", workersChangeWatcher, workersGetChildrenCallback, null);
}

// 节点列表监视点对象
private Watcher workersChangeWatcher = new Watcher() {
	public void process(WatchedEvent event) {
		if (event.getType() == Event.EventType.NodeChildrenChanged) {
			assert "/workers".equals(event.getPath());
			getWorkers();
		}
	}
};

private AsyncCallback.ChildrenCallback workersGetChildrenCallback = new AsyncCallback.ChildrenCallback() {
	public void processResult(int rc, String path, Object ctx, List<String> children) {
		switch (KeeperException.Code.get(rc)) {
			case CONNECTIONLOSS:
				// 连接丢失时，重新获取子节点并设置监视点的操作
				getWorkers();
				break;
			case OK:
				// 更新本地从节点列表缓存，如果有节点被移除，需要重新分配任务
				reassignAndSet(children);
				break;
			default:
				// LOGGER.error
		}
	}
};

private void assignTasks(List<String> tasks) {
	for (String task : tasks) {
		getTaskData(task);
	}
}

private void getTaskData(String task) {
	zk.getData("/tasks/" + task, false, taskDataCallback, task);
}

private AsyncCallback.DataCallback taskDataCallback = new AsyncCallback.DataCallback() {
	public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
		switch (KeeperException.Code.get(rc)) {
			case CONNECTIONLOSS:
				// 丢失连接时，重新获取数据
				getTaskData((String) ctx);
				break;
			case OK:
				// 随机选择一个工作从节点
				int worker = new Random().nextInt(workerList.size());
				String designatedWorker = workerList.get(worker);
				// 分配任务
				String assignmentPath = "/assign/" + designatedWorker + "/" + (String) ctx;
				createAssignment(assignmentPath, data);
				break;
			default:
				// LOGGER.error
		}
	}
};

// 创建分配任务节点/assign/worker-id/task-num
private void createAssignment(String path, byte[] data) {
	zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, assignTaskCallback, data);
}

private AsyncCallback.StringCallback assignTaskCallback = new AsyncCallback.StringCallback() {
	public void processResult(int rc, String path, Object ctx, String name) {
		switch (KeeperException.Code.get(rc)) {
			case CONNECTIONLOSS:
				createAssignment(path, (byte[]) ctx);
				break;
			case OK:
				// 删除/tasks下对应的任务节点
				deleteTask(name.substring(name.lastIndexOf("/") + 1));
				break;
			default:
				// LOGGER.error
		}
	}
};

List<OpResult> results = zk.multi(Arrays.asList(Op.delete("/a/b", -1), Op.delete("/a", -1)));

Transaction t = zk.transaction();
t.delete("/a/b", -1);
t.delete("/a", -1);
// 同步
List<OpResult> results = t.commit();

// 鉴权模式 + 鉴权信息
public void addAuthInfo(String scheme, byte auth[]);

ZooDefs.Ids.OPEN_ACL_UNSAFE
ZooDefs.Ids.CREATOR_ALL_ACL
ZooDefs.Ids.READ_ACL_UNSAFE

//...
zk.sync(path, voidCallback, ctx);
zk.getData(path, watcher, dataCallback, ctx);
// ...

// RetryPolicy提供重试策略的接口，可以让用户实现自定义的重试策略
// 默认提供了以下实现，分别为ExponentialBackoffRetry、BoundedExponentialBackoffRetry、RetryForever、RetryNTimes、RetryOneTime、RetryUntilElapsed
// 给定一个初始化sleep时间baseSleepTimeMs，然后在此基础上设置重试次数
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
// 还可以传入sessionTimeoutMs、connectionTimeoutMs
// 两种方法内部实现一样，只是对外包装成不同的方法。它们的底层都是通过第三个方法builder来实现的
CuratorFramework zkc = CuratorFrameworkFactory.newClient("127.0.0.1:2181", retryPolicy);
zkc.start();

// 指定一个Zookeeper的根路径，即该客户端对Zookeeper上的数据节点的操作都是基于该目录进行的
// 设置Chroot可以将客户端应用与Zookeeper服务端的一课子树相对应，在多个应用共用一个Zookeeper集群的场景下，这对于实现不同应用之间的相互隔离十分有意义
zkc = CuratorFrameworkFactory.builder()
					.sessionTimeoutMs(5000)
					.connectionTimeoutMs(5000)
					.retryPolicy(retryPolicy)
					// 将在/base下操作
					.namespace("base")
					.build();

// 创建一个初始内容为空的节点，默认持久节点且内容为空
zkc.create().forPath(path);
// 创建一个包含内容的节点
zkc.create().forPath(path, content.getBytes());
// 创建一个包含内容的临时节点
zkc.create().withMode(CreateMode.EPHEMERAL).forPath(path, content.getBytes());
// 创建一个临时空的节点，并递归创建父节点，在递归创建父节点时，父节点为持久节点
zkc.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);

// 删除一个子节点，且只能删除叶子节点
zkc.delete().forPath(path);
// 指定版本进行删除，如果此版本已经不存在，则删除异常
zkc.delete().withVersion(1).forPath(path);
// 删除节点并递归删除其子节点
zkc.delete().deletingChildrenIfNeeded().forPath(path);
// 强制保证删除一个节点，只要客户端会话有效，那么Curator会在后台持续进行删除操作，直到节点删除成功。比如遇到一些网络异常的情况，此guaranteed的强制删除就会很有效果
zkc.delete().guaranteed().forPath(path);

// 普通查询
zkc.getData().forPath(path);
// 包含状态查询
Stat stat = new Stat();
zkc.getData().storingStatIn(stat).forPath(path);

// 普通更新，当前最新版本，会返回一个Stat实例
zkc.setData().forPath(path, "new".getBytes());
// 指定版本更新，如果版本不一致抛出异常
zkc.setData().withVersion(1).forPath(path);

// 返回一个Stat实例，用于检查ZNode是否存在的操作
zkc.checkExists().forPath(path);

List<String> children = zkc.getChildren().forPath(path);

T inBackground();
T inBackground(Object var1);
T inBackground(BackgroundCallback var1);
T inBackground(BackgroundCallback var1, Object var2);
// 允许用一个专门线程池来处理返回结果之后的业务逻辑
T inBackground(BackgroundCallback var1, Executor var2);
T inBackground(BackgroundCallback var1, Object var2, Executor var3);

zkc.inTransaction().check().forPath("/path")
                .and()
                .create().withMode(CreateMode.EPHEMERAL).forPath("/path", "data".getBytes())
                .and()
                .setData().withVersion(10).forPath("/path", "data2".getBytes())
                .and()
                .commit();

List<CuratorOp> operations = new ArrayList<CuratorOp>();
operations.add(zkc.transactionOp().check().forPath("/path"));
operations.add(zkc.transactionOp().create().withMode(CreateMode.EPHEMERAL).forPath("/path", "data".getBytes()));
operations.add(zkc.transactionOp().setData().withVersion(10).forPath("/path", "data2".getBytes()));
zkc.transaction().forOperations(operations);

byte[] content = zkc.getData().usingWatcher(new Watcher() {
	public void process(WatchedEvent watchedEvent) {
		System.out.println("监听器watchedEvent：" + watchedEvent);
	}
}).forPath(path);
System.out.println("监听节点内容：" + new String(content));

CuratorListener listener = new CuratorListener() {
	public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
		System.out.println("监听事件触发，event内容为：" + event);
		try {
			switch (event.getType()) {
				case CHILDREN:
					// ...
					break;
				case CREATE:
					// ...
					break;
				case DELETE:
					// ...
					break;
				case WATCHED:
					// ...
					break;
			}
		} catch (Exception e) {
			// LOGGER.error
		}
	}
};
// 注册监听器
zkc.getCuratorListenable().addListener(listener);

UnhandledErrorListener errorListener = new UnhandledErrorListener() {
	public void unhandledError(String message, Throwable e) {
		// LOGGER.error      
	}
};
zkc.getUnhandledErrorListenable().addListener(errorListener);

public NodeCache(CuratorFramework client, String path);
public NodeCache(CuratorFramework client, String path, boolean dataIsCompressed);

final NodeCache nodeCache = new NodeCache(zkc, path);
// start方法有一个带Boolean参数的方法，如果设置为true则在首次启动时就会缓存节点内容到Cache中
nodeCache.start();
nodeCache.getListenable().addListener(new NodeCacheListener() {
	public void nodeChanged() throws Exception {
		System.out.println("重新获得节点内容为：" + new String(nodeCache.getCurrentData().getData()));
	}
});

PathChildrenCache pathChildrenCache = new PathChildrenCache(zkc, parentPath, true);
pathChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
	public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
		System.out.println("事件类型："  + event.getType() + "；操作节点：" + event.getData().getPath());
	}
});

List<LeaderLatch> latchList = new ArrayList();
List<CuratorFramework> clients = new ArrayList();
String path = "/master";
try {
	for (int i = 0; i < 5; i++) {
		final CuratorFramework client = connect();
		clients.add(client);
		String clientId = "client#" + i;
		final LeaderLatch leaderLatch = new LeaderLatch(client, path, clientId);
		leaderLatch.addListener(new LeaderLatchListener() {
			public void isLeader() {
				System.out.println(leaderLatch.getId() +  ":I am leader. I am doing jobs!");
				try {
					Thread.sleep(1000 * 2);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				// 主动close释放领导权
				client.close();
			}
			public void notLeader() {
				System.out.println(leaderLatch.getId() +  ":I am not leader. I will do nothing!");
			}
		});
		leaderLatch.start();
	}
	Thread.sleep(Integer.MAX_VALUE);
} catch (Exception e) {
	e.printStackTrace();
} finally {
	for(CuratorFramework client : clients){
		CloseableUtils.closeQuietly(client);
	}

	for(LeaderLatch leaderLatch : latchList){
		CloseableUtils.closeQuietly(leaderLatch);
	}
}

List<LeaderSelector> selectors = new ArrayList();
List<CuratorFramework> clients = new ArrayList();
String path = "/master";
try {
	for (int i = 0; i < 5; i++) {
		CuratorFramework client = connect();
		clients.add(client);

		final String name = "client#" + i;
		LeaderSelector leaderSelector = new LeaderSelector(client, path, new LeaderSelectorListener() {
			public void takeLeadership(CuratorFramework client) throws Exception {
				System.out.println(name + ":I am leader.");
				Thread.sleep(2 * 1000);
			}
			public void stateChanged(CuratorFramework client, ConnectionState newState) {
				switch (newState) {
					case CONNECTED:
						// ...
						break;
					case SUSPENDED:
						// ...
						break;
					case LOST:
						// ...
				}
			}
		});
		leaderSelector.autoRequeue();
		leaderSelector.start();
		selectors.add(leaderSelector);
	}
	Thread.sleep(Integer.MAX_VALUE);
} catch (Exception e) {
	e.printStackTrace();
} finally {
	for(CuratorFramework client : clients){
		CloseableUtils.closeQuietly(client);
	}

	for(LeaderSelector selector : selectors){
		CloseableUtils.closeQuietly(selector);
	}
}

InterProcessMutex interProcessMutex = new InterProcessMutex(zkc, path);

interProcessMutex.acquire();
// interProcessMutex.acquire(timeout, unit);

interProcessMutex.release();