Soul中使用的zookeeper监听功能尝试和nacos同步源码解读

Soul中使用的zookeeper监听功能尝试

一直很好奇soul中基于zookeeper的数据变更是如何做到的。虽然看到了是基于zkClient的实现的，还是想自己尝试下

zookeeper监听尝试

首先新建一个maven项目，引入zkClient

        <dependency>
           <groupId>com.101tec</groupId>
           <artifactId>zkclient</artifactId>
           <version>0.10</version>
       </dependency>

尝试着对zokkeeper中的数据和节点做一些改变，而这个改变就是Soul使用zookeeper中对于插件，选择器和规则变化将其映射到zookeeper上的结果

public class ZkClientTest {


   public static final String connect = "127.0.0.1:2181";
   private static ZkClient zkClient = null;
   private static String nodePath = "/zkclient1";
   private static String nodeChildPath = "/zkclient1/n1/n11/n111/n1111";
   public static void main(String[] args) throws Exception{
       //初始化
       init(connect,5000);

       //订阅节点数据改变或者子节点变化，只需要订阅一次，便可以一直使用。而原生zookeeper的监听是一次性的，需要重复注册。
       subscribe();

       //新增
       create(nodePath,"n1");
       //递归新增
       createRecursion(nodeChildPath,"n1");

       //查询
       query(nodePath);

       //修改
       update(nodePath,"n11");

       //单个节点删除
//        delete(nodePath);
       //递归删除
       deleteRecursion(nodePath);


       Thread.sleep(5000);
   }

   private static void deleteRecursion(String path) {
       boolean result = zkClient.deleteRecursive(path);
       System.out.println("delete:"+"["+path+"],result:"+result);
   }

   private static void delete(String path) {
       boolean result = zkClient.delete(path);
       System.out.println("delete:"+"["+path+"],result:"+result);
   }

   private static void update(String path, String data) {
       zkClient.writeData(path, data);
       System.out.println();
       //System.out.println("setData:"+"["+path+"],stat:"+stat);
   }

   private static void query(String path) {
       Object o = zkClient.readData(path);
       System.out.println("query:"+"["+path+"],result:"+o);
   }

   private static void createRecursion(String path,String data)  {
       zkClient.createPersistent(path,true);
       System.out.println("create:"+"["+path+"-->"+data);
   }

   private static void create(String path, String data) {
       boolean exists = zkClient.exists(path);
       if(exists){
           System.out.println("节点["+path+"]已存在，不能新增");
           return;
       }
       String result = zkClient.create(path, data, CreateMode.PERSISTENT);
       System.out.println("create:"+"["+path+"-->"+data+"],result:"+result);
   }

   private static void subscribe() {
       //订阅节点内容改变
       zkClient.subscribeDataChanges(nodePath, new IZkDataListener() {
                   public void handleDataChange(String dataPath, Object data) throws Exception {
                       System.out.println("handleDataChange----->"+dataPath+"|"+data);
                   }

                   public void handleDataDeleted(String dataPath) throws Exception {
                       System.out.println("handleDataDeleted----->"+dataPath);
                   }
               }
       );

       //订阅子节点改变
       zkClient.subscribeChildChanges(nodePath, new IZkChildListener() {
           public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
               System.out.println("handleChildChange----->"+parentPath+"|"+currentChilds);
           }
       });

   }

   private static void init(String connect, int sessionTimeout) {
       zkClient = new ZkClient(connect, sessionTimeout);
   }
}

具体的日志表现如下图所示

而在具体的zKclient类中可以看到，zkclient实际上是维护了一个while循环用来监听数据以及节点的变化

这就是Soul在ZookeeperDataChangedListener实际使用的具体的时候的具体使用的方案

Soul中Nacos是如何监听数据变化的

nacos的监听与zk类似，soul中的nacos一开始就监听了soul各个层级的数据的变化

  /**
     * Start.
     */
    public void start() {
        watcherData(PLUGIN_DATA_ID, this::updatePluginMap);
        watcherData(SELECTOR_DATA_ID, this::updateSelectorMap);
        watcherData(RULE_DATA_ID, this::updateRuleMap);
        watcherData(META_DATA_ID, this::updateMetaDataMap);
        watcherData(AUTH_DATA_ID, this::updateAuthMap);
    }

然后通过函数式编程的方式将数据通过各个方法更新到缓存中

    protected void watcherData(final String dataId, final OnChange oc) {
        Listener listener = new Listener() {
            @Override
            public void receiveConfigInfo(final String configInfo) {
                oc.change(configInfo);
            }

            @Override
            public Executor getExecutor() {
                return null;
            }
        };
        oc.change(getConfigAndSignListener(dataId, listener));
        LISTENERS.getOrDefault(dataId, new ArrayList<>()).add(listener);
    }

这里的最关键是Listener接口的获取配置信息的两个方法。可以看到这个即是通过预先定义号的元数据id,规则id来进行监听数据是否发生变化的

上面就是nacos和zookeeper监听数据变化的过程，现在可以想到。不管是配置中心或者注册中心。数据变化的监听都是有必要的，可以为使用的应用做出跟随配置改变而改变应用逻辑的能力。这也是Soul网关中很重要的一项能力

欢迎搜索关注本人与朋友共同开发的微信面经小程序【大厂面试助手】和公众号【微瞰技术】，以及总结的分类面试题https://github.com/zhendiao/JavaInterview