此前我们曾用万字长文解释了Apache DolphinScheduler的Worker模块源码，今天，我们再来一起看看Master模块源码的原理。
Master Slot盘算
核心代码逻辑：
org.apache.dolphinscheduler.server.master.registry.MasterSlotManager.SlotChangeListener#notify- public void notify(Map<String, MasterHeartBeat> masterNodeInfo) {
- List<Server> serverList = masterNodeInfo.values().stream()
- // TODO 这里其实就是过滤掉buzy的master节点
- .filter(heartBeat -> !heartBeat.getServerStatus().equals(ServerStatus.BUSY))
- .map(this::convertHeartBeatToServer).collect(Collectors.toList());
- // TODO 同步master节点
- syncMasterNodes(serverList);
- }
- private void syncMasterNodes(List<Server> masterNodes) {
- slotLock.lock();
- try {
- this.masterPriorityQueue.clear();
- // TODO 这里会把所有的master节点都放入到masterPriorityQueue中，比如说 [192.168.220.1:12345,192.168.220.2:12345]
- this.masterPriorityQueue.putAll(masterNodes);
- // TODO 就是获取到本地ip的在队列中的位置
- int tempCurrentSlot = masterPriorityQueue.getIndex(masterConfig.getMasterAddress());
- // TODO 所有节点数量
- int tempTotalSlot = masterNodes.size();
- // TODO 正常情况下不会小于0
- if (tempCurrentSlot < 0) {
- totalSlot = 0;
- currentSlot = 0;
- log.warn("Current master is not in active master list");
- } else if (tempCurrentSlot != currentSlot || tempTotalSlot != totalSlot) {
- // TODO 这里其实就是记录的是比如说一共有两个slot，我的slot是0或者1
- totalSlot = tempTotalSlot;
- currentSlot = tempCurrentSlot;
- log.info("Update master nodes, total master size: {}, current slot: {}", totalSlot, currentSlot);
- }
- } finally {
- slotLock.unlock();
- }
- }
- public void putAll(Collection<Server> serverList) {
- for (Server server : serverList) {
- this.queue.put(server);
- }
- // TODO 这里更新了hostIndexMap，存放的是 <host:port> -> 索引
- refreshMasterList();
- }
- private void refreshMasterList() {
- hostIndexMap.clear();
- Iterator<Server> iterator = queue.iterator();
- int index = 0;
- while (iterator.hasNext()) {
- Server server = iterator.next();
- String addr = NetUtils.getAddr(server.getHost(), server.getPort());
- hostIndexMap.put(addr, index);
- index += 1;
- }
- }
command终极的获取逻辑:- 比如说两个Master节点 :
- masterCount=2 thisMasterSlot=0 master1
- masterCount=2 thisMasterSlot=1 master2
- command中的数据如下 :
- 1 master2
- 2 master1
- 3 master2
- 4 master1
- select *
- from t_ds_command
- where id % #{masterCount} = #{thisMasterSlot}
- order by process_instance_priority, id asc
- limit #{limit}
- 比如说，master1节点是这样的
- 1 master2
- 2 master1
- 3 master2
- 4 master1
- 比如说，master2节点是这样的，是不是发现master2节点都是他的，都可以拉取消费？那就导致重复消费，比如说1这个command
- 1 master1
- 2 master1
- 3 master1
- 4 master1
- @Transactional
- public @Nullable ProcessInstance handleCommand(String host,
- Command command) throws CronParseException, CodeGenerateException {
- // TODO 创建流程实例
- ProcessInstance processInstance = constructProcessInstance(command, host);
- // cannot construct process instance, return null
- if (processInstance == null) {
- log.error("scan command, command parameter is error: {}", command);
- commandService.moveToErrorCommand(command, "process instance is null");
- return null;
- }
- processInstance.setCommandType(command.getCommandType());
- processInstance.addHistoryCmd(command.getCommandType());
- processInstance.setTestFlag(command.getTestFlag());
- // if the processDefinition is serial
- ProcessDefinition processDefinition = this.findProcessDefinition(processInstance.getProcessDefinitionCode(),
- processInstance.getProcessDefinitionVersion());
- // TODO 是否是串行执行
- if (processDefinition.getExecutionType().typeIsSerial()) {
- saveSerialProcess(processInstance, processDefinition);
- if (processInstance.getState() != WorkflowExecutionStatus.RUNNING_EXECUTION) {
- setSubProcessParam(processInstance);
- triggerRelationService.saveProcessInstanceTrigger(command.getId(), processInstance.getId());
- deleteCommandWithCheck(command.getId());
- // todo: this is a bad design to return null here, whether trigger the task
- return null;
- }
- } else {
- // TODO 并行执行
- processInstanceDao.upsertProcessInstance(processInstance);
- }
- // TODO 这里其实还会向triggerRelation表中插入一条数据，是流程实例和triggerCode的关系
- triggerRelationService.saveProcessInstanceTrigger(command.getId(), processInstance.getId());
- // TODO 设置子流程参数
- setSubProcessParam(processInstance);
- // TODO 删除command
- deleteCommandWithCheck(command.getId());
- return processInstance;
- }
工作流启动流程
DAG切分 & 任务提交
Master事件状态流转
图连接 : Master事件状态流转
TaskEventService组件中的TaskEventDispatchThread(线程)和TaskEventHandlerThread(线程)解析
其实就是Master本身状态(DISPATCH)和Worker汇报上来的状态(RUNNING、UPDATE_PID、RESULT)都会放入到eventQueue，TaskEventDispatchThread(线程)会阻塞的方式举行获取，然后放入到对应的TaskExecuteRunnable中(留意 : 不执行的)，只有通过TaskEventHandlerThread(线程)才会使用TaskExecuteThreadPool线程举行TaskExecuteRunnable的提交。
转载自Journey
原文链接：https://segmentfault.com/a/1190000044992842
本文由 白鲸开源 提供发布支持！
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
|
