DolphinScheduler技术分析（十）

2021SC@SDUSC

这篇分析一下十分重要的workserver类

首先

stop方法

首先调用Stopper.stop设置全局变量。停止所有线程的“死”循环，然后休眠3秒，停止worker心跳。

停止worker任务线程池。停止killExecutor线程池。停止fetchTask线程池。停止zookeeper客户端。

ZKWorkerClient

初始化过程，调用initSystemZNode、listenerWorker、registWorker。

protected void initSystemZNode(){try {createNodePath(getMasterZNodeParentPath());createNodePath(getWorkerZNodeParentPath());createNodePath(getDeadZNodeParentPath());} catch (Exception e) {logger.error("init system znode failed : " + e.getMessage(),e);}
}

listenerWorker就是监听worker的CHILD_REMOVED事件，监听到该事件之后，调用了checkServerSelfDead。

private void listenerWorker(){workerPathChildrenCache = new PathChildrenCache(zkClient, getZNodeParentPath(ZKNodeType.WORKER), true, defaultThreadFactory);try {workerPathChildrenCache.start();workerPathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {@Overridepublic void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {switch (event.getType()) {case CHILD_ADDED:logger.info("node added : {}" ,event.getData().getPath());break;case CHILD_REMOVED:String path = event.getData().getPath();//find myself deadString serverHost = getHostByEventDataPath(path);if(checkServerSelfDead(serverHost, ZKNodeType.WORKER)){return;}break;case CHILD_UPDATED:break;default:break;}}});}catch (Exception e){logger.error("monitor worker failed : " + e.getMessage(),e);}
}

registerServer首先检查了当前节点是否存在，存在则退出；不存在则创建节点。最后调用handleDeadServer，就是查找死掉的节点，然后从zk中删除。

public String registerServer(ZKNodeType zkNodeType) throws Exception {String registerPath = null;String host = OSUtils.getHost();if(checkZKNodeExists(host, zkNodeType)){logger.error("register failure , {} server already started on host : {}" ,zkNodeType.toString(), host);return registerPath;}registerPath = createZNodePath(zkNodeType);// handle dead serverhandleDeadServer(registerPath, zkNodeType, Constants.DELETE_ZK_OP);return registerPath;
}

listenerWorker就是监听当前节点是否超时被zookeeper删除，删除后则调用stop方法，退出。

ZKMasterClient

initDao。初始化alertDao，调用DaoFactory.getDaoInstance(AlertDao.class)。
申请/dolphinscheduler/lock/failover/master路径的分布式锁。
申请到锁之后，依次调用initSystemZNode、listenerMaster、listenerWorker、registerMaster

private ZKMasterClient(ProcessDao processDao){this.processDao = processDao;init();
}

removeZKNodePath方法

主要是处理死掉的节点、进行预警，如果要进行故障转移，就调用failoverServerWhenDown。

failoverMaster

查询了指定master节点运行的流程定义实例，然后调动processNeedFailoverProcessInstances进行处理。

private void failoverMaster(String masterHost) {logger.info("start master failover ...");List<ProcessInstance> needFailoverProcessInstanceList = processDao.queryNeedFailoverProcessInstances(masterHost);//updateProcessInstance host is null and insert into commandfor(ProcessInstance processInstance : needFailoverProcessInstanceList){processDao.processNeedFailoverProcessInstances(processInstance);}logger.info("master failover end");
}

DolphinScheduler故障转移

worker如果与zookeeper连接超时，则停止心跳，停止获取任务，等待所有任务实例执行结束（正常或失败）并更新数据库状态
master如果与zookeeper连接超时，则停止心跳，停止获取流程定义实例，停止调度所有流程定义实例
master如果发现某个流程定义实例中的任务实例失败且属于ExecutionStatus.NEED_FAULT_TOLERANCE状态，则重新运行。