Zookeeper之创建组，加入组，列出组成员和删除组

public class CreateGroup implements Watcher {

private static final int SESSION_TIMEOUT = 5000;

//ZooKeeper类是客户端API的主要类，用于维护客户端和ZooKeeper服务之间的连接
private ZooKeeper zk;
//锁存器(latch)此计数器为1，表示在释放所有等待线程之前需要发生的事件数，
private CountDownLatch connectedSignal = new CountDownLatch(1);

public void connect(String hosts) throws IOException, InterruptedException {
//参数this表示一个Watcher对象接收来自于Zookeeper的回调，以获得各种事件的通知，在此表示CreateGroup对象
zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);
connectedSignal.await();
}

@Override
public void process(WatchedEvent event) { // Watcher interface
if (event.getState() == KeeperState.SyncConnected) {
//在调用这个方法表示计数器递减1，若计数器的值变为0，则await()方法返回
connectedSignal.countDown();
}
}

public void create(String groupName) throws KeeperException,
InterruptedException {
String path = "/" + groupName;
String createdPath = zk.create(path, null/*data*/, Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);// 持久化的 znode
System.out.println("Created " + createdPath);
}

public void close() throws InterruptedException {
zk.close();
}

public static void main(String[] args) throws Exception {
CreateGroup createGroup = new CreateGroup();
createGroup.connect("192.168.8.88");
createGroup.create("zoo");
createGroup.close();
}
}

// vv ConnectionWatcher
public class ConnectionWatcher implements Watcher {

private static final int SESSION_TIMEOUT = 5000;

protected ZooKeeper zk;
private CountDownLatch connectedSignal = new CountDownLatch(1);

public void connect(String hosts) throws IOException, InterruptedException {
zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);
connectedSignal.await();
}

@Override
public void process(WatchedEvent event) {
if (event.getState() == KeeperState.SyncConnected) {
connectedSignal.countDown();
}
}

public void close() throws InterruptedException {
zk.close();
}
}
// ^^ ConnectionWatcher

// vv JoinGroup
public class JoinGroup extends ConnectionWatcher {

public void join(String groupName, String memberName) throws KeeperException,
InterruptedException {
String path = "/" + groupName + "/" + memberName;
String createdPath = zk.create(path, null/*data*/, Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
System.out.println("Created " + createdPath);
}

public static void main(String[] args) throws Exception {
JoinGroup joinGroup = new JoinGroup();
joinGroup.connect("192.168.8.88");
joinGroup.join("zoo", "goat");

// stay alive until process is killed or thread is interrupted
Thread.sleep(Long.MAX_VALUE);
}
}
// ^^ JoinGroup

// vv ListGroup
public class ListGroup extends ConnectionWatcher {

public void list(String groupName) throws KeeperException,
InterruptedException {
String path = "/" + groupName;

try {
//第二个参数若为true表示对这个父节点设置监视，应用程序可以接收组成员的加入，退出和这个父组znode被删除的有关通知
List<String> children = zk.getChildren(path, false);
if (children.isEmpty()) {
System.out.printf("No members in group %s\n", groupName);
System.exit(1);
}
for (String child : children) {
System.out.println(child);
}
} catch (KeeperException.NoNodeException e) {
System.out.printf("Group %s does not exist\n", groupName);
System.exit(1);
}
}

public static void main(String[] args) throws Exception {
ListGroup listGroup = new ListGroup();
listGroup.connect("192.168.8.88");
listGroup.list("zoo");
listGroup.close();
}
}
// ^^ ListGroup

// vv DeleteGroup
public class DeleteGroup extends ConnectionWatcher {

public void delete(String groupName) throws KeeperException,
InterruptedException {
String path = "/" + groupName;

try {
List<String> children = zk.getChildren(path, false);
for (String child : children) {
zk.delete(path + "/" + child, -1);//-1表示绕过版本检测机制，不管znode版本是什么直接将其删除
}
zk.delete(path, -1);
} catch (KeeperException.NoNodeException e) {
System.out.printf("Group %s does not exist\n", groupName);
System.exit(1);
}
}

public static void main(String[] args) throws Exception {
DeleteGroup deleteGroup = new DeleteGroup();
deleteGroup.connect("192.168.8.88");
deleteGroup.delete("zoo");
deleteGroup.close();
}
}
// ^^ DeleteGroup