Zookeeper_实际应用讲解

我告诉你zookeeper一般是做什么事的,这里会有一个管理工具,到底怎么去用呢,eclipse和工具去集成

输入zookeeperBrower这个名字,输入http://www.massedynamic.org/eclipse/updates/网址,无非就是把这个插件装上,Help里面有一个eclipsemarket,肯定要求输入名字和地址,安装这个插件,安装上了就可以用这个zookeeper explorer,这个名字其实你可以随便起的,但是地址你不能随便弄,你如果想在eclipse上去观察的话你可以用这个,我也会发给你第三方的工具,咱们看一下cluster,这是随便模拟的一个例子,他并不是做一个很真实的场景

如果有一个实际的场景的话,我希望对分布式系统配置,进行集中地管理,我可能是集群,我有好多的服务,比如开始连的是同样的数据库,都要连zookeeper,一下子变化了,zookeeper地址一下子变化了,怎么办,原先zookeeper在1.1上,1.1,1.2,1.3,现在整体变成1.4,1.5,1.6了,3台无所谓,如果是30台,300台呢,你可能一个机器一个机器的去改太麻烦了,这个时候就可以使用watcher来做这个事情,去观察,动态的去修改,这个后期再去讲实际的案例吧,我现在有一个非常简单的DEMO,首先我有连个client端,client1和client2,你可以把这两个哥们当成两台机器,这两台机器在工作中怎么去用呢,项目启动的时候,我都要watcher启动起来,然后我就一直在这里等着,就是把项目启动了,我就直接运行,第一台机器连上了就是运行的,第二台机器也是这样的,首先每一个机器都是watcher,分别有两台机器去watcher了,是什么节点呢,你可以去看一下,这两个client都同时实例化一个类,以后在工作中你可以把watcher写在一个common里面,也可以写在每一个项目中,这都是行的 package com.learn.zookeeper.cluster;import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch;import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.Watcher.Event.EventType; import org.apache.zookeeper.Watcher.Event.KeeperState; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooKeeper;/*** 这是好久之前模拟的* 代码写的也比较low* 主要是让你看到这个效果* 以后用到一些Curator框架的时候* 那才是最好的* 自己刚学会CopyOnWriteArrayList这种集合* 像这种节点频繁变更的* 不应该采用这种list* 你宁可采用ArrayList* 或者并发的List* 所以你千万别使用这个* 你就当看一个逻辑* 而不是看代码怎么去实现的* 我要你知道子节点变更的时候是有多麻烦* 我要知道子节点到底发生了什么样的变化* 原生的API写的时候是比较麻烦的* 你看一下process就知道了* process的时候我做了一堆事* 我自己都不知道怎么去写的* 以后我们就用Curator会是更好的方式* 主要讲的是一个场景* 我们在真正的工作中有好几台服务器* 都要公用一份config文件* 那么我就可以用zookeeper去做这个事情* * * * @author Leon.Sun**/ public class ZKWatcher implements Watcher {/** zk变量 */private ZooKeeper zk = null;/** 父节点path *//*** 有一个/super节点* super节点它是key* 默认的时候是先去get /super* 然后把IP地址获得* 获取是其他的服务器* 咱们举个例子* 一旦有super节点发生update的时候* 谁watcher了我就给谁发update* Node Change这个事件* 直接把新的value给更新上了* 更新上了就做一个集体的切换* 很多时候工作就是做这个事* 这是最简单的一种实现* * */static final String PARENT_PATH = "/super";/** 信号量设置，用于等待zookeeper连接建立之后 通知阻塞程序继续向下执行 */private CountDownLatch connectedSemaphore = new CountDownLatch(1);private List<String> cowaList = new CopyOnWriteArrayList<String>();/** zookeeper服务器地址 */ // public static final String CONNECTION_ADDR = "192.168.80.88:2181,192.168.80.87:2181,192.168.80.86:2181";public static final String CONNECTION_ADDR = "59.110.138.145:2181";/** 定义session失效时间 */public static final int SESSION_TIMEOUT = 30000;public ZKWatcher() throws Exception{zk = new ZooKeeper(CONNECTION_ADDR, SESSION_TIMEOUT, this);System.out.println("开始连接ZK服务器");connectedSemaphore.await();}@Overridepublic void process(WatchedEvent event) {// 连接状态KeeperState keeperState = event.getState();// 事件类型EventType eventType = event.getType();// 受影响的pathString path = event.getPath();System.out.println("受影响的path : " + path);if (KeeperState.SyncConnected == keeperState) {// 成功连接上ZK服务器if (EventType.None == eventType) {System.out.println("成功连接上ZK服务器");connectedSemaphore.countDown();try {if(this.zk.exists(PARENT_PATH, false) == null){this.zk.create(PARENT_PATH, "root".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); }List<String> paths = this.zk.getChildren(PARENT_PATH, true);for (String p : paths) {System.out.println(p);this.zk.exists(PARENT_PATH + "/" + p, true);}} catch (KeeperException | InterruptedException e) {e.printStackTrace();} } //创建节点else if (EventType.NodeCreated == eventType) {System.out.println("节点创建");try {this.zk.exists(path, true);} catch (KeeperException | InterruptedException e) {e.printStackTrace();}} //更新节点else if (EventType.NodeDataChanged == eventType) {System.out.println("节点数据更新");try {//update nodes call functionthis.zk.exists(path, true);} catch (KeeperException | InterruptedException e) {e.printStackTrace();}} //更新子节点else if (EventType.NodeChildrenChanged == eventType) {System.out.println("子节点 ... 变更");try {List<String> paths = this.zk.getChildren(path, true);if(paths.size() >= cowaList.size()){paths.removeAll(cowaList);for(String p : paths){this.zk.exists(path + "/" + p, true);//this.zk.getChildren(path + "/" + p, true);System.out.println("这个是新增的子节点 : " + path + "/" + p);//add new nodes call function}cowaList.addAll(paths);} else {cowaList = paths;}System.out.println("cowaList: " + cowaList.toString());System.out.println("paths: " + paths.toString());} catch (KeeperException | InterruptedException e) {e.printStackTrace();}} //删除节点else if (EventType.NodeDeleted == eventType) {System.out.println("节点 " + path + " 被删除");try {//delete nodes call functionthis.zk.exists(path, true);} catch (KeeperException | InterruptedException e) {e.printStackTrace();}}else ;} else if (KeeperState.Disconnected == keeperState) {System.out.println("与ZK服务器断开连接");} else if (KeeperState.AuthFailed == keeperState) {System.out.println("权限检查失败");} else if (KeeperState.Expired == keeperState) {System.out.println("会话失效");}else ;System.out.println("--------------------------------------------");}} package com.learn.zookeeper.cluster;public class Client1 {public static void main(String[] args) throws Exception{ZKWatcher myWatcher = new ZKWatcher();Thread.sleep(100000000);} } package com.learn.zookeeper.cluster;public class Client2 {public static void main(String[] args) throws Exception{ZKWatcher myWatcher = new ZKWatcher();Thread.sleep(100000000);} } package com.learn.zookeeper.cluster;import java.util.concurrent.CountDownLatch;import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.Watcher.Event.EventType; import org.apache.zookeeper.Watcher.Event.KeeperState; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooKeeper;/*** 机器1和机器2是一个集群的方式* 以后要连同样的一份代码* 要连redis集群* 你再开发一套应用* 这个应用就是用来管理zookeeper上的节点的* 我这块添加了一个节点了* 这两台机器应该马上感知到* 我再添加删除* 然后这里面的业务逻辑变更* 其实就是这么一个道理* 我把test运行一下吧* 运行完了之后Test就停止了* * * @author Leon.Sun**/ public class Test {/** zookeeper地址 */ // static final String CONNECT_ADDR = "192.168.1.106:2181,192.168.1.107:2181,192.168.1.108:2181";/*** 这个Test无论做什么事* 这两个client都能感觉到的* 比如我这个Test做什么事* * */static final String CONNECT_ADDR = "59.110.138.145:2181";/** session超时时间 */static final int SESSION_OUTTIME = 2000;//ms /** 信号量，阻塞程序执行，用于等待zookeeper连接成功，发送成功信号 */static final CountDownLatch connectedSemaphore = new CountDownLatch(1);public static void main(String[] args) throws Exception{/*** new了一个zookeeper* 这个Watcher就做了一个很简单的操作* 连接上了就OK了* */ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, new Watcher(){@Overridepublic void process(WatchedEvent event) {//获取事件的状态KeeperState keeperState = event.getState();EventType eventType = event.getType();//如果是建立连接if(KeeperState.SyncConnected == keeperState){if(EventType.None == eventType){//如果建立连接成功，则发送信号量，让后续阻塞程序向下执行connectedSemaphore.countDown();System.out.println("zk 建立连接");}}}});//进行阻塞connectedSemaphore.await();// //创建子节点/*** 创建了4个节点* c1,c2,c3,c4*/zk.create("/super/c1", "c1".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);//创建子节点zk.create("/super/c2", "c2".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);//创建子节点zk.create("/super/c3", "c3".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);//创建子节点zk.create("/super/c4", "c4".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);// zk.create("/super/c4/c44", "c44".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);//获取节点信息 // byte[] data = zk.getData("/testRoot", false, null); // System.out.println(new String(data)); // System.out.println(zk.getChildren("/testRoot", false));//修改节点的值/*** 修改了c1和c2*/zk.setData("/super/c1", "modify c1".getBytes(), -1);zk.setData("/super/c2", "modify c2".getBytes(), -1);byte[] data = zk.getData("/super/c2", false, null);System.out.println(new String(data)); // //判断节点是否存在 // System.out.println(zk.exists("/super/c3", false)); // //删除节点/*** 删除了c3* 其实就是这么简单的操作* 删除c3了* 就只剩下c1,c2,c4了* 我做完这个操作之后* client1和client2都应该能感知到才行* 都是触发了这些东西* client1是这样* client2也是这样* 没有任何变化* 这两个哥们C1和c2收到的信息是一样的* 可以这么去理解* 这两个都监听到了* 就是这个意思* 理解这个事就行了* 很多块时候都需要做这个事情* 降低重复工作* * * */zk.delete("/super/c3", -1);zk.close();} }

 

总结

以上是生活随笔为你收集整理的Zookeeper_实际应用讲解的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。