<sub id="gqw76"><listing id="gqw76"></listing></sub>
      <sub id="gqw76"><listing id="gqw76"></listing></sub>

    1. <form id="gqw76"><legend id="gqw76"></legend></form>
    2. Loading

      ZooKeeper學習筆記三:使用ZooKeeper實現一個簡單的配置中心

      作者:Grey

      原文地址:ZooKeeper學習筆記三:使用ZooKeeper實現一個簡單的配置中心

      前置知識

      完成ZooKeeper集群搭建以及熟悉ZooKeeperAPI基本使用

      需求

      很多程序往往是通過心跳檢測來判斷配置的變更,通過zk的回調機制,我們可以實現比心跳更為快速的配置檢測機制,包括配置的新增,修改和刪除

      流程也比較簡單:
      image

      環境準備

      一個zk集權,ip和端口分別為:

      • 192.168.205.145:2181
      • 192.168.205.146:2181
      • 192.168.205.147:2181
      • 192.168.205.148:2181

      定義主方法

      App.java

      public class App {
          public static void main(String[] args) {
      		// 需要監控的路徑是/AppConf
              String path = "/AppConf";
              while (true) {
                  ConfigCenter configCenter = new ConfigCenter(path);
                  String conf = configCenter.getConf();
      			// 配置不為空則拿到最新的配置
                  if (null != conf && !conf.trim().isEmpty()) {
                      System.out.println(conf);
                  }
      			// 睡眠一段時間
                  pending(1000);
              }
          }
      }
      

      zk初始化工具類

      參考ZooKeeperAPI基本使用中的ZookeeperConfig類,主要的方法為:

      public class ZookeeperConfig {
          private static final String ADDRESS = "192.168.205.145:2181,192.168.205.146:2181,192.168.205.147:2181,192.168.205.148:2181";
          private static ZooKeeper zk;
          static CountDownLatch latch;
      	// 獲取一個zk客戶端
          public static ZooKeeper create() {
              latch = new CountDownLatch(1);
              try {
                  zk = new ZooKeeper(ADDRESS, 3000, new DefaultWatch());
                  latch.await();
              } catch (IOException | InterruptedException e) {
                  e.printStackTrace();
              }
              return zk;
          }
      ...
      }
      

      實現配置中心邏輯

      配置中心的入口方法為:getConf()

      public String getConf() {
              aWait();
              return value;
      }
      

      其中aWait()方法用于監聽配置信息的變動(比如修改,刪除,增加),并且通過CountDownLatch阻塞運行,zk一旦監聽到配置信息的變動,即會觸發回調,并執行countDown(),這樣前面的CountDownLatch即可往下運行:

       public void aWait() {
              zk.exists(conf, this, this, "dasdfa");
              try {
                  latch.await();
              } catch (InterruptedException e) {
                  e.printStackTrace();
              }
          }
      

      ConfigCenter類實現了Watcher, AsyncCallback.StatCallback, AsyncCallback.DataCallback三個接口,所以監聽方法和回調方法都在ConfigCenter中實現,其中,監聽方法:

       @Override
          public void process(WatchedEvent event) {
      
              Event.EventType type = event.getType();
              switch (type) {
                  case None:
                      break;
                  case NodeCreated:
                      System.out.println("node created");
                      zk.getData(conf, this, this, "node created");
                      latch.countDown();
                      break;
                  case NodeDeleted:
                      try {
                          System.out.println("config deleted");
                          this.value = "";
                          latch = new CountDownLatch(1);
                      } catch (Exception e) {
                          e.printStackTrace();
                      }
                      break;
                  case NodeDataChanged:
                      System.out.println("node changed");
                      zk.getData(conf, this, this, "node changed");
                      latch.countDown();
                      break;
                  case NodeChildrenChanged:
                      break;
                  case DataWatchRemoved:
                      break;
                  case ChildWatchRemoved:
                      break;
                  case PersistentWatchRemoved:
                      break;
              }
          }
      

      如上,當監聽到節點創建和修改的時候,觸發getData,當監聽到節點刪除的時候,重新初始化CountDownLatch,讓線程阻塞。

      回調方法如下:

      @Override
          public void processResult(int rc, String path, Object ctx, Stat stat) {
              if (stat != null) {
                  zk.getData(conf, this, this, "getData");
              }
          }
      
          @Override
          public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
              if (data != null) {
                  this.value = new String(data);
                  latch.countDown();
              }
          }
      

      方法1表示,當節點存在(stat!=null)的時候,獲取一次數據(重新注冊監聽)。
      方法2表示,當節點數據存在的時候,將節點數據取出,并且countDown(),解除阻塞。

      運行效果

      先把zk中的/AppConf節點刪除,執行App.java,程序阻塞

      通過zkCli.sh連接任意zk服務端,創建一個/AppConf節點,并賦值

      [zk: localhost:2181(CONNECTED) 58] create /AppConf "hello"
      Created /AppConf
      
      

      控制臺同時打印出:

      node created
      hello
      hello
      hello
      hello
      hello
      hello
      hello
      hello
      

      繼續通過zkCli重新設置/AppConf節點的值

      [zk: localhost:2181(CONNECTED) 59] set /AppConf "world"
      [zk: localhost:2181(CONNECTED) 60] 
      
      

      配置信息立馬生效,控制臺打印

      ...
      hello
      hello
      node changed
      world
      world
      world
      ...
      

      通過zkCli.sh刪除/AppConf節點,控制臺阻塞運行,并打印

      config deleted
      

      通過zkCli.sh重新創建/AppConf節點,

      [zk: localhost:2181(CONNECTED) 61] create /AppConf "hello2"
      Created /AppConf
      
      

      控制臺立即解除阻塞,并打印

      config deleted
      node created
      hello2
      hello2
      ..
      

      完整代碼

      Github

      posted @ 2021-06-07 22:20  Grey Zeng  閱讀(285)  評論(0編輯  收藏  舉報
      最新chease0ldman老人|无码亚洲人妻下载|大香蕉在线看好吊妞视频这里有精品www|亚洲色情综合网

        <sub id="gqw76"><listing id="gqw76"></listing></sub>
        <sub id="gqw76"><listing id="gqw76"></listing></sub>

      1. <form id="gqw76"><legend id="gqw76"></legend></form>