海豚调度Dolphinscheduler源码分析（二）

项目结构

模块

1. dolphinscheduler-ui 前端页面模块
2. dolphinscheduler-server 核心模块。包括master/worker等功能
3. dolphinscheduler-common 公共模块。公共方法或类
4. dolphinscheduler-api Restful接口。前后端交互层，与master/worker交互等功能
5. dolphinscheduler-dao 数据操作层。实体定义、数据存储
6. dolphinscheduler-alert 预警模块。与预警相关的方法、功能
7. dolphinscheduler-dist 与编译、分发相关的模块。没有具体逻辑功能
8. dolphinscheduler-microbench 基准测试
9. dolphinscheduler-remote
10. dolphinscheduler-service 核心模块。zk客户端，包括日志等功能1.3版本中不再使用 GRPC 进行通信了

先来看dolphinscheduler-service 模块

zookeeper相关类源码分析：

1.zookeeperConfig类 -- zk的相关配置

相关代码入下：只粘贴了一些重要的

/**
 * zookeeper conf
 */
@Component
@PropertySource("classpath:zookeeper.properties")
public class ZookeeperConfig {

    //zk connect config
    @Value("${zookeeper.quorum:39.100.43.16:2181}")
    private String serverList;

    @Value("${zookeeper.retry.base.sleep:100}")
    private int baseSleepTimeMs;

    @Value("${zookeeper.retry.max.sleep:30000}")
    private int maxSleepMs;

    @Value("${zookeeper.retry.maxtime:10}")
    private int maxRetries;

    @Value("${zookeeper.session.timeout:60000}")
    private int sessionTimeoutMs;

    @Value("${zookeeper.connection.timeout:30000}")
    private int connectionTimeoutMs;

    @Value("${zookeeper.connection.digest: }")
    private String digest;

    @Value("${zookeeper.dolphinscheduler.root:/dolphinscheduler}")
    private String dsRoot;

    @Value("${zookeeper.max.wait.time:10000}")
    private int maxWaitTime;
}

其中大部分都是常见配置，如serverList，sessionTimeout 等，其中有一个配置没有见过故单独贴出来。

@Value("${zookeeper.connection.digest: }")
private String digest;

这个是zookeeper的ACL权限认证方式，读者可以自行查看相关资料，这里先不展开过多介绍:

相关连接：https://www.sohu.com/a/336744170_120173656

相关链接：https://www.cnblogs.com/dalianpai/p/12748144.html

相关链接：https://www.cnblogs.com/zwcry/p/10414001.html

---

 

2.CuratorZookeeperClient 类 初始化连接zookeeper客户端类

 

首先介绍一个框架Curator，Curator是Netflix公司开源的一套zookeeper客户端框架，解决了很多Zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册Watcher和NodeExistsException异常等等。Patrixck Hunt（Zookeeper）以一句“Guava is to Java that Curator to Zookeeper”给Curator予高度评价。

Curator有“馆长”，“管理者”之意。

Curator包含了几个包：

• curator-framework：对zookeeper的底层api的一些封装。
• curator-client：提供一些客户端的操作，例如重试策略等。
• curator-recipes：封装了一些高级特性，如：Cache事件监听、选举、分布式锁、分布式计数器、分布式Barrier等。

相关链接：http://www.throwable.club/2018/12/16/zookeeper-curator-usage/

//声明curator zk客户端对象
    private CuratorFramework zkClient;

    //通过springboot 创建bean对象之前，对bean进行初始化
    @Override
    public void afterPropertiesSet() throws Exception {
        this.zkClient = buildClient();
        initStateLister();
    }
    //通过工厂创建Curator
    private CuratorFramework buildClient() {
        logger.info("zookeeper registry center init, server lists is: {}.", zookeeperConfig.getServerList());

        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder().ensembleProvider(new DefaultEnsembleProvider(checkNotNull(zookeeperConfig.getServerList(),"zookeeper quorum can't be null")))
                .retryPolicy(new ExponentialBackoffRetry(zookeeperConfig.getBaseSleepTimeMs(), zookeeperConfig.getMaxRetries(), zookeeperConfig.getMaxSleepMs()));

        //these has default value
        if (0 != zookeeperConfig.getSessionTimeoutMs()) {
            builder.sessionTimeoutMs(zookeeperConfig.getSessionTimeoutMs());
        }
        if (0 != zookeeperConfig.getConnectionTimeoutMs()) {
            builder.connectionTimeoutMs(zookeeperConfig.getConnectionTimeoutMs());
        }
        //当配置文件中有digest认证方式，需要通过digest认证来获取权限
        if (StringUtils.isNotBlank(zookeeperConfig.getDigest())) {
            builder.authorization("digest", zookeeperConfig.getDigest().getBytes(StandardCharsets.UTF_8)).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;
                }
            });
        }
        zkClient = builder.build();
        zkClient.start();
        try {
            zkClient.blockUntilConnected();
        } catch (final Exception ex) {
            throw new RuntimeException(ex);
        }
        return zkClient;
    }

---

3.ZookeeperOperator类 ——zookeeper数据节点操作类（创建节点，删除节点，获取节点等）

这个类主要是对CuratorZookeeperClient 类创建的zkClient 客户端进行一些zk 数据阶段操作。

Zookeeper的节点创建模式：

• PERSISTENT：持久化
• PERSISTENT_SEQUENTIAL：持久化并且带序列号
• EPHEMERAL：临时
• EPHEMERAL_SEQUENTIAL：临时并且带序列号

    //递归创建持久节点
    public void persist(final String key, final String value) {
        try {
            if (!isExisted(key)) {
                zookeeperClient.getZkClient().create().creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT).forPath(key, value.getBytes(StandardCharsets.UTF_8));
            } else {
                update(key, value);
            }
        } catch (Exception ex) {
            logger.error("persist key : {} , value : {}", key, value, ex);
        }
    }

//创建递归顺序临时节点
    public void persistEphemeralSequential(final String key, String value) {
        try {
            zookeeperClient.getZkClient().create().creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(key, value.getBytes(StandardCharsets.UTF_8));
        } catch (final Exception ex) {
            logger.error("persistEphemeralSequential key : {}", key, ex);
        }
    }

其中这个creatingParentContainersIfNeeded()接口非常有用，因为一般情况开发人员在创建一个子节点必须判断它的父节点是否存在，如果不存在直接创建会抛出NoNodeException，使用creatingParentContainersIfNeeded()之后Curator能够自动递归创建所有所需的父节点。

---

4.ZookeeperCachedOperator类 缓存类

这个类继承了 上面的ZookeeperOperator类，并加入监听节点的cache方法

Zookeeper原生支持通过注册Watcher来进行事件监听，但是开发者需要反复注册(Watcher只能单次注册单次使用)。Cache是Curator中对事件监听的包装，可以看作是对事件监听的本地缓存视图，能够自动为开发者处理反复注册监听。Curator提供了三种Watcher(Cache)来监听结点的变化。

1.Path Cache

2.Node Cache

3.Tree Cache

Tree Cache可以监控整个树上的所有节点，类似于PathCache和NodeCache的组合，主要涉及到下面四个类：

• TreeCache - Tree Cache实现类
• TreeCacheListener - 监听器类
• TreeCacheEvent - 触发的事件类
• ChildData - 节点数据

**注意：**TreeCache在初始化(调用start()方法)的时候会回调TreeCacheListener实例一个事TreeCacheEvent，而回调的TreeCacheEvent对象的Type为INITIALIZED，ChildData为null，此时event.getData().getPath()很有可能导致空指针异常，这里应该主动处理并避免这种情况。

海豚调度 主要是使用Tree Cache缓存 主要代码入下：

//zk缓存类
//Curator提供了三种Watcher(Cache)来监听结点的变化。
//        1.Path Cache
//        2.Node Cache
//        3.Tree Cache 海豚调度是使用了Tree Cache缓存
public class ZookeeperCachedOperator extends ZookeeperOperator {

    private final Logger logger = LoggerFactory.getLogger(ZookeeperCachedOperator.class);

    private TreeCache treeCache;
    /**
     * register a unified listener of /${dsRoot},
     * 在/dolphinscheduler下注册监听节点
     */
    @Override
    protected void registerListener() {
        treeCache = new TreeCache(getZkClient(), getZookeeperConfig().getDsRoot() + "/nodes");
        logger.info("add listener to zk path: {}", getZookeeperConfig().getDsRoot());
        try {
            treeCache.start();
        } catch (Exception e) {
            logger.error("add listener to zk path: {} failed", getZookeeperConfig().getDsRoot());
            throw new RuntimeException(e);
        }

        treeCache.getListenable().addListener((client, event) -> {
            String path = null == event.getData() ? "" : event.getData().getPath();
            if (path.isEmpty()) {
                return;
            }
            dataChanged(client, event, path);
        });

    }

---

5.AbstractZKClient类—— 抽象zk客户端 （抽象类）

这个是对master，worker的抽象 

这个类继承了上面第4个类，并且是个抽象类，具体的实现有ZKMasterClient和ZookeeperMonitor类，具体的UML图入下

这个类有个分布式锁

提醒：

1.推荐使用ConnectionStateListener监控连接的状态，因为当连接LOST时你不再拥有锁

2.分布式的锁全局同步， 这意味着任何一个时间点不会有两个客户端都拥有相同的锁。

可重入共享锁—Shared Reentrant Lock

Shared意味着锁是全局可见的， 客户端都可以请求锁。 Reentrant和JDK的ReentrantLock类似，即可重入， 意味着同一个客户端在拥有锁的同时，可以多次获取，不会被阻塞。 它是由类InterProcessMutex来实现。 它的构造函数为：

public InterProcessMutex(CuratorFramework client, String path)

这个类用到锁的代码入下：

/**

     * 释放锁
     * release mutex
     * @param mutex mutex
     */
    public void releaseMutex(InterProcessMutex mutex) {
        if (mutex != null){
            try {
                mutex.release();
            } catch (Exception e) {
                if("instance must be started before calling this method".equals(e.getMessage())){
                    logger.warn("lock release");
                }else{
                    logger.error("lock release failed",e);
                }
            }
        }
    }

另外这个获取节点父路径的比较有意思：

/**
     *
     * @param zkNodeType zookeeper node type
     * @return get zookeeper node parent path
     */
    public String getZNodeParentPath(ZKNodeType zkNodeType) {
        String path = "";
        switch (zkNodeType){
            case MASTER:
                return getMasterZNodeParentPath();
            case WORKER:
                return getWorkerZNodeParentPath();
            case DEAD_SERVER:
                return getDeadZNodeParentPath();
            default:
                break;
        }
        return path;
    }

另一个特殊的地方是，官方文档说是创建master和worker节点都为临时节点，但是下面创建父路径的代码确是持久节点，而且zookeeper中，节点也为持久节点，可能是后期代码迭代，将znode节点改为了永久节点，但不知道其用意，等待后面的学习。

/**
     *  init system znode
     */
    protected void initSystemZNode(){
        try {
            persist(getMasterZNodeParentPath(), "");
            persist(getWorkerZNodeParentPath(), "");
            persist(getDeadZNodeParentPath(), "");

            logger.info("initialize server nodes success.");
        } catch (Exception e) {
            logger.error("init system znode failed",e);
        }
    }

persist这个方法之前介绍过，为创建持久节点，所以创建的Master，worker，deadznode节点都为持久节点。
查看下zookeeper的节点状态，发现都为持久节点：

 

好了，今天基本介绍了service模块下的zk相关代码，今天的源码阅读就到这里了！