zookeeper集群搭建及java程序测试
下载官网：http://zookeeper.apache.org/releases.html 下载：zookeeper-3.4.8.tar.gz 安装因为资源有限，所以我在同一个服务器上面创建3个目录 server1、server2、server3 来模拟3台服务器集群。 cd server1 tar -zxvf zookeeper-3.4.8.tar.gz mkdir data mkdir dataLog data 为数据目录，dataLog 为日志目录。 配置 zookeeper-3.4.8/conf 创建文件 zoo.cfg，内容如下： tickTime=2000 initLimit=5syncLimit=2dataDir=/opt/zookeeper/server1/datadataLogDir=/opt/zookeeper/server1/dataLogclientPort=2181server.1=127.0.0.1:2888:3888server.2=127.0.0.1:2889:3889server.3=127.0.0.1:2890:3890 tickTime：zookeeper中使用的基本时间单位, 毫秒值。 initLimit：这个配置项是用来配置 Zookeeper 接受客户端（这里所说的客户端不是用户连接 Zookeeper 服务器的客户端，而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器）初始化连接时最长能忍受多少个 tickTime 时间间隔数。这里设置为5表名最长容忍时间为 5 * 2000 = 10 秒。 syncLimit：这个配置标识 Leader 与 Follower 之间发送消息，请求和应答时间长度，最长不能超过多少个 tickTime 的时间长度，总的时间长度就是 2 * 2000 = 4 秒。 dataDir 和 dataLogDir 看配置就知道干吗的了，不用解释。 clientPort：监听client连接的端口号，这里说的client就是连接到Zookeeper的代码程序。 server.{myid}={ip}:{leader服务器交换信息的端口}:{当leader服务器挂了后, 选举leader的端口} maxClientCnxns：对于一个客户端的连接数限制，默认是60，这在大部分时候是足够了。但是在我们实际使用中发现，在测试环境经常超过这个数，经过调查发现有的团队将几十个应用全部部署到一台机器上，以方便测试，于是这个数字就超过了。 修改zoo.cfg非常简单，然后还需要创建myid文件： cd server1 echo 1 > myid 然后拷贝server1为server2和server3，并修改其中的zoo.cfg配置，当然也要修改myid的内容为2和3。 下面给出3个server的zoo.cfg 内容： server1 tickTime=2000initLimit=5syncLimit=2dataDir=/opt/zookeeper/server1/datadataLogDir=/opt/zookeeper/server1/dataLogclientPort=2181server.1=127.0.0.1:2888:3888server.2=127.0.0.1:2889:3889server.3=127.0.0.1:2890:3890 server2 tickTime=2000initLimit=5syncLimit=2dataDir=/opt/zookeeper/server2/datadataLogDir=/opt/zookeeper/server2/dataLogclientPort=2182server.1=127.0.0.1:2888:3888server.2=127.0.0.1:2889:3889server.3=127.0.0.1:2890:3890 server3 tickTime=2000initLimit=5syncLimit=2dataDir=/opt/zookeeper/server3/datadataLogDir=/opt/zookeeper/server3/dataLogclientPort=2183server.1=127.0.0.1:2888:3888server.2=127.0.0.1:2889:3889server.3=127.0.0.1:2890:3890 这里做下说明：因为我们是在同一个机器上模拟的集群，所以要注意server端口号和clientPort不要重复了，不然会出现端口冲突。所以，如果我们是3个不同的机器上做的3个server，那么我们的zoo.cfg配置都是一样的（注意server.{myid}=后面的IP地址使用具体的IP地址，如192.168.0.88）。还有就是，每一个server的myid内容都不能一样，这也可以理解为不同server的标识。 启动进入 zookeeper-3.4.8/bin 目录，使用 ./zkServer.sh start 启动zk服务。（你也可以使用 ./zkServer.sh start myzoo.cfg 指定配置文件启动，这在自动化运维的时候很有用） 使用 tail -f zookeeper.out 查看日志。 要说的是：在启动第一个的时候，日志中会出现一堆错误，仔细一看就能明白，是因为另外2个server还没有启动它连接不上的错误。然后当我们启动第二个server的时候，日志中的错误将会减少。最后我们把所有server都启动起来后，日志中便没有错误了。 测试随便进入一个zk目录，连接一个server测试。 cd zookeeper-3.4.8/bin zkCli.sh -server 127.0.0.1:2181 如果你要连接别的服务器，请指定具体的IP地址。 几个基本命令说明： ls 查看指定节点中包含的子节点（如：ls / 或 ls /app1/server1） create 创建节点并赋值 get 读取节点内容 set 改变节点内容 delete 删除节点 注意zk中所有节点都基于路径确定，如你要删除 /app1/server1/nodeA 的命令为： delete /app1/server1/nodeA 下面是基本操作截图： 这里写图片描述 Java程序Demo创建一个Maven工程 打开pom.xml文件添加zookeeper依赖 <dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.4.6</version> </dependency> 创建Demo.java，代码如下： package com.shanhy.demo.zookeeper; 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; /** Zookeeper测试* @create 2018年7月10日*/ public class Test { // 会话超时时间，设置为与系统默认时间一致 private static final int SESSION_TIMEOUT = 30 * 1000; // 创建 ZooKeeper 实例 private ZooKeeper zk; // 创建 Watcher 实例 private Watcher wh = new Watcher() { /** * Watched事件 */ public void process(WatchedEvent event) { System.out.println("WatchedEvent >>> " + event.toString()); } }; // 初始化 ZooKeeper 实例 private void createZKInstance() throws IOException { // 连接到ZK服务，多个可以用逗号分割写 zk = new ZooKeeper("192.168.19.130:2181,192.168.19.130:2182,192.168.19.130:2183", Test.SESSION_TIMEOUT, this.wh); } private void ZKOperations() throws IOException, InterruptedException, KeeperException { System.out.println("\n1. 创建 ZooKeeper 节点 (znode ： zoo2, 数据： myData2 ，权限： OPEN_ACL_UNSAFE ，节点类型： Persistent"); zk.create("/zoo2", "myData2".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println("\n2. 查看是否创建成功： "); System.out.println(new String(zk.getData("/zoo2", this.wh, null)));// 添加Watch // 前面一行我们添加了对/zoo2节点的监视，所以这里对/zoo2进行修改的时候，会触发Watch事件。 System.out.println("\n3. 修改节点数据 "); zk.setData("/zoo2", "shanhy20160310".getBytes(), -1); // 这里再次进行修改，则不会触发Watch事件，这就是我们验证ZK的一个特性“一次性触发”，也就是说设置一次监视，只会对下次操作起一次作用。 System.out.println("\n3-1. 再次修改节点数据 "); zk.setData("/zoo2", "shanhy20160310-ABCD".getBytes(), -1); System.out.println("\n4. 查看是否修改成功： "); System.out.println(new String(zk.getData("/zoo2", false, null))); System.out.println("\n5. 删除节点 "); zk.delete("/zoo2", -1); System.out.println("\n6. 查看节点是否被删除： "); System.out.println(" 节点状态： [" + zk.exists("/zoo2", false) + "]"); } private void ZKClose() throws InterruptedException { zk.close(); } public static void main(String[] args) throws IOException, InterruptedException, KeeperException { Test dm = new Test(); dm.createZKInstance(); dm.ZKOperations(); dm.ZKClose(); } } 我想代码不用解释了，该注释的里面都注释了。 下面有一种特殊的情况的处理思路： 有server1、server2、server3这三个服务，在client去连接zk的时候，指向server1初始化的过程中是没有问题的，然而刚刚初始化完成，准备去连接server1的时候，server1因为网络等原因挂掉了。 然而对client来说，它会拿server1的配置去请求连接，这时肯定会报连接被拒绝的异常以致启动退出。 所以优雅的解决这个问题的方法思路就是“在连接的时候判断连接状态，如果未连接成功，程序自动使用其他连接去请求连接”，这样来避开这种罕见的异常问题。 代码如下： // 初始化 ZooKeeper 实例 private void createZKInstance() throws IOException { // 连接到ZK服务，多个可以用逗号分割写 zk = new ZooKeeper("192.168.19.130:2181,192.168.19.130:2182,192.168.19.130:2183", Test.SESSION_TIMEOUT, this.wh); if(!zk.getState().equals(States.CONNECTED)){ while(true){ if(zk.getState().equals(States.CONNECTED)){ break; } try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); } } } } 上面的代码是基于zk提供的库的API来你使用的，为了更易于使用，有人写了开源的zkclient，我们可以直接使用它来操作zk。 zkclient 开源地址：https://github.com/sgroschupf/zkclient maven 依赖配置： <!--zkclient --> <dependency> <groupId>com.101tec</groupId> <artifactId>zkclient</artifactId> <version>0.7</version> </dependency> zkClient 针对 zk 的一次性watcher，做了重新封装，然后定义了 stateChanged、znodeChanged、dataChanged 三种监听器。 监听children变化监听节点数据变化监听连接状态变化代码如下： package com.shanhy.demo.zookeeper; import java.util.List;import java.util.concurrent.TimeUnit; import org.I0Itec.zkclient.DataUpdater;import org.I0Itec.zkclient.IZkChildListener;import org.I0Itec.zkclient.IZkDataListener;import org.I0Itec.zkclient.IZkStateListener;import org.I0Itec.zkclient.ZkClient;import org.apache.zookeeper.Watcher.Event.KeeperState; /** ZkClient的使用测试 @create 2018年7月11日*/ public class ZkClientTest { public static void main(String[] args) { ZkClient zkClient = new ZkClient("192.168.19.130:2181,192.168.19.130:2182,192.168.19.130:2183"); String node = "/myapp"; // 订阅监听事件 childChangesListener(zkClient, node); dataChangesListener(zkClient, node); stateChangesListener(zkClient); if (!zkClient.exists(node)) { zkClient.createPersistent(node, "hello zookeeper"); } System.out.println(zkClient.readData(node)); zkClient.updateDataSerialized(node, new DataUpdater<String>() { public String update(String currentData) { return currentData + "-123"; } }); System.out.println(zkClient.readData(node)); try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } } /** * 订阅children变化 * * @param zkClient * @param path * @create 2018年7月11日 */ public static void childChangesListener(ZkClient zkClient, final String path) { zkClient.subscribeChildChanges(path, new IZkChildListener() { public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception { System.out.println("clildren of path " + parentPath + ":" + currentChilds); } }); } /** * 订阅节点数据变化 * * @param zkClient * @param path * @create 2018年7月11日 */ public static void dataChangesListener(ZkClient zkClient, final String path){ zkClient.subscribeDataChanges(path, new IZkDataListener(){ public void handleDataChange(String dataPath, Object data) throws Exception { System.out.println("Data of " + dataPath + " has changed."); } public void handleDataDeleted(String dataPath) throws Exception { System.out.println("Data of " + dataPath + " has changed."); } }); } /** * 订阅状态变化 * * @param zkClient * @create 2018年7月11日 */ public static void stateChangesListener(ZkClient zkClient){ zkClient.subscribeStateChanges(new IZkStateListener() { public void handleStateChanged(KeeperState state) throws Exception { System.out.println("handleStateChanged"); } public void handleSessionEstablishmentError(Throwable error) throws Exception { System.out.println("handleSessionEstablishmentError"); } public void handleNewSession() throws Exception { System.out.println("handleNewSession"); } }); } }
