安装
关于zookeeper的安装，请参考这篇文章：ZooKeeper伪分布式集群安装及使用
我在这里使用的是文中提到的zookeeper伪分布模式的安装和使用。
命令行
下面说明命令行方式下使用zookeeper。
切换到zookeeper安装目录的bin目录下：
命令：zkServer.sh start zoo0.cfg
该命令启动zkServer服务器。我这里使用的是伪分布模式，三个服务器都是本机，只不过采用不同的端口号而已，并配置在不同的配置文件中，即zoo0.cfg、zoo1.cfg、zoo2.cfg这三个文件中配置了相应的参数和端口号。
启动了三个服务器之后，查看三个服务器的状态：
命令：zkServer.sh status zoo0.cfg
三个服务器的状态有两种，一种跟随状态，一种处于领导状态。
服务器启动之后，可以通过客户端进行连接，如下：
命令：zkCli.sh -server centos:2181
其中centos是我的主机名，2181是配置文件中配置的客户端连接服务器的端口号，可以在zoo0.cfg文件中查看到这个配置。相应的，按照我的伪分布配置，zkCli.sh -server centos:2182 这个命令可以连接到我的第二个服务器，该配置可以在zoo1.cfg文件中。大家在使用该命令时，需要按照自己的配置相应的更改一下即可。
通过上面的日志，Welcom to ZooKeeper！输出说明客户端连接服务器成功！！
输入help命令：
通过该命令，我们可以查看到客户端可以使用的命令行有哪些！！！
其实也不多，就这么几个。
命令：
ls /
stat /zookeeper
ls /命令查看根节点下的节点有哪些。可以看到只有zookeeper一个节点。
stat /zookeeper命令查看节点zookeeper节点的状态信息。
cZxid表示创建该节点时候的zxid，
mZxid表示当前的，
zxid是用来为选举leader服务的。
ctime和mtime，前者表示创建的时间，后者表示最近一次更新的时间。
要注意的是ctime就定死了创建的那一刻，而mtime会在你调用更改节点函数的时候重新设置；
但是exists和get是不会引起其更新的。
命令：get /nodename
命令get /nodename获取节点路径的数据。注意：path必须是绝对路径，也就是path必须是/开头的路径。
刚开始节点没有保存数据，所以，这里的三个节点下均无数据。
获取子节点quota信息：
设置数据命令：
set /nodepath 数据信息
图中设置根节点的数据为字符串“longyin”,然后命令获取：get /
可以看到dataLength的数值也跟着变化了。
创建节点：create /path data
创建节点node，然后通过ls输出根节点下的子节点有两个。
获取数据并删除节点,命令：
get /node
delete /node
退出客户端，命令：
quit
- java API 接口操作
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 BasicDemo1 {
public static void main(String[] args)
throws IOException, KeeperException, InterruptedException {
/**
* 创建一个与服务器的连接
* 参数一：服务器地址和端口号（该端口号值服务器允许客户端连接的端口号，配置文件的默认端口号2181）
* 参数二：连接会话超时时间
* 参数三：观察者，连接成功会触发该观察者。不过只会触发一次。
* 该Watcher会获取各种事件的通知
*/
ZooKeeper zk = new ZooKeeper("centos:2181", 60000, new Watcher() {
// 监控所有被触发的事件
public void process(WatchedEvent event) {
System.out.println("监控所有被触发的事件:EVENT:" + event.getType());
}
});
System.out.println("*******************************************************");
// 查看根节点的子节点
System.out.println("查看根节点的子节点:ls / => " + zk.getChildren("/", true));
System.out.println("*******************************************************");
// 创建一个目录节点
if (zk.exists("/node", true) == null) {
/**
* 参数一：路径地址
* 参数二：想要保存的数据，需要转换成字节数组
* 参数三：ACL访问控制列表（Access control list）,
* 参数类型为ArrayList<ACL>，Ids接口提供了一些默认的值可以调用。
* OPEN_ACL_UNSAFE This is a completely open ACL
* 这是一个完全开放的ACL，不安全
* CREATOR_ALL_ACL This ACL gives the
* creators authentication id's all permissions.
* 这个ACL赋予那些授权了的用户具备权限
* READ_ACL_UNSAFE This ACL gives the world the ability to read.
* 这个ACL赋予用户读的权限，也就是获取数据之类的权限。
* 参数四：创建的节点类型。枚举值CreateMode
* PERSISTENT (0, false, false)
* PERSISTENT_SEQUENTIAL (2, false, true)
* 这两个类型创建的都是持久型类型节点，回话结束之后不会自动删除。
* 区别在于，第二个类型所创建的节点名后会有一个单调递增的数值
* EPHEMERAL (1, true, false)
* EPHEMERAL_SEQUENTIAL (3, true, true)
* 这两个类型所创建的是临时型类型节点，在回话结束之后，自动删除。
* 区别在于，第二个类型所创建的临时型节点名后面会有一个单调递增的数值。
* 最后create()方法的返回值是创建的节点的实际路径
*/
zk.create("/node", "conan".getBytes(),
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
System.out.println("创建一个目录节点:create /node conan");
/**
* 查看/node节点数据，这里应该输出"conan"
* 参数一：获取节点的路径
* 参数二：说明是否需要观察该节点，设置为true，则设定共享默认的观察器
* 参数三：stat类，保存节点的信息。例如数据版本信息，创建时间，修改时间等信息
*/
System.out.println("查看/node节点数据:get /node => "
+ new String(zk.getData("/node", false, null)));
/**
* 查看根节点
* 在此查看根节点的值，这里应该输出上面所创建的/node节点
*/
System.out.println("查看根节点:ls / => " + zk.getChildren("/", true));
}
System.out.println("*******************************************************");
// 创建一个子目录节点
if (zk.exists("/node/sub1", true) == null) {
zk.create("/node/sub1", "sub1".getBytes(),
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
System.out.println("创建一个子目录节点:create /node/sub1 sub1");
// 查看node节点
System.out.println("查看node节点:ls /node => "
+ zk.getChildren("/node", true));
}
System.out.println("*******************************************************");
/**
* 修改节点数据
* 修改的数据会覆盖上次所设置的数据
* setData()方法参数一、参数二不多说，与上面类似。
* 参数三：数值型。需要传入该界面的数值类型版本号！！！
* 该信息可以通过Stat类获取，也可以通过命令行获取。
* 如果该值设置为-1，就是忽视版本匹配，直接设置节点保存的值。
*/
if (zk.exists("/node", true) != null) {
zk.setData("/node", "changed".getBytes(), -1);
// 查看/node节点数据
System.out.println("修改节点数据:get /node => "
+ new String(zk.getData("/node", false, null)));
}
System.out.println("*******************************************************");
// 删除节点
if (zk.exists("/node/sub1", true) != null) {
zk.delete("/node/sub1", -1);
zk.delete("/node", -1);
// 查看根节点
System.out.println("删除节点:ls / => " + zk.getChildren("/", true));
}
// 关闭连接
zk.close();
}
}代码中大部分关键代码已经给出了详细的注释。代码完成的功能也和命令行相同。首先启动zookeeper，然后运行代码：
结果如下：
可以看到，结果和我们使用命令行的操作一样的效果。代码中我们可以使用观察者，接收各种事件。
最后再给出一个小示例代码：
import java.io.IOException;
import java.util.ArrayList;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
public class QueueZooKeeper {
public static void main(String[] args) throws Exception {
if (args.length == 0) {
doOne();
} else {
doAction(Integer.parseInt(args[0]));
}
}
public static void doOne() throws Exception {
String host1 = "centos:2181";
ZooKeeper zk = connection(host1);
initQueue(zk);
joinQueue(zk, 1);
joinQueue(zk, 2);
joinQueue(zk, 3);
zk.close();
}
public static void doAction(int client) throws Exception {
String host1 = "centos:2181";
String host2 = "centos:2182";
String host3 = "centos:2183";
ZooKeeper zk = null;
switch (client) {
case 1:
zk = connection(host1);
initQueue(zk);
joinQueue(zk, 1);
break;
case 2:
zk = connection(host2);
initQueue(zk);
joinQueue(zk, 2);
break;
case 3:
zk = connection(host3);
initQueue(zk);
joinQueue(zk, 3);
break;
}
}
// 创建一个与服务器的连接
public static ZooKeeper connection(String host) throws IOException {
ZooKeeper zk = new ZooKeeper(host, 60000, new Watcher() {
// 监控所有被触发的事件
public void process(WatchedEvent event) {
if (event.getType() == Event.EventType.NodeCreated
&& event.getPath().equals("/queue/start")) {
System.out.println("Queue has Completed.Finish testing!!!");
}
}
});
return zk;
}
public static void initQueue(ZooKeeper zk)
throws KeeperException, InterruptedException {
System.out.println("WATCH => /queue/start");
// zk.exists("/queue/start", true);
/**
* 创建节点/queue.永久节点
* 存在则不创建，如果不存在，则创建
*/
if (zk.exists("/queue", false) == null) {
System.out.println("create=> /queue task-queue");
zk.create("/queue", "task-queue".getBytes(),
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
} else {
System.out.println("/queue is exist!");
}
}
/**
* 创建节点/queue的子节点，（临时节点）,在会话退出之后，临时节点删除。并且临时
* 节点有顺序号。
* @param zk
* @param x
* @throws KeeperException
* @throws InterruptedException
*/
public static void joinQueue(ZooKeeper zk, int x)
throws KeeperException, InterruptedException {
System.out.println("create=> /queue/x" + x + " x" + x);
zk.create("/queue/x" + x, ("x" + x).getBytes(),
Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
isCompleted(zk);
}
public static void isCompleted(ZooKeeper zk)
throws KeeperException, InterruptedException {
int size = 3;
ArrayList<String> list = (ArrayList<String>)
zk.getChildren("/queue", true);
for (String str:list) {
System.out.println("获取节点queue的子节点：ls /queue:"+str);
}
int length = list.size();
System.out.println("Queue Complete:" + length + "/" + size+"(子节点个数/总长度)");
if (length >= size) {
System.out.println("创建临时型节点：create /queue/start start");
zk.create("/queue/start", "start".getBytes(),
Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
}
}
}运行结果如下：
从上面的结果可以看出，我们创建临时型节点的时候，会话结束，临时节点自动删除。我们使用命令行查看一下根节点的信息：
代码中创建的节点都没有了。
另一个要注意的是，我们使用的节点类型是临时型的有顺序号的节点类型，我们发现，会话期间，我们打印的节点名后面有000000014，000000015，000000016，由于我运行了几次代码，所以这个值在单条递增，增加到了16，从这里我们应该知道了顺序号类型的节点的意思了。
OK，结束！
本项目的代码地址：请猛戳这里（欢迎关注我的GITHUB）
项目使用eclipse构建。方便易用，代码注释详细。
6007
被折叠的 条评论
为什么被折叠?
到【灌水乐园】发言
