zookeeper配置中心 -ag凯发k8国际

场景

多个客户端从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;} }

总结

以上是ag凯发k8国际为你收集整理的zookeeper配置中心的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得ag凯发k8国际网站内容还不错，欢迎将ag凯发k8国际推荐给好友。