elastic-job 源码解读之从源码看zookeeper的API的使用

2019年6月9日 216次阅读来源: 一滴水的坚持

在众多的zookeeper客户端工具中，elastic-job拥抱了Curator，而不是ZKClient，或者原生的zookeeper原生的客户端。

先看一下Elastic-Job引入的依赖：

《elastic-job 源码解读之从源码看zookeeper的API的使用》 curator.png

从图中看，引入了curator-client, curator-framework,curator-recipes。他们各自是做了哪些事情？(答案来自搜索引擎)

curator-framework：对zookeeper的底层api的一些封装
curator-client：提供一些客户端的操作，例如重试策略等
curator-recipes：封装了一些高级特性，如：Cache事件监听、选举、分布式锁、分布式计数器、分布式Barrier等
注册中心的设计.png
看类图，在elasticJob中，所有的关于zk的操作都会交给ZookeeperRegistryCenter该对象去协调，在这里去初始化zookeeper，操作节点，使用缓存等。

elastic-Job创建zk的Client方法：

public void init() {
   CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory
     .builder()
     .connectString(zkConfig.getServerLists()) 
     .retryPolicy(new ExponentialBackoffRetry( //重试策略
       zkConfig.getBaseSleepTimeMilliseconds()
       , zkConfig.getMaxRetries(), zkConfig.getMaxSleepTimeMilliseconds()))
     .namespace(zkConfig.getNamespace());//设置namespace
  builder.sessionTimeoutMs(zkConfig.getSessionTimeoutMilliseconds());//session超时时间 if条件忽略
  builder.connectionTimeoutMs(zkConfig.getConnectionTimeoutMilliseconds());//连接超时时间 if条件忽略
  builder.authorization("digest", zkConfig.getDigest().getBytes(Charsets.UTF_8))
    //令牌验证 if条件忽略
                    .aclProvider(new ACLProvider() {
                        @Override
                        public List<ACL> getDefaultAcl() {
                            return ZooDefs.Ids.CREATOR_ALL_ACL;
                        }
                        @Override
                        public List<ACL> getAclForPath(final String path) {
                            return ZooDefs.Ids.CREATOR_ALL_ACL;
                        }
                    });
        }
        client = builder.build(); //创建了client
        client.start();
 try {
   //阻塞，等到连接成功，或者连接重试超时
   if (!client.blockUntilConnected(
        zkConfig.getMaxSleepTimeMilliseconds() * 
        zkConfig.getMaxRetries(), TimeUnit.MILLISECONDS)) {
                client.close();
                throw new KeeperException.OperationTimeoutException();
            }
        } catch (final Exception ex) {
            RegExceptionHandler.handleException(ex);
        }
｝

关闭zookeeper：

 @Override
    public void close() {
        for (Entry<String, TreeCache> each : caches.entrySet()) {
            each.getValue().close(); //先关缓存
        }
       //等到缓存关闭 这里是用线程等待 因为关闭缓存和client都是异步，关闭client之前需要关闭缓存
        waitForCacheClose(); // Thread.sleep(500L);
        CloseableUtils.closeQuietly(client); //关闭zkClient
    }

获取节点：

  @Override
    public String get(final String key) {
        TreeCache cache = findTreeCache(key); //从caches 中获取treeCache
        if (null == cache) {
            return getDirectly(key); //获取不到，直接从注册中心获取 
        }
        ChildData resultInCache = cache.getCurrentData(key);
        //缓存中获取不到，再从注册中心获取
        if (null != resultInCache) {
            return null == resultInCache.getData() ? null : new String(resultInCache.getData(), Charsets.UTF_8);
        }
        return getDirectly(key);
    }

直接从注册中心获取，使用api:client.getData().forPath(key)：

 @Override
    public String getDirectly(final String key) {
        try {
            return new String(client.getData().forPath(key), Charsets.UTF_8);
        //CHECKSTYLE:OFF
        } catch (final Exception ex) {
        //CHECKSTYLE:ON
            RegExceptionHandler.handleException(ex);
            return null;
        }
    }

获取注册中心子节点集合：client.getChildren().forPath(key)

 @Override
    public List<String> getChildrenKeys(final String key) {
        try {
            List<String> result = client.getChildren().forPath(key);
            Collections.sort(result, new Comparator<String>() {
                
                @Override
                public int compare(final String o1, final String o2) {
                    return o2.compareTo(o1);
                }
            });
            return result;
         //CHECKSTYLE:OFF
        } catch (final Exception ex) {
        //CHECKSTYLE:ON
            RegExceptionHandler.handleException(ex);
            return Collections.emptyList();
        }
    }

获取子节点数量：client.checkExists().forPath(key).getNumChildren();

    @Override
    public int getNumChildren(final String key) {
        try {
            Stat stat = client.checkExists().forPath(key);
            if (null != stat) {
                return stat.getNumChildren();
            }
            //CHECKSTYLE:OFF
        } catch (final Exception ex) {
            //CHECKSTYLE:ON
            RegExceptionHandler.handleException(ex);
        }
        return 0;
    }

判断节点是否存在：null!=client.checkExists().forPath(key)

 @Override
    public boolean isExisted(final String key) {
        try {
            return null != client.checkExists().forPath(key);
        //CHECKSTYLE:OFF
        } catch (final Exception ex) {
        //CHECKSTYLE:ON
            RegExceptionHandler.handleException(ex);
            return false;
        }
    }

创建节点：client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(key,value);

//判断节点是否存在，存在更新，不存在创建
@Override
    public void persist(final String key, final String value) {
        try {
            if (!isExisted(key)) {
            client.create().creatingParentsIfNeeded()
              .withMode(CreateMode.PERSISTENT)
              .forPath(key, value.getBytes(Charsets.UTF_8));
            } else {
                update(key, value);
            }
        //CHECKSTYLE:OFF
        } catch (final Exception ex) {
        //CHECKSTYLE:ON
            RegExceptionHandler.handleException(ex);
        }
    }

更新节点：client.inTransaction().check().forPath(key).and().setData().forPath(key,value).and().commit();

@Override
    public void update(final String key, final String value) {
        try {
            client.inTransaction().check()
              .forPath(key)
              .and()
              .setData()
              .forPath(key, value.getBytes(Charsets.UTF_8)).and().commit();
        //CHECKSTYLE:OFF
        } catch (final Exception ex) {
        //CHECKSTYLE:ON
            RegExceptionHandler.handleException(ex);
        }
    }

插入临时节点client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(key,value);

//如果节点存在，先删除，然后再创建临时节点   
@Override
    public void persistEphemeral(final String key, final String value) {
        try {
            if (isExisted(key)) {
                client.delete().deletingChildrenIfNeeded().forPath(key);
            }
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(key, value.getBytes(Charsets.UTF_8));
        //CHECKSTYLE:OFF
        } catch (final Exception ex) {
        //CHECKSTYLE:ON
            RegExceptionHandler.handleException(ex);
        }
    }

持久化顺序注册数据client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(key,value);

@Override
    public String persistSequential(final String key, final String value) {
        try {
            return client
              .create()
              .creatingParentsIfNeeded()
              .withMode(CreateMode.PERSISTENT_SEQUENTIAL)
              .forPath(key, value.getBytes(Charsets.UTF_8));
        //CHECKSTYLE:OFF
        } catch (final Exception ex) {
        //CHECKSTYLE:ON
            RegExceptionHandler.handleException(ex);
        }
        return null;
    }

持久化顺序临时注册数据client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(key,value);

@Override
    public String persistEphemeralSequential(final String key, final String value) {
        try {
            return client
              .create()
              .creatingParentsIfNeeded()
              .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
              .forPath(key, value.getBytes(Charsets.UTF_8));
        //CHECKSTYLE:OFF
        } catch (final Exception ex) {
        //CHECKSTYLE:ON
            RegExceptionHandler.handleException(ex);
        }
        return null;
    }

删除节点：client.delete().deletingChildrenIfNeeded().forPath(key);

@Override
    public void remove(final String key) {
        try {
            client.delete().deletingChildrenIfNeeded().forPath(key);
        //CHECKSTYLE:OFF
        } catch (final Exception ex) {
        //CHECKSTYLE:ON
            RegExceptionHandler.handleException(ex);
        }
    }

获取节点注册时间：client.checkExists().forPath(key).getMtime();，

注册中心时间为创建一个临时节点，然后获取该节点时间。

 @Override
    public long getRegistryCenterTime(final String key) {
        long result = 0L;
        try {
            persist(key, "");
            result = client.checkExists().forPath(key).getMtime();
        //CHECKSTYLE:OFF
        } catch (final Exception ex) {
        //CHECKSTYLE:ON
            RegExceptionHandler.handleException(ex);
        }
        Preconditions.checkState(0L != result, "Cannot get registry center time.");
        return result;
    }

添加缓存数据： new TreeCache(client, cachePath).start();

@Override
    public void addCacheData(final String cachePath) {
        TreeCache cache = new TreeCache(client, cachePath);
        try {
            cache.start();
        //CHECKSTYLE:OFF
        } catch (final Exception ex) {
        //CHECKSTYLE:ON
            RegExceptionHandler.handleException(ex);
        }
        caches.put(cachePath + "/", cache);
    }

删除缓存：将缓存从map中删除，关闭treeCache

@Override
    public void evictCacheData(final String cachePath) {
        TreeCache cache = caches.remove(cachePath + "/");
        if (null != cache) {
            cache.close();
        }
    }

    原文作者：一滴水的坚持
    原文地址: https://www.jianshu.com/p/cc7efd91b904
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。