分布式调度框架elasticJob对Curator的使用解析

前言：

由于公司使用到elasticJob作为分布式调度框架的基础，所以也多多少少研究了一下这个框架。

任务调度的框架有很多，单机情况下我们可以用Quartz，但是分布式调度的情况下，Quartz就无能为力了，这个时候分布式调度就上线了。

当然这篇博客不是用来介绍elasticjob的使用，而是介绍elasticjob对zookeeper的使用。

正好也与上一篇博客相呼应，上一篇笔者自己封装了关于Curator框架的使用（地址：https://blog.csdn.net/qq_26323323/article/details/104441345 ）。这篇就分析一下大牛对Curator的封装使用，也算是找找差距。

任务背景：

笔者不打算用elasticjob的示例，因为实在太复杂（当然，代码肯定还是会拷一部分的），所以笔者就设定了一个任务，背景如下：

项目A（多份部署），有多个任务（job），每个job启动后会去进行主节点选举，成为主节点的那台服务则执行job的具体内容，同时job所在服务器注册到instance中。

根据以上任务，则zookeeper注册节点信息为：

namespace：[域值，所有job的前缀]

jobA：[具体任务名称]

leader：

election:

instance:[aaa] -- 所在服务器地址

latch:

instances:[aaa,bbb]

Curator框架的封装：

看看大佬们对Curator的使用，感觉实在是惭愧。

java的三大牛逼之处（封装、继承、多态），在我身上完全没有体现到。好吧，话不多说，转入正题。

1.面向接口而不是类编程

这句话耳熟能详，但是感受一直不强烈。看完大佬的代码之后，问了自己一个问题，如果以后有一个更牛逼的框架，我们想引入进来，那么对目前的项目影响度几何？

按照我目前的设计方式，那么影响是100%，因为我完全面向类编程，所以所有使用到这个类的地方都需要替换掉。所以这个是硬伤。

下面看看大佬的方案：

1）首先设计一个接口，里面包含最基本的增删改查的方法。

public interface RegistryCenter {/*** 初始化注册中心.*/void init();/*** 关闭注册中心.*/void close();/*** 获取注册数据.* * @param key 键* @return 值*/String get(String key);/*** 获取数据是否存在.* * @param key 键* @return 数据是否存在*/boolean isExisted(String key);/*** 持久化注册数据.* * @param key 键* @param value 值*/void persist(String key, String value);/*** 更新注册数据.* * @param key 键* @param value 值*/void update(String key, String value);/*** 删除注册数据.* * @param key 键*/void remove(String key);/*** 获取注册中心当前时间.* * @param key 用于获取时间的键* @return 注册中心当前时间*/long getRegistryCenterTime(String key);/*** 直接获取操作注册中心的原生客户端.* 如：Zookeeper或Redis等原生客户端.* * @return 注册中心的原生客户端*/Object getRawClient();
}

这个接口里面包含了我们操作的一些必须方法，我们可以把这个作为基础接口，后续操作可以面向基础接口编程。

2）接口分层级

接口为什么要分层级？这个问题想不太明白，但是看到Curator把增删改查的操作交由不同的操作类来处理就明白了。我们的特殊需求，可以交由不同的特定实现来处理，而这些特定实现也针对接口编程，这时候就有接口分层的需求了。

子接口设计如下：

public interface CoordinatorRegistryCenter extends RegistryCenter {/*** 直接从注册中心而非本地缓存获取数据.* * @param key 键* @return 值*/String getDirectly(String key);/*** 获取子节点名称集合.* * @param key 键* @return 子节点名称集合*/List<String> getChildrenKeys(String key);/*** 获取子节点数量.** @param key 键* @return 子节点数量*/int getNumChildren(String key);/*** 持久化临时注册数据.* * @param key 键* @param value 值*/void persistEphemeral(String key, String value);/*** 持久化顺序注册数据.** @param key 键* @param value 值* @return 包含10位顺序数字的znode名称*/String persistSequential(String key, String value);/*** 持久化临时顺序注册数据.* * @param key 键*/void persistEphemeralSequential(String key);/*** 添加本地缓存.* * @param cachePath 需加入缓存的路径*/void addCacheData(String cachePath);/*** 释放本地缓存.** @param cachePath 需释放缓存的路径*/void evictCacheData(String cachePath);/*** 获取注册中心数据缓存对象.* * @param cachePath 缓存的节点路径* @return 注册中心数据缓存对象*/Object getRawCache(String cachePath);
}

2.Curator的封装使用

看了大佬的方案，还看明白一个道理，我们是用什么技术来实现，不要直接暴露出去，一定要封装在内部，这样，对外不变，对内随意变。

这个时候才明白了封装的威力。以下是大佬的封装方案：

// 实现接口
public final class ZookeeperRegistryCenter implements CoordinatorRegistryCenter {// 这个类都是zookeeper相关配置项，封装在一个单独配置类中@Getter(AccessLevel.PROTECTED)private ZookeeperConfiguration zkConfig;// 真正的zookeeper操作client@Getterprivate CuratorFramework client;// 构造方法public ZookeeperRegistryCenter(final ZookeeperConfiguration zkConfig) {this.zkConfig = zkConfig;}// 初始化方法，根据参数初始化client@Overridepublic void init() {log.debug("Elastic job: zookeeper registry center init, server lists is: {}.", zkConfig.getServerLists());CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder().connectString(zkConfig.getServerLists()).retryPolicy(new ExponentialBackoffRetry(zkConfig.getBaseSleepTimeMilliseconds(), zkConfig.getMaxRetries(), zkConfig.getMaxSleepTimeMilliseconds())).namespace(zkConfig.getNamespace());if (0 != zkConfig.getSessionTimeoutMilliseconds()) {builder.sessionTimeoutMs(zkConfig.getSessionTimeoutMilliseconds());}if (0 != zkConfig.getConnectionTimeoutMilliseconds()) {builder.connectionTimeoutMs(zkConfig.getConnectionTimeoutMilliseconds());}...client = builder.build();client.start();try {if (!client.blockUntilConnected(zkConfig.getMaxSleepTimeMilliseconds() * zkConfig.getMaxRetries(), TimeUnit.MILLISECONDS)) {client.close();throw new KeeperException.OperationTimeoutException();}//CHECKSTYLE:OFF} catch (final Exception ex) {//CHECKSTYLE:ONRegExceptionHandler.handleException(ex);}}

以上就是实现类的初始化方法，通过ZookeeperConfiguration封装参数，在init方法中创建CuratorFramework client。

完美！

而至于具体操作就顺利成章了，笔者随意截取几个方法：

   @Overridepublic void remove(final String key) {try {client.delete().deletingChildrenIfNeeded().forPath(key);//CHECKSTYLE:OFF} catch (final Exception ex) {//CHECKSTYLE:ONRegExceptionHandler.handleException(ex);}}@Overridepublic 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:ONRegExceptionHandler.handleException(ex);}}

3.zookeeper路径的拼装操作

这个才是最棒的地方，设计的实在有点不可思议。正常来说，都是拼接各路径，然后直接操作即可。但是大佬们把这个过程也抽象出来，设计出多个类来实现。下面我们来一步步剖析。

1）拼装路径基类JobNodePath

@RequiredArgsConstructor
public final class JobNodePath {private static final String LEADER_HOST_NODE = "leader/election/instance";private static final String INSTANCES_NODE = "instances";// 不同的jobName对应不同的JobNodePathprivate final String jobName;// 拼装全路径public String getLeaderHostNodePath() {return String.format("/%s/%s", jobName, LEADER_HOST_NODE);}public String getInstancesNodePath() {return String.format("/%s/%s", jobName, INSTANCES_NODE);}public String getFullPath(final String node) {return String.format("/%s/%s", jobName, node);}

上面这个基础类，就把所有的拼装路径的动作囊括进来，而每个不同节点层（比如之前提到的leader、instances）的操作都抽象到高层次

2）instances节点操作类

public final class InstanceNode {/*** 运行实例信息根节点.这里的根节点是相对于namespace/jobName而言的*/public static final String ROOT = "instances";private static final String INSTANCES = ROOT + "/%s";private final String jobName;private final JobNodePath jobNodePath;public InstanceNode(final String jobName) {this.jobName = jobName;jobNodePath = new JobNodePath(jobName);}/*** 获取作业运行实例全路径.* 这里就是获取instance全路径的地方* @return 作业运行实例全路径*/public String getInstanceFullPath() {return jobNodePath.getFullPath(InstanceNode.ROOT);}// 其他操作public boolean isInstancePath(final String path) {return path.startsWith(jobNodePath.getFullPath(InstanceNode.ROOT));}

以上分层次操作每个不同节点路径，将公共的操作抽象到JobNodePath中，将个性化操作抽象到InstanceNode中。

4.实现Curator对zookeeper节点路径操作

实际就是如何将上面提到的2和3节整合起来，2节实现了对Curator的封装，实现了基本的操作；3则实现了节点路径的拼装操作，那么我们现在面临的问题就是如何将两者组合起来，真正实现对zookeeper的操作？

大佬的建议还是封装，将操作类和节点配置类封装在一起。下面来看下源码：

1）操作基类JobNodeStorage

public final class JobNodeStorage {// 操作类private final CoordinatorRegistryCenter regCenter;// 任务名称private final String jobName;// 节点路径拼装类private final JobNodePath jobNodePath;public JobNodeStorage(final CoordinatorRegistryCenter regCenter, final String jobName) {this.regCenter = regCenter;this.jobName = jobName;jobNodePath = new JobNodePath(jobName);}// 以下为具体操作public void fillEphemeralJobNode(final String node, final Object value) {regCenter.persistEphemeral(jobNodePath.getFullPath(node), value.toString());}public void removeJobNodeIfExisted(final String node) {if (isJobNodeExisted(node)) {regCenter.remove(jobNodePath.getFullPath(node));}}...

我们可以把JobNodeStorage看做节点操作的基类。

怎么类比呢？就好比我们常规web项目的baseDAO，下面就是service层，封装dao层操作

2）service层操作类InstanceService

public final class InstanceService {// baseDAO，操作类private final JobNodeStorage jobNodeStorage;   // 节点路径拼装类private final InstanceNode instanceNode;public InstanceService(final CoordinatorRegistryCenter regCenter, final String jobName) {jobNodeStorage = new JobNodeStorage(regCenter, jobName);instanceNode = new InstanceNode(jobName);}/*** 持久化作业运行实例上线相关信息.*/public void persistOnline() {jobNodeStorage.fillEphemeralJobNode(instanceNode.getLocalInstanceNode(), "");}/*** 删除作业运行状态.*/public void removeInstance() {jobNodeStorage.removeJobNodeIfExisted(instanceNode.getLocalInstanceNode());}...

就这样通过一层层的封装调用，实现所谓的高内聚低耦合。

总结：

先写到这吧，写的时间蛮久了。

实际把这些代码看下来也都是比较稀松平常的代码，没有什么太高难度，但是笔者再设计代码时从来没这么想过来这样设计。

虽说代码也敲了蛮多年了，但是大多数还是CRUD，源码也看了不少，但是更多的是看懂即可，浅尝辄止的感觉，很少看到他们的设计原理，只是单纯的觉得好，但是具体好在哪，还是很迷糊。

今天算是静下心来好好的分析一下大佬的代码，真心觉得是真的好。

之前总会觉得，搞那么多层次代码干嘛，都写在一起不好吗，但是当我们真正去实现功能的时候，尤其当我们修改别人的代码的时候，优秀的代码真心感觉不一样，很有层次感，动静分离，真正高内聚低耦合，向这个目标奋进！