zookeeper配置中心

场景

多个客户端从zookeeper 的配置中心拿到配置，如果配置中心没有配置就阻塞，如果配置修改了就拿到新的。

原理

因为zookeeper的同步性质，即单线程的分两段式（当发生修改时，第一阶段：leader向各个follower发送log任务，过半成功返回后进行第二段的具体的修改）的事务方式，如果节点发生更改，则要么成功且所有客户端都get到最新的修改结果，要么失败。所以代码只要注册发生修改（包括数据修改，节点增加，删除等，以下同）的事件，死循环取就行。

本文只写了简单的框架，复杂的业务逻辑没有。

用zookeeper模拟另外的客户端做修改的操作。

代码

目录结构

package org.faithgreen.conf;import org.apache.zookeeper.ZooKeeper; import org.junit.After; import org.junit.Before; import org.junit.Test;/*** 模拟 zookeeper 获取配置的操作*/ public class Main {ZooKeeper zk;@Beforepublic void before() {zk = ZkUtils.getZK();}@Afterpublic void after() {try {zk.close();} catch (InterruptedException e) {e.printStackTrace();}}/*** 核心是 watcher 和回调 callback 的处理*/@Testpublic void getConf() {WatchCallBack w = new WatchCallBack();MyConf conf = new MyConf();w.setConf(conf);w.setZk(zk);w.await();while (true) {if (conf.getConfStr().equals("")) {// 如果取不到，就要阻塞等到取到System.out.println("配置丢了 ....");w.await();} else {System.out.println("conf: " + conf.getConfStr());}}} } package org.faithgreen.conf;import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher;import java.util.concurrent.CountDownLatch;/*** 默认基于 session 的事件，只用来阻塞主线程，让 zookeeper 连接成功后再继续运行*/ public class DefaultWatcher implements Watcher {CountDownLatch c;public void setC(CountDownLatch c) {this.c = c;}@Overridepublic void process(WatchedEvent e) {Event.EventType type = e.getType();Event.KeeperState state = e.getState();String path = e.getPath();switch (type) {case None:break;case NodeCreated:break;case NodeDeleted:break;case NodeDataChanged:break;case NodeChildrenChanged:break;}switch (state) {case Unknown:break;case Disconnected:break;case NoSyncConnected:break;case SyncConnected:c.countDown();break;case AuthFailed:break;case ConnectedReadOnly:break;case SaslAuthenticated:break;case Expired:break;}} } package org.faithgreen.conf;/*** 模拟配置中心数据，zookeeper最大 1M* 比如是xml文件，需要自己实现io流和编解码的逻辑*/ public class MyConf {private String confStr;public String getConfStr() {return confStr;}public void setConfStr(String confStr) {this.confStr = confStr;} } package org.faithgreen.conf;import org.apache.zookeeper.AsyncCallback; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.Stat;import java.util.concurrent.CountDownLatch;/*** 注册了节点的事件和回调*/ public class WatchCallBack implements Watcher, AsyncCallback.DataCallback, AsyncCallback.StatCallback {ZooKeeper zk;CountDownLatch cc = new CountDownLatch(1);MyConf conf;public ZooKeeper getZk() {return zk;}public void setZk(ZooKeeper zk) {this.zk = zk;}public CountDownLatch getCc() {return cc;}public void setCc(CountDownLatch cc) {this.cc = cc;}public MyConf getConf() {return conf;}public void setConf(MyConf conf) {this.conf = conf;}public void await() {zk.exists("/appConf", this, this, "abc");try {// 让主线程阻塞住，上面那行代码运行下去cc.await();} catch (InterruptedException e) {e.printStackTrace();}}/*** Watcher 的接口实现** @param e event*/@Overridepublic void process(WatchedEvent e) {String path = e.getPath();Event.EventType type = e.getType();Event.KeeperState state = e.getState();switch (type) {case None:break;case NodeCreated:// 如果节点被创建，就获取它，让它调用回调，给conf设值zk.getData("/appConf", this, this, "def");break;case NodeDeleted:// 如果节点被删除了，考虑到容忍性的// 配置置空conf.setConfStr("");// 重新枷锁cc = new CountDownLatch(1);break;case NodeDataChanged:// 如果节点被改了，就重新获取,目的是让他调用回调函数，给conf设置新的值zk.getData("/appConf", this, this, "def");break;case NodeChildrenChanged:break;}}/*** DataCallBack 的实现** @param i 版本* @param s path* @param o o* @param bytes data* @param stat stat*/@Overridepublic void processResult(int i, String s, Object o, byte[] bytes, Stat stat) {if (bytes != null) {String s1 = new String(bytes);conf.setConfStr(s1);// 取到数据了，让主线程继续往下走cc.countDown();}}/*** StatCallBack 的实现** @param i 版本* @param s path* @param o o* @param stat stat*/@Overridepublic void processResult(int i, String s, Object o, Stat stat) {if (stat != null) {zk.getData("/appConf", this, this, "def");}} } package org.faithgreen.conf;import org.apache.zookeeper.ZooKeeper;import java.util.concurrent.CountDownLatch;public class ZkUtils {static ZooKeeper zk;final static String address = "192.168.172.3:2181,192.168.172.4:2181,192.168.172.5:2181,192.168.172.6:2181/testConf";final static DefaultWatcher defaultWatcher = new DefaultWatcher();static CountDownLatch c = new CountDownLatch(1);public static ZooKeeper getZK() {try {zk = new ZooKeeper(address, 4000, defaultWatcher);defaultWatcher.setC(c);c.await();} catch (Exception e) {e.printStackTrace();}return zk;} }

总结

以上是生活随笔为你收集整理的zookeeper配置中心的全部内容，希望文章能够帮你解决所遇到的问题。

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