ZooKeeper分布式配置,看這篇就夠了
本文轉(zhuǎn)載自微信公眾號「牧小農(nóng)」,作者牧小農(nóng)。轉(zhuǎn)載本文請聯(lián)系牧小農(nóng)公眾號。
ZooKeeper 的由來
PS:這個不重要,不感興趣的,可以直接看下面
來源:《從 Paxos 到 ZooKeeper 》
ZooKeeper 最早起源于雅虎研究院的一個研究小組,在當時,研究人員發(fā)現(xiàn),在雅虎內(nèi)部有很多的大型系統(tǒng)基本上都需要依賴一個類似的系統(tǒng)來進行分布式協(xié)調(diào),但是這些系統(tǒng)往往都存在分布式單點的問題,所有雅虎的開發(fā)人員就嘗試開發(fā)了一個通用的無單點問題的分布式協(xié)調(diào)框架,以便讓開發(fā)人員將精力集中在處理業(yè)務邏輯上。關(guān)于"ZooKeeper"這個項目的名字。也有一個故事,在項目開始初期,因為考慮到內(nèi)部的很多項目都是用動物的名字來命名的(例如:Pig項目),所以雅虎的工程師也希望給這個項目也取一個動物的名字,這個時候擔任研究院的首席科學家 Raghu Ramakrishnan 開玩笑地說:“在這樣下去,我們這兒就變成動物園”,此話一出,大家紛紛表示就叫動物園管理員吧,因為各個以動物命名的分布式組件放在一起,這個分布式系統(tǒng)看上去就像一個大型的動物園了,而ZooKeeper 正好要用來進行分布式環(huán)境的協(xié)調(diào),于是ZooKeeper 的名字也就由此誕生了。
分布式配置中心
在上一期中我們講解了 ZooKeeper集群的配置和安裝,ZooKeeper集群 主要是幫我們做分布式協(xié)調(diào)的,今天我們用ZK實現(xiàn)分布式配置。關(guān)于ZooKeeper 集群的配置大家可以參考上一篇文章《Zookeeper 集群部署的那些事兒》。
為什么需要分布式配置中心
對于剛開始的時候,很多公司的服務器可能是由單個組成,但是隨著業(yè)務的發(fā)展,單一節(jié)點的服務無法滿足業(yè)務的飛速發(fā)展,后面就出現(xiàn)了分布式、集群的概念,到了現(xiàn)在形成的微服務,技術(shù)的改進能夠更好的滿足業(yè)務的需要。
假設(shè)我們線上有很多個微服務分布在不同的服務器上,其中一個微服務,我們就叫它 goods-service,當 goods-service的IP地址需要變更的時候,但是 goods-service又對很多其他的程序提供了服務,這個時候如果沒有一個統(tǒng)一配置的東西,每一個應用到 goods-service的應用程序都要做相應的IP地址修改,這是一個很麻煩的事情!
如果使用ZooKeeper來做分布式配置的話,是可以解決這個問題的。
注冊中心對比
如果我們只考慮服務治理的話,Eureka是比較合適的,Eureka是比較純粹的注冊中心了,和Eureka不同Apache ZooKeeper 在設(shè)計的時候就遵循 CP原則,任何時候?qū)?ZooKeeper 的訪問請求都能得到一致的數(shù)據(jù)結(jié)果,同事系統(tǒng)對網(wǎng)絡分割具有容錯性,今天我們講解的就是關(guān)于ZooKeeper 的注冊發(fā)現(xiàn)。
配置中心的核心
低延遲: 配置改變( create/update/delete)后能夠最快的把最新的配置同步到其他節(jié)點中
高可用: 配置中心可以穩(wěn)定的對外提供服務
其中 低延遲 我們可以通過 ZooKeeper 的 Watcher 機制來實現(xiàn)(等下會講到Watcher機制)。約定一個節(jié)點用來存放配置信息,每個客戶端都來監(jiān)聽這個節(jié)點的NodeDataChanged事件,當配置發(fā)生改變時將最新的配置更新到這個節(jié)點上,誰更新無所謂,任何節(jié)點都可以更新,當這個節(jié)點觸發(fā) NodeDataChanged 事件后,在去通知所有監(jiān)聽這個節(jié)點的客戶端去獲取這個節(jié)點的最新信息,因為watcher 是一次性的,所以當我們在獲取最新信息的時候需要設(shè)置監(jiān)聽事件,大部分查詢信息都是具有原子性的,所以ZooKeeper中的 getData 也是具有原子性操作,能夠保證我們?nèi)〉玫男畔⑹亲钚碌摹?/p>
對于 高可用 我們首先需要保證的多集群操作來進行ZooKeeper進行部署,在代碼層不太需要做過多的工作。
Watch 機制
Watch 是 ZooKeeper 針對節(jié)點的一次性觀察者機制,就如同我們上面 * 低延遲* 中講到的,一次觸發(fā)后就失效,需要手工重新創(chuàng)建Watch。
當Watch監(jiān)視的數(shù)據(jù)發(fā)生變化的時候,就會通知設(shè)置了 Watch 的客戶端,就是我們API中的Watcher,Watcher機制就是為了監(jiān)聽Znode節(jié)點發(fā)生了哪些變化,所以會有對應的事件類型和狀態(tài)類型,用過代碼中switch進行監(jiān)聽,一個客戶端可以鏈接多個節(jié)點,只要Znode節(jié)點發(fā)生變化就會執(zhí)行 process(WatchedEventevent)。
如下圖所示:
從上圖中我們可以看到,在ZooKeeper中,Watch采用的是推送機制,而不是客戶端輪詢,有些中間件采用的是拉取的模式,例如:KafKa。
Watch有兩種監(jiān)聽模式,分別為 事件類型和狀態(tài)類型 :
事件類型:Znode 節(jié)點關(guān)聯(lián),主要是針對節(jié)點的操作
- 創(chuàng)建節(jié)點:EventType.NodeCreated
- 節(jié)點數(shù)據(jù)發(fā)生變化:EventType.NodeDataChanged
- 當前節(jié)點的子節(jié)點發(fā)生變化:EventType.NodeChildrenChanged
- 刪除節(jié)點:EventType.NodeDeleted
狀態(tài)類型:客戶端關(guān)聯(lián),主要是針對于ZooKeeper集群和應用服務之間的狀態(tài)的變更
- 未連接:KeeperState.Disconnected
- 已連接:KeeperState.SyncConnected
- 認證失敗:KeeperState.AuthFailed
- 過期:KeeperState.Expired
- 客戶端連接到只讀服務器:KeeperState.ConnectedReadOnly
watch的特性
一次性觸發(fā): 對于ZooKeeper的Watcher事件,是一次性觸發(fā)的,當 Watch 監(jiān)視的數(shù)據(jù)發(fā)生變化的時候,通知設(shè)置當前Watch 的 Client,就是我們對應的 Watcher,因為ZooKeeper 的監(jiān)控都是一次性的,所以我們需要在每次觸發(fā)后設(shè)置監(jiān)控。
客戶端串行執(zhí)行: 客戶端Watcher回調(diào)的過程是一個串行同步的過程,可以為我們保證順序的執(zhí)行。
輕量級: WatchedEvent是ZooKeeper整個Watcher通知機制的最小通知單元,總共包含三個部分(通知狀態(tài)、事件類型和節(jié)點路徑),Watcher通知,只會告訴客戶端發(fā)生事件而不會告知具體內(nèi)容,需要客戶端主動去進行獲取,比如 當監(jiān)聽到 WatchedEvent.NodeDataChanged 信息變化的時候,只會告訴我們這個節(jié)點的數(shù)據(jù)發(fā)生了變更,你快來獲取最新的值吧。
客戶端設(shè)置的每個監(jiān)視點與會話關(guān)聯(lián),如果會話過期,等待中的監(jiān)視點將會被刪除。不過監(jiān)視點可以跨越不同服務端的連接而保持,例如,當一個ZooKeeper客戶端與一個ZooKeeper服務端的連接斷開后連接到集合中的另一個服務端,客戶端會發(fā)送未觸發(fā)的監(jiān)視點列表,在注冊監(jiān)視點時,服務端將要檢查已監(jiān)視的znode節(jié)點在之前注冊監(jiān)視點之后是否已經(jīng)變化,如果znode節(jié)點已經(jīng)發(fā)生變化,一個監(jiān)視點的事件就會被發(fā)送給客戶端,否則在新的服務端上注冊監(jiān)視點。這一機制使得我們可以關(guān)心邏輯層會話,而非底層連接本身。
客戶端注冊
ZooKeeper 注冊的時候會向ZooKeeper 服務端請求注冊,服務端會返回請求響應,不管成功失敗,都會返回響應結(jié)果,當響應成功的時候,ZooKeeper服務端會把Watcher對象放到客戶端的WatchManager管理并返回響應給客戶端
服務端注冊
FinalRequestProcessor.ProcessRequest()會判斷當前請求是否需要注冊Watcher
如果ZooKeeper判斷當前客戶端需要進行Watcher注冊,會將當前的ServerCnxn 對象和數(shù)據(jù)路徑傳入 getData 方法中去。ServerCnxn 是ZooKeeper 客戶端和服務端之間的連接接口,代表了一個客戶端和服務端的連接,可以將 ServerCnxn 當做一個 Watcher 對象,因為它實現(xiàn)了 Watcher 的 process 接口。
WatcherManager
WatcherManager是 ZK服務端 Watcher 的管理器,分為 WatchTable 和 Watch2Paths 兩個存儲結(jié)構(gòu),這兩個是不同的存儲結(jié)構(gòu) 1)WatchTable:從數(shù)據(jù)節(jié)點路徑的顆粒度管理 Watcher 2)Watch2Paths:從Watcher的顆粒度來控制時間出發(fā)的數(shù)據(jù)節(jié)點
在服務端,DataTree 中會托管兩個 WatchManager, 分別是 dataWatches (數(shù)據(jù)變更Watch) 和 childWatches(子節(jié)點變更Watch)。
Watcher 觸發(fā)邏輯
1)封裝WatchedEven:將(KeeperState(通知狀態(tài)),EventType(事件類型),Path(節(jié)點路徑))封裝成一個 WatchedEvent 對象 2)查詢Watcher:根據(jù)路徑取出對應的Watcher,如果存在,取出數(shù)據(jù)同時從 WatcherManager(WatchTable/Watch2Paths) 中刪除 3)調(diào)用Process方法觸發(fā)Watcher
4.客戶端回調(diào) Watcher
1)反序列化:字節(jié)流轉(zhuǎn)換成 WatcherEvent 對象 2)處理 chrootPath:如果客戶端設(shè)置了 chrootPath 屬性,那么需要對服務器傳過來的完整節(jié)點路徑進行 chrootPath 處理,生成客戶端的一個相對節(jié)點路徑。比如(/mxn/app/love,經(jīng)過chrootPath處理,會變成 /love) 3)還原 WatchedEvent:WatcherEvent 轉(zhuǎn)換成 WatchedEvent 4)回調(diào) Watcher:將 WatcherEvent 對象交給 EventThread 線程,在下一個輪詢周期中進行 Watcher 回調(diào)
EventThread 處理時間通知
1) SendThread 接收到服務端的通知事件后,會通過調(diào)用 EventThread.queueEvent 方法將事件傳給 EventThread 線程 2)queueEvent 方法首先會根據(jù)該通知事件,從 ZKWatchManager 中取出所有相關(guān)的 Watcher 客戶端識別出 事件類型 EventType 后,會從相應的 Watcher 存儲 (即3個注冊方法( dataWatches、existWatcher 或 childWatcher)中去除對應的 Watcher 3) 獲取到相關(guān)的所有 Watcher 后,會將其放入 waitingEvents 這個隊列去
代碼實現(xiàn)
下面我們就來演示如何使用代碼來實現(xiàn)ZooKeeper的配置
首先我們需要引入ZK的jar
- <dependency>
- <groupId>org.apache.zookeeper</groupId>
- <artifactId>zookeeper</artifactId>
- <version>3.6.3</version>
- </dependency>
配置類
既然我們要做的是分布式配置,首先我們需要模擬一個配置,這個配置用來同步服務的地址
- /**
- * @program: mxnzookeeper
- * @ClassName MyConf
- * @description: 配置類
- * @author: muxiaonong
- * @create: 2021-10-19 22:18
- * @Version 1.0
- **/
- public class MyConfig {
- private String conf ;
- public String getConf() {
- return conf;
- }
- public void setConf(String conf) {
- this.conf = conf;
- }
- }
Watcher
創(chuàng)建ZooKeeper的時候,我們需要一個Watcher進行監(jiān)聽,后續(xù)對Znode節(jié)點操作的時候,我們也需要使用到Watcher,但是這兩類的功能不一樣,所以我們需要定義一個自己的watcher類,如下所示:
- import org.apache.zookeeper.WatchedEvent;
- import org.apache.zookeeper.Watcher;
- import java.util.concurrent.CountDownLatch;
- /**
- * @program: mxnzookeeper
- * @ClassName DefaultWatch
- * @description:
- * @author: muxiaonong
- * @create: 2021-10-19 22:02
- * @Version 1.0
- **/
- public class DefaultWatch implements Watcher {
- CountDownLatch cc;
- public void setCc(CountDownLatch cc) {
- this.cc = cc;
- }
- @Override
- public void process(WatchedEvent event) {
- System.out.println(event.toString());
- switch (event.getState()) {
- case Unknown:
- break;
- case Disconnected:
- break;
- case NoSyncConnected:
- break;
- case SyncConnected:
- System.out.println("連接成功。。。。。");
- //連接成功后,執(zhí)行countDown,此時便可以拿zk對象使用了
- cc.countDown();
- break;
- case AuthFailed:
- break;
- case ConnectedReadOnly:
- break;
- case SaslAuthenticated:
- break;
- case Expired:
- break;
- case Closed:
- break;
- }
- }
- }
由于是異步進行操作的,我們創(chuàng)建一個ZooKeeper對象之后,如果不進行阻塞操作的話,有可能還沒有連接完成就執(zhí)行后續(xù)的操作,所以這里我們用 CountDownLatch進行阻塞操作,當監(jiān)測連接成功后,進行 countDown放行,執(zhí)行后續(xù)的ZK的動作。
當我們連接成功 ZooKeeper 之后,我們需要通過 exists判斷是否存在節(jié)點,存在就進行 getData操作。這里我們創(chuàng)建一個 WatchCallBack因為 exists和getData都需要一個callback,所以除了實現(xiàn)Watcher以外還需要實現(xiàn) 節(jié)點狀態(tài):AsyncCallback.StatCallback數(shù)據(jù)監(jiān)聽:AsyncCallback.DataCallback
- 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;
- /**
- * @program: mxnzookeeper
- * @ClassName WatchCallBack
- * @description:
- * @author: muxiaonong
- * @create: 2021-10-19 22:13
- * @Version 1.0
- **/
- public class WatchCallBack implements Watcher, AsyncCallback.StatCallback, AsyncCallback.DataCallback {
- ZooKeeper zk ;
- MyConfig conf ;
- CountDownLatch cc = new CountDownLatch(1);
- public MyConfig getConf() {
- return conf;
- }
- public void setConf(MyConfig conf) {
- this.conf = conf;
- }
- public ZooKeeper getZk() {
- return zk;
- }
- public void setZk(ZooKeeper zk) {
- this.zk = zk;
- }
- public void aWait(){
- //exists的異步實現(xiàn)版本
- zk.exists(ZKConstants.ZK_NODE,this,this ,"exists watch");
- try {
- cc.await();
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
- /** @Author mxn
- * @Description //TODO 此回調(diào)用于檢索節(jié)點的stat
- * @Date 21:24 2021/10/20
- * @param rc 調(diào)用返回的code或結(jié)果
- * @param path 傳遞給異步調(diào)用的路徑
- * @param ctx 傳遞給異步調(diào)用的上下文對象
- * @param stat 指定路徑上節(jié)點的Stat對象
- * @return
- **/
- @Override
- public void processResult(int rc, String path, Object ctx, Stat stat) {
- if(stat != null){
- //getData的異步實現(xiàn)版本
- zk.getData(ZKConstants.ZK_NODE,this,this,"status");
- }
- }
- /** @Author mxn
- * @Description //TODO 此回調(diào)用于檢索節(jié)點的數(shù)據(jù)和stat
- * @Date 21:23 2021/10/20
- * @param rc 調(diào)用返回的code或結(jié)果
- * @param path 傳遞給異步調(diào)用的路徑
- * @param ctx 傳遞給異步調(diào)用的上下文對象
- * @param data 節(jié)點的數(shù)據(jù)
- * @param stat 指定節(jié)點的Stat對象
- * @return
- **/
- @Override
- public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
- if(data != null ){
- String s = new String(data);
- conf.setConf(s);
- cc.countDown();
- }
- }
- /** @Author mxn
- * @Description //TODO Watcher接口的實現(xiàn)。
- * Watcher接口指定事件處理程序類必須實現(xiàn)的公共接口。
- * ZooKeeper客戶機將從它連接到的ZooKeeper服務器獲取各種事件。
- * 使用這種客戶機的應用程序通過向客戶機注冊回調(diào)對象來處理這些事件。
- * 回調(diào)對象應該是實現(xiàn)監(jiān)視器接口的類的實例。
- * @Date 21:24 2021/10/20
- * @Param watchedEvent WatchedEvent表示監(jiān)視者能夠響應的ZooKeeper上的更改。
- * WatchedEvent包含發(fā)生了什么,
- * ZooKeeper的當前狀態(tài),以及事件中涉及的znode的路徑。
- * @return
- **/
- @Override
- public void process(WatchedEvent event) {
- switch (event.getType()) {
- case None:
- break;
- case NodeCreated:
- //當一個node被創(chuàng)建后,獲取node
- //getData中又會觸發(fā)StatCallback的回調(diào)processResult
- zk.getData(ZKConstants.ZK_NODE,this,this,"sdfs");
- break;
- case NodeDeleted:
- //節(jié)點刪除
- conf.setConf("");
- //重新開啟CountDownLatch
- cc = new CountDownLatch(1);
- break;
- case NodeDataChanged:
- //節(jié)點數(shù)據(jù)被改變了
- //觸發(fā)DataCallback的回調(diào)
- zk.getData(ZKConstants.ZK_NODE,this,this,"sdfs");
- break;
- //子節(jié)點發(fā)生變化的時候
- case NodeChildrenChanged:
- break;
- }
- }
- }
當前面準備好了之后,我們可以編寫測試用例了:
ZKUtils 工具類
- import org.apache.zookeeper.ZooKeeper;
- import java.util.concurrent.CountDownLatch;
- /**
- * @program: mxnzookeeper
- * @ClassName ZKUtils
- * @description:
- * @author: muxiaonong
- * @create: 2021-10-19 21:59
- * @Version 1.0
- **/
- public class ZKUtils {
- private static ZooKeeper zk;
- //192.168.5.130:2181/mxn 這個后面/mxn,表示客戶端如果成功建立了到zk集群的連接,
- // 那么默認該客戶端工作的根path就是/mxn,如果不帶/mxn,默認根path是/
- //當然我們要保證/mxn這個節(jié)點在ZK上是存在的
- private static String address ="192.18.5.129:2181,192.168.5.130:2181,192.168.5.130:2181/mxn";
- private static DefaultWatch watch = new DefaultWatch();
- private static CountDownLatch init = new CountDownLatch(1);
- public static ZooKeeper getZK(){
- try {
- //因為是異步的,所以要await,等到連接上zk集群之后再進行后續(xù)操作
- zk = new ZooKeeper(address,1000,watch);
- watch.setCc(init);
- init.await();
- } catch (Exception e) {
- e.printStackTrace();
- }
- return zk;
- }
- }
測試類:
- import org.apache.zookeeper.ZooKeeper;
- import org.junit.Before;
- import org.junit.Test;
- /**
- * @program: mxnzookeeper
- * @ClassName TestConfig
- * @description:
- * @author: muxiaonong
- * @create: 2021-10-19 22:04
- * @Version 1.0
- **/
- public class TestConfig {
- ZooKeeper zk;
- @Before
- public void conn(){
- zk = ZKUtils.getZK();
- }
- /** @Author mxn
- * @Description //TODO 關(guān)閉ZK
- * @Date 21:16 2021/10/20
- * @Param
- * @return
- **/
- public void close(){
- try {
- zk.close();
- }catch (Exception e){
- e.printStackTrace();
- }
- }
- @Test
- public void getConf(){
- WatchCallBack watchCallBack = new WatchCallBack();
- watchCallBack.setZk(zk);
- MyConfig myConfig = new MyConfig();
- watchCallBack.setConf(myConfig);
- //阻塞等待
- watchCallBack.aWait();
- while(true){
- if(myConfig.getConf().equals("")){
- System.out.println("zk node 節(jié)點丟失了 ......");
- watchCallBack.aWait();
- }else{
- System.out.println(myConfig.getConf());
- }
- //
- try {
- //每隔500毫秒打印一次
- Thread.sleep(500);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
- }
運行測試
首先我們要知道,因為我們連接IP的時候加上了 /mxn這個目錄結(jié)構(gòu),所以我們在服務器初始狀態(tài)就必須要有這個節(jié)點:
集群初始狀態(tài):
- [zk: localhost:2181(CONNECTED) 7] ls /
- [mxn, zookeeper]
我們啟動程序看看
連接成功
ZooKeeper 下 /mxn 現(xiàn)在也是空
- [zk: localhost:2181(CONNECTED) 9] ls /mxn
- []
- [zk: localhost:2181(CONNECTED) 10]
現(xiàn)在我們來創(chuàng)建一個 /mxn/myZNode節(jié)點數(shù)據(jù)
- [zk: localhost:2181(CONNECTED) 10] create /mxn/myZNode "muxiaonong666"
- Created /mxn/myZNode
可以看到,創(chuàng)建完成之后,程序馬上給出響應,打印出了我配置的值,muxiaonong666
此時,再設(shè)置 /mxn/myZNode的值為 muxiaonong6969
啪,很快啊!!!我們就可以看到值瞬間改變了
這個時候我們?nèi)绻麆h除 /mxn/myZNode節(jié)點,會發(fā)生什么呢,前面我們已經(jīng)寫了watch,如果Znode被刪除了,,watch and callback執(zhí)行
- case NodeDeleted:
- //節(jié)點刪除
- conf.setConf("");
- //重新開啟CountDownLatch
- cc = new CountDownLatch(1);
- break;
- if(myConfig.getConf().equals("")){
- System.out.println("zk node 節(jié)點丟失了 ......");
- ////此時應該阻塞住,等待著node重新創(chuàng)建
- watchCallBack.aWait();
- }
刪除 /mxn/myZNode 節(jié)點
- delete /mxn/myZNode
我們可以看到前面還在打印數(shù)據(jù),后面就提示丟失。
但是這個時候我們客戶端沒有關(guān)閉,而是還在等待數(shù)據(jù)的更新,如果這個時候當重新進行創(chuàng)建 /mxn/myZNode節(jié)點的時候,程序又會繼續(xù)瘋狂輸出。
- create/mxn/myZNode"muxiaonong666"
程序正常運行,并且成功獲取到了zk配置的最新數(shù)據(jù),到這里基本上就實現(xiàn)了,ZooKeeper的分布式配置中心功能了
在這里我測試用的是 getData,但是在項目實戰(zhàn)用可能用的更多的是 子節(jié)點的操作 getChildren
總結(jié)
到這里我們這篇 ZooKeeper分布式配置注冊發(fā)現(xiàn) 就講完了,如果有疑問的地方歡迎進行討論,ZooKeeper 可以作為分布式配置中心,也可以用來當然微服務的注冊,不過現(xiàn)在微服務都有自己的一套服務發(fā)現(xiàn),對于了解ZooKeeper可以我們方便我們在進行技術(shù)選型的時候更好的去抉擇, ZooKeeper 的高可用和最終一致性也是比較穩(wěn)定。
本文代碼地址:https://github.com/muxiaonong/ZooKeeper/tree/master/mxnzookeeper