場景

多個客戶端從zookeeper 的配置中心拿到配置，如果配置中心沒有配置就阻塞，如果配置修改了就拿到新的。

原理

因?yàn)閦ookeeper的同步性質(zhì)，即單線程的分兩段式（當(dāng)發(fā)生修改時，第一階段：leader向各個follower發(fā)送log任務(wù)，過半成功返回后進(jìn)行第二段的具體的修改）的事務(wù)方式，如果節(jié)點(diǎn)發(fā)生更改，則要么成功且所有客戶端都get到最新的修改結(jié)果，要么失敗。所以代碼只要注冊發(fā)生修改（包括數(shù)據(jù)修改，節(jié)點(diǎn)增加，刪除等，以下同）的事件，死循環(huán)取就行。

本文只寫了簡單的框架，復(fù)雜的業(yè)務(wù)邏輯沒有。

用zookeeper模擬另外的客戶端做修改的操作。

代碼

目錄結(jié)構(gòu)

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 和回調(diào) 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;/*** 默認(rèn)基于 session 的事件，只用來阻塞主線程，讓 zookeeper 連接成功后再繼續(xù)運(yùn)行*/ 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;/*** 模擬配置中心數(shù)據(jù)，zookeeper最大 1M* 比如是xml文件，需要自己實(shí)現(xiàn)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;/*** 注冊了節(jié)點(diǎn)的事件和回調(diào)*/ 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 {// 讓主線程阻塞住，上面那行代碼運(yùn)行下去cc.await();} catch (InterruptedException e) {e.printStackTrace();}}/*** Watcher 的接口實(shí)現(xiàn)** @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:// 如果節(jié)點(diǎn)被創(chuàng)建，就獲取它，讓它調(diào)用回調(diào)，給conf設(shè)值zk.getData("/appConf", this, this, "def");break;case NodeDeleted:// 如果節(jié)點(diǎn)被刪除了，考慮到容忍性的// 配置置空conf.setConfStr("");// 重新枷鎖cc = new CountDownLatch(1);break;case NodeDataChanged:// 如果節(jié)點(diǎn)被改了，就重新獲取,目的是讓他調(diào)用回調(diào)函數(shù)，給conf設(shè)置新的值zk.getData("/appConf", this, this, "def");break;case NodeChildrenChanged:break;}}/*** DataCallBack 的實(shí)現(xiàn)** @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);// 取到數(shù)據(jù)了，讓主線程繼續(xù)往下走cc.countDown();}}/*** StatCallBack 的實(shí)現(xiàn)** @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;} }

總結(jié)

以上是生活随笔為你收集整理的zookeeper配置中心的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。