代码实现

全部以异步回调的方式实现。

DistributeCfgClient类

  1. import org.apache.zookeeper.ZooKeeper;
  2. import java.io.IOException;
  3. import java.util.concurrent.CountDownLatch;
  5. public class DistributeCfgClient {
  6.     //zookeeper集群,字符串最后跟上一个路径/testConfig表示连接后只能看到/testConfig目录下的内容，其它兄弟节点是看不到的
  7.     private static final String ZK_SERVER_LIST = "120.53.119.107:2181,120.53.119.107:2182,120.53.119.107:2183/testConfig";
  8.     //设置超时时间，当节点3s后没反应就认该节点已经挂了，比如创建临时节点客户端关闭链接后3s,临时节点才移除
  9.     private static final int sessionTimeout = 3000;
  10.     //创建new ZooKeeper时传递的watch
  11.     private static DefaultWatch defaultWatch = new DefaultWatch();
  12.     private static CountDownLatch countDownLatch = new CountDownLatch(1);
  15.     public ZooKeeper zooKeeper() throws IOException, IOException, InterruptedException {
  16.         defaultWatch.setCountDownLatch(countDownLatch);
  17.         //这个会立即返回一个ZooKeeper对象，如果别人直接拿过去用，可能还没有连接成功，所以我们在这里要阻塞一下，当连接成功后再返回
  18.         ZooKeeper zk = new ZooKeeper(ZK_SERVER_LIST,sessionTimeout,defaultWatch);
  19.         countDownLatch.await();
  20.         return zk;
  21.     }
  24.     public static void main(String[] args) throws InterruptedException, IOException {
  26.         DistributeCfgClient distributeCfgClient = new DistributeCfgClient();
  27.         ZooKeeper zk = distributeCfgClient.zooKeeper();
  28.         WallStatCallBack wallStatCallBack = new WallStatCallBack();
  29.         wallStatCallBack.setZk(zk);
  30.         ReceiveDataConf receiveDataConf = new ReceiveDataConf();
  31.         wallStatCallBack.setConfigData(receiveDataConf);
  32.         //获取数据
  33.         wallStatCallBack.getConfigDataAwait();
  35.         while (true){ //业务代码，以后只需关心配置和这部分代码就可以了
  36.             if (receiveDataConf.getConf().equals("")){
  37.                 //等待获取数据
  38.                 wallStatCallBack.getConfigDataAwait();
  39.                 System.out.println(receiveDataConf.getConf());
  40.             }else {
  41.                 System.out.println(receiveDataConf.getConf());
  42.             }
  44.             Thread.sleep(2000);
  45.         }
  46.     }
  47. }

DefaultWatch类

  1. import org.apache.zookeeper.WatchedEvent;
  2. import org.apache.zookeeper.Watcher;
  4. import java.util.concurrent.CountDownLatch;
  6. /**
  7.  * 创建zookeeper对象时使用的watch，可以理解session级别的watch
  8.  */
  9. public class DefaultWatch implements Watcher {
  10.     private  CountDownLatch countDownLatch;
  12.     public  void setCountDownLatch(CountDownLatch countDownLatch) {
  13.         this.countDownLatch = countDownLatch;
  14.     }
  16.     @Override
  17.     public void process(WatchedEvent event) {
  19.         System.out.println(event.toString());
  21.         Event.KeeperState state = event.getState();
  22.         switch (state) {
  23.             case Unknown -> {
  24.             }
  25.             case Disconnected -> {
  26.             }
  27.             case NoSyncConnected -> {
  28.             }
  29.             case SyncConnected -> { //连接成功
  30.                 countDownLatch.countDown();
  32.             }
  33.             case AuthFailed -> {
  34.             }
  35.             case ConnectedReadOnly -> {
  36.             }
  37.             case SaslAuthenticated -> {
  38.             }
  39.             case Expired -> {
  40.             }
  41.             case Closed -> {
  42.             }
  43.         }
  44.     }
  45. }

需要获取的配置数据

  1. public class ReceiveDataConf {
  2.     private String conf;
  4.     public String getConf() {
  5.         return conf;
  6.     }
  8.     public void setConf(String conf) {
  9.         this.conf = conf;
  10.     }
  11. }

WallStatCallBack核心类

  1. import org.apache.zookeeper.AsyncCallback;
  2. import org.apache.zookeeper.WatchedEvent;
  3. import org.apache.zookeeper.Watcher;
  4. import org.apache.zookeeper.ZooKeeper;
  5. import org.apache.zookeeper.data.Stat;
  7. import java.util.concurrent.CountDownLatch;
  9. /**
  10.  * WallStatCallBack扮演Watcher角色、StatCallback角色、DataCallback角色
  11.  */
  12. public class WallStatCallBack implements Watcher, AsyncCallback.StatCallback, AsyncCallback.DataCallback {
  14.     private ZooKeeper zk;
  16.     private ReceiveDataConf configData; //接受的数据
  17.     private CountDownLatch awaitDataLatch = new CountDownLatch(1);
  19.     public void setConfigData(ReceiveDataConf configData) {
  20.         this.configData = configData;
  21.     }
  23.     public void setZk(ZooKeeper zk) {
  24.         this.zk = zk;
  25.     }
  27.     public void getConfigDataAwait() throws InterruptedException {
  28.         //这里统一使用异步的方式进行开发,如果节点存在就会异步调用StateCallback
  29.         zk.exists("/DataBaseConf", this, this,"abc");
  30.         awaitDataLatch.await();
  32.     }
  34.     /**
  35.      * watch的Callback，监听到数据的变化后调用
  36.      * @param event
  37.      */
  38.     @Override
  39.     public void process(WatchedEvent event) {
  41.        Event.EventType type = event.getType();
  42.         switch (type) {
  43.             case None -> {
  44.             }
  45.             case NodeCreated -> {
  46.                 System.out.println("NodeCreated====");
  47.                 zk.getData("/DataBaseConf",this,this,"abc");
  48.             }
  49.             case NodeDeleted -> {
  50.                 //如果删除，可以自己根据业务情况实现，容忍性
  51.                 configData.setConf(""); //设置获取的配置数据为空
  52.                 awaitDataLatch = new CountDownLatch(1); //让外界继续阻塞
  53.             }
  54.             case NodeDataChanged -> {
  55.                 zk.getData("/DataBaseConf",this,this,"abc");
  56.             }
  57.             case NodeChildrenChanged -> {
  58.             }
  59.             case DataWatchRemoved -> {
  60.             }
  61.             case ChildWatchRemoved -> {
  62.             }
  63.             case PersistentWatchRemoved -> {
  64.             }
  65.         }
  66.     }
  68.     /**
  69.      * 数据返回的callback，比如zk.getData,获取完数据后异步的方式返回数据
  70.      * @param rc
  71.      * @param path
  72.      * @param ctx
  73.      * @param data
  74.      * @param stat
  75.      */
  76.     @Override
  77.     public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
  79.         if (data != null){
  80.             String s = new String(data);
  81.             configData.setConf(s);
  82.             awaitDataLatch.countDown();
  83.         }
  84.     }
  86.     /**
  87.      *节点状态的callback，比如判断节点是否存在等
  88.      * @param rc
  89.      * @param path
  90.      * @param ctx
  91.      * @param stat
  92.      */
  93.     @Override
  94.     public void processResult(int rc, String path, Object ctx, Stat stat) {
  95.         if (stat != null){ //有需要的节点
  96.             zk.getData("/DataBaseConf",this,this,"abc");
  97.         }
  99.     }
  100. }