Skip to content

B
byit-myth-job
Commit 5add5521 by huangfusuper
Options

功能： 

1. 添加特殊的补批方式
2. 添加工作流的公共参数
3. 修正脚本运行命令替换
4. 添加java任务返回值携带回写参数
parent f81a2a95
Showing with 956 additions and 311 deletions
byit-myth-admin/src/main/java/com/byit/api/ApiFlowOperatingController.java
package com.byit.api;
import com.byit.dto.specials.SpecialJobParam;
import com.byit.service.ApiFlowOperatingService;
import io.swagger.annotations.Api;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
......@@ -18,10 +19,19 @@ import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("api/operating/node/")
public class ApiFlowOperatingController {
private final ApiFlowOperatingService apiFlowOperatingService;
public ApiFlowOperatingController(ApiFlowOperatingService apiFlowOperatingService) {
this.apiFlowOperatingService = apiFlowOperatingService;
}
/**
* 特殊的补批接口 可以依照魔衣工作流下的某一个节点，自动补批其下所由的节点
* @param specialJobParam 采纳数信息
*/
@PostMapping("specialRunBatch")
public void specialRunBatch(@RequestBody SpecialJobParam specialJobParam){}
public void specialRunBatch(@RequestBody SpecialJobParam specialJobParam){
apiFlowOperatingService.specialRunBatch(specialJobParam);
}
}
byit-myth-admin/src/main/java/com/byit/service/ApiFlowOperatingService.java 0 → 100644
package com.byit.service;
import com.byit.dto.specials.SpecialJobParam;
/**
* 工作流操作实现解耦
*
* @author huangfu
* @date 2020年9月23日11:20:12
*/
public interface ApiFlowOperatingService {
/**
* 另外一种实现的补批接口
* 基于某一节点自动识别下面所有的节点
*
* @param specialJobParam 补批参数
*/
void specialRunBatch(SpecialJobParam specialJobParam);
}
byit-myth-admin/src/main/java/com/byit/service/impl/ApiFlowOperatingServiceImpl.java 0 → 100644
package com.byit.service.impl;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.byit.dto.NodeRelyDto;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.specials.SpecialJobParam;
import com.byit.enums.NodeTypeEnum;
import com.byit.enums.PlaceholderEnum;
import com.byit.enums.ScheduleTypeEnum;
import com.byit.job.utils.CurrentUserUtils;
import com.byit.job.utils.PlaceholderUtils;
import com.byit.mapper.RunRecordingMapper;
import com.byit.mapper.WaitingRecordMapper;
import com.byit.mapper.WaitingTaskMapper;
import com.byit.model.*;
import com.byit.service.*;
import com.byit.util.FlowNodeRelyParseUtil;
import com.byit.util.IDGenerationStrategy;
import com.byit.utils.ValidationUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;
/**
* 工作流操作实现
*
* @author huangfu
*/
@Slf4j
@Service
public class ApiFlowOperatingServiceImpl implements ApiFlowOperatingService {
private final static SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
public static final String PRE = "${";
public static final String SUFFER = "}";
@Value("${server.port}")
private Integer serverPort;
private final FlowService flowService;
private final WorkspaceService workspaceService;
private final NodeService nodeService;
private final NodeDependencyService nodeDependencyService;
private final CurrentUserUtils currentUserUtils;
private final WaitingRecordMapper waitingRecordMapper;
private final RunRecordingMapper runRecordingMapper;
private final WaitingTaskMapper waitingTaskMapper;
public ApiFlowOperatingServiceImpl(FlowService flowService, WorkspaceService workspaceService, NodeService nodeService,
NodeDependencyService nodeDependencyService, CurrentUserUtils currentUserUtils,
WaitingRecordMapper waitingRecordMapper, RunRecordingMapper runRecordingMapper, WaitingTaskMapper waitingTaskMapper) {
this.flowService = flowService;
this.workspaceService = workspaceService;
this.nodeService = nodeService;
this.nodeDependencyService = nodeDependencyService;
this.currentUserUtils = currentUserUtils;
this.waitingRecordMapper = waitingRecordMapper;
this.runRecordingMapper = runRecordingMapper;
this.waitingTaskMapper = waitingTaskMapper;
}
@Override
public void specialRunBatch(SpecialJobParam specialJobParam) {
//获取操作人
//String operator = currentUserUtils.account();
String operator = "huangfusuper";
//生成本次补批的运行标识
String runId = IDGenerationStrategy.runIdGenerationStrategy(serverPort);
//工作空间名称
String workspaceName = specialJobParam.getWorkspaceName();
ValidationUtil.dataNotBank(workspaceName, "工作空间名称不允许为空");
//那个工作流
String flowName = specialJobParam.getFlowName();
ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空");
//补批节点的所由参数
Map<String, String> paramCarrier = specialJobParam.getParamCarrier();
//开始节点任务名称
String startNodeTaskName = specialJobParam.getStartNodeTaskName();
ValidationUtil.dataNotBank(startNodeTaskName, "开始节点名称不允许为空");
//补批的类型 跑不跑本节点 startNodeTaskName
String supplementStatus = specialJobParam.getSupplementStatus();
ValidationUtil.dataNotBank(supplementStatus, "补批的类型不允许为空");
//获取公共参数
Map<String, String> publicParam = specialJobParam.getPublicParam();
//补批时间
List<String> repairTimeList = specialJobParam.getRepairTimeList();
ValidationUtil.isTrueValidation(CollectionUtil.isEmpty(repairTimeList), "补批时间不允许为空");
//获取工作空间
Workspace workspace = workspaceService.getByName(workspaceName);
ValidationUtil.dataNotNull(workspace, String.format("查询不到%s对应的工作空间！", workspaceName));
log.debug("-----开始补批工作空间：{}下的工作流------", workspace);
//获取对应的工作流
Flow flowByName = flowService.findFlowByName(flowName, workspace.getWorkspaceId());
ValidationUtil.dataNotNull(workspace, String.format("查询不到%s对应的工作流！", flowName));
//查询对应工作流下的所由节点 当前版本
List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(flowByName.getFlowId());
ValidationUtil.isTrueValidation(CollectionUtil.isEmpty(nodeByFlowIdAndVersionName), String.format("查询到工作流%s下不存在任何节点！请联系调度中心人员！", flowName));
log.debug("-----查询到该工作流下有{}个节点------", nodeByFlowIdAndVersionName.size());
//筛选节点信息
List<Integer> integerList = nodeByFlowIdAndVersionName.stream().map(Node::getNodeId).collect(Collectors.toList());
List<NodeDependencyKey> allNodeDependencyKey = nodeDependencyService.findAllNodeDependencyKey(integerList);
//解析对应的版本依赖
Set<NodeRelyDto> nodeRelyDtoSet = FlowNodeRelyParseUtil.parseThisVersionNodeRely(nodeByFlowIdAndVersionName, startNodeTaskName, allNodeDependencyKey, "1".equals(supplementStatus));
if (CollectionUtil.isNotEmpty(nodeRelyDtoSet) && CollectionUtil.isNotEmpty(paramCarrier)) {
//循环所有的结果集
nodeRelyDtoSet.forEach(nodeWrapped -> {
Node node = nodeWrapped.getNode();
String nodeName = node.getNodeName();
//基于节点名称替换参数
String param = paramCarrier.get(nodeName);
if (StringUtils.isNoneBlank(param)) {
//获取包装逻辑
RunParamWrapped runParamWrapped = new RunParamWrapped();
runParamWrapped.setPrivateParam(param);
runParamWrapped.setPublicParamMap(publicParam);
//构建参数
node.setRunParam(JSON.toJSONString(runParamWrapped));
}
});
}
if (CollectionUtil.isNotEmpty(nodeRelyDtoSet)) {
//构建等待队列
List<WaitingRecord> waitingRecords = buildWaitingRecordList(flowByName, runId, nodeRelyDtoSet.size(), operator, repairTimeList);
waitingRecords.forEach(waitingRecordMapper::insertSelective);
List<RunRecording> runRecordings = buildRunRecordingList(waitingRecords);
List<WaitingTask> waitingTaskList = buildWaitingTaskResult(waitingRecords, nodeRelyDtoSet);
runRecordings.forEach(runRecordingMapper::saveRunRecording);
waitingTaskList.forEach(waitingTaskMapper::insertSelective);
}
}
/**
* 构建单个等待队列的节点
*
* @param waitingRecord 等待实例对垒
* @param nodeRelyDtoSet 节点对象
* @return 返回一个等待队的全部节点
*/
private List<WaitingTask> buildWaitingTask(WaitingRecord waitingRecord, Set<NodeRelyDto> nodeRelyDtoSet) {
return nodeRelyDtoSet.stream().map(nodeRelyDto -> {
Node node = nodeRelyDto.getNode();
String relyId = nodeRelyDto.getRelyId();
String repeatTime = waitingRecord.getRepeatTime();
WaitingTask waitingTask = new WaitingTask();
BeanUtils.copyProperties(node, waitingTask);
waitingTask.setNodeDepend(relyId);
waitingTask.setFlowName(waitingRecord.getFlowName());
waitingTask.setTriggerTime(waitingRecord.getTriggerTime());
waitingTask.setRunParam(node.getRunParam());
waitingTask.setOperator(waitingRecord.getOperator());
waitingTask.setScheduleType(ScheduleTypeEnum.REPAIR.getCode());
waitingTask.setWaitId(waitingRecord.getWaitId());
try {
Date repairDate = sdf.parse(repeatTime);
ValidationUtil.isTrueValidation(!repairDate.before(new Date()), "只能补过去时间的批次!");
} catch (ParseException e) {
log.error("补批日期不符合规范，例：20200101");
ValidationUtil.isTrueValidation(true, "补批日期不符合规范，例：20200101");
}
//补批只替换不是java的节点
if (!NodeTypeEnum.JAVA.getCode().equals(waitingTask.getJobType())) {
if (StringUtils.isNotEmpty(waitingTask.getRunParam())) {
RunParamWrapped runParamWrapped = JSON.parseObject(waitingTask.getRunParam(), RunParamWrapped.class);
String param = PlaceholderUtils.paramPlaceholder(runParamWrapped.getPrivateParam(), repeatTime);
//设置替换完成后的参数
runParamWrapped.setPrivateParam(param);
param = JSON.toJSONString(runParamWrapped);
//设置参数
waitingTask.setRunParam(param);
}
//获取原始命令
String runCommand = waitingTask.getRunCommand();
//替换参数命令
String replaceRunCommand = runCommand.replace(PRE + PlaceholderEnum.DATE_PLACEHOLDER.getName() + SUFFER, repeatTime);
waitingTask.setRunCommand(replaceRunCommand);
}
return waitingTask;
}).collect(Collectors.toList());
}
/**
* 构建所有实例的等待节点信息
*
* @param waitingRecords 等待实例
* @param nodeRelyDtoSet 节点集合
* @return 全部的节点信息
*/
private List<WaitingTask> buildWaitingTaskResult(List<WaitingRecord> waitingRecords, Set<NodeRelyDto> nodeRelyDtoSet) {
List<WaitingTask> waitingTaskListResult = new ArrayList<>(8);
waitingRecords.forEach(waitingRecord -> {
List<WaitingTask> waitingTaskList = buildWaitingTask(waitingRecord, nodeRelyDtoSet);
waitingTaskListResult.addAll(waitingTaskList);
});
return waitingTaskListResult;
}
/**
* 构建实例队列
*
* @param waitingRecords
* @return
*/
private List<RunRecording> buildRunRecordingList(List<WaitingRecord> waitingRecords) {
return waitingRecords.stream().map(waitingRecord -> {
RunRecording runRecording = new RunRecording();
BeanUtils.copyProperties(waitingRecord, runRecording);
runRecording.setFlowStatus("1");
runRecording.setTriggerTime(waitingRecord.getTriggerTime());
runRecording.setWorkspaceId(waitingRecord.getWorkspaceId());
return runRecording;
}).collect(Collectors.toList());
}
/**
* 构建等待队列
*
* @param flow 工作流
* @param runId 运行标识
* @param nodeCount 节点数量
* @param operator 操作人
* @param repairTimeList 补时间
* @return 等待队列集合
*/
private List<WaitingRecord> buildWaitingRecordList(Flow flow, String runId, Integer nodeCount, String operator, List<String> repairTimeList) {
//查看当前排队的工作流最大排队序号
Integer order = waitingRecordMapper.findOrderByFlowId(flow.getFlowId());
if (order == null) {
order = 0;
}
List<WaitingRecord> waitingRecords = new ArrayList<WaitingRecord>(2);
for (String repairTime : repairTimeList) {
WaitingRecord waitingRecord = new WaitingRecord();
BeanUtils.copyProperties(flow, waitingRecord);
//设置排期
BeanUtils.copyProperties(flow, waitingRecord);
waitingRecord.setFlowVersionName(flow.getVersionName());
waitingRecord.setRunId(runId);
waitingRecord.setFlowNodeCount(nodeCount);
waitingRecord.setOperator(operator);
waitingRecord.setScheduleType(ScheduleTypeEnum.REPAIR.getCode());
waitingRecord.setRepeatTime(repairTime);
waitingRecord.setWaitOrder(++order);
//需要按着时间先后来设置时间
waitingRecord.setTriggerTime(System.currentTimeMillis());
waitingRecords.add(waitingRecord);
}
return waitingRecords;
}
}
byit-myth-admin/src/main/java/com/byit/service/impl/ApiFlowServiceImpl.java
......@@ -7,6 +7,7 @@ import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.byit.dto.StatisticsConditionDto;
import com.byit.dto.api.DeleteDto;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.plugin.*;
import com.byit.enums.*;
import com.byit.enums.plugin.PluginNodeTypeEnum;
......@@ -102,7 +103,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
@Resource
private CurrentUserUtils currentUserUtils;
private void parseParam(String param, DeleteDto deletDto){
private void parseParam(String param, DeleteDto deletDto) {
ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
JSONObject jsonObject = JSON.parseObject(param);
//获取工作空间名称
......@@ -133,17 +134,18 @@ public class ApiFlowServiceImpl implements ApiFlowService {
FlowVersion flowVersion = saveFlowVersion(flow);
int checkFlowNode = apiNodeCheckService.checkFlowNode(flow.getFlowId());
log.warn("-------删除错误节点{}个--------",checkFlowNode);
log.warn("-------删除错误节点{}个--------", checkFlowNode);
log.debug("插件端通过API保存成功！");
}
/**
* 保存工作流
*
* @param pluginFlow
* @param isInnner
* @param workspaceId
*/
private Flow saveFlow(PluginFlow pluginFlow, Integer workspaceId, boolean isInnner) throws Exception {
private Flow saveFlow(PluginFlow pluginFlow, Integer workspaceId, boolean isInnner) throws Exception {
Flow flow = new Flow();
flow.setFlowName(pluginFlow.getName());
......@@ -160,12 +162,14 @@ public class ApiFlowServiceImpl implements ApiFlowService {
flow.setAlarmlAction(pluginFlow.getConfig().getAlarmlAction());
flow.setAlarmEmail(pluginFlow.getConfig().getAlarmEmail());
flow.setScanMark("1");
Map<String, String> publicParamMap = pluginFlow.getConfig().getPublicParam();
flow.setFlowDesc(JSON.toJSONString(publicParamMap));
//设置可执行次数
//若没设置或者设置为-1，则响应设置剩余的执行次数
if (null != pluginFlow.getConfig().getRepeatCount() && !"-1".equals(pluginFlow.getConfig().getRepeatCount())){
if (null != pluginFlow.getConfig().getRepeatCount() && !"-1".equals(pluginFlow.getConfig().getRepeatCount())) {
flow.setRepeatCount(pluginFlow.getConfig().getRepeatCount());
flow.setRemainingCount(pluginFlow.getConfig().getRepeatCount());
}else {
} else {
flow.setRepeatCount(-1);
}
flow.setScheduleFollow(StringUtils.isEmpty(pluginFlow.getConfig().getScheduleFollow()) ? "1" : pluginFlow.getConfig().getScheduleFollow());
......@@ -175,31 +179,31 @@ public class ApiFlowServiceImpl implements ApiFlowService {
flow.setFlowTimeout(null == pluginFlow.getConfig().getFlowTimeout() ? 1000 * 60 * 120 : pluginFlow.getConfig().getFlowTimeout());
Flow oldFlow = flowMapper.getByWorkSpaceAndName(workspaceId, pluginFlow.getName());
//判断是否是重发
if ((pluginFlow.isRePublish() && !isInnner) || (isInnner && null != oldFlow)){
if ((pluginFlow.isRePublish() && !isInnner) || (isInnner && null != oldFlow)) {
String versionName = pluginFlow.getVersionName();
flow.setFlowId(oldFlow.getFlowId());
//当版本号不存在
if(StringUtils.isBlank(versionName)){
if (StringUtils.isBlank(versionName)) {
String version = oldFlow.getVersionName();
Integer versionTag = JobVersionGenUtil.parsingVersionName(version);
flow.setVersionName(JobVersionGenUtil.generateVersionName(versionTag));
}else{
} else {
flow.setVersionName(versionName);
}
flowMapper.updateByIdSelective(flow);
updateNodeByFlow(pluginFlow.getNodeList(), flow);
updateNodeByFlow(pluginFlow.getNodeList(), flow, publicParamMap);
}else {
} else {
String versionName = pluginFlow.getVersionName();
if(StringUtils.isBlank(versionName)){
if (StringUtils.isBlank(versionName)) {
flow.setVersionName(JobVersionGenUtil.generateVersionName(1));
}else{
} else {
flow.setVersionName(versionName);
}
Integer flowId = flowMapper.insertSelective(flow);
saveNode(pluginFlow.getNodeList(), flow);
saveNode(pluginFlow.getNodeList(), flow, publicParamMap);
}
return flow;
......@@ -207,10 +211,11 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* 删除原有工作流下的节点
*
* @param pluginNodeList
* @param flow
*/
private void updateNodeByFlow(List<PluginBaseNode> pluginNodeList, Flow flow) throws Exception{
private void updateNodeByFlow(List<PluginBaseNode> pluginNodeList, Flow flow, Map<String, String> publicParamMap) throws Exception {
//将工作流下的所有节点设为不在工作流调度上
log.debug("将工作流【{}】所有的节点移下调度,并删除相关的依赖关系", flow.getFlowName());
......@@ -221,7 +226,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
//查找并删除工作流下的虚节点
List<Node> virtualNodeList = nodeMapper.findVirtualByFlowId(flow.getFlowId());
if (null != virtualNodeList && virtualNodeList.size() > 0){
if (null != virtualNodeList && virtualNodeList.size() > 0) {
for (Node node : virtualNodeList) {
Flow virtualFlow = new Flow();
virtualFlow.setFlowId(node.getMapFlowId());
......@@ -238,9 +243,10 @@ public class ApiFlowServiceImpl implements ApiFlowService {
//重新组建节点和依赖关系
HashMap<String, Integer> nameIdRel = new HashMap<>();
//保存节点信息
for(PluginBaseNode pluginNode : pluginNodeList){
for (PluginBaseNode pluginNode : pluginNodeList) {
Node node = new Node();
buildNode(pluginNode, flow, node);
buildNode(pluginNode, flow, node, publicParamMap);
Integer nodeId = nodeMapper.insertSelective(node);
nameIdRel.put(node.getNodeName(), node.getNodeId());
}
......@@ -250,15 +256,16 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* 保存工作流
*
* @param nodeList
* @param flow
*/
private void saveNode(List<PluginBaseNode> nodeList, Flow flow) throws Exception{
private void saveNode(List<PluginBaseNode> nodeList, Flow flow, Map<String, String> publicParamMap) throws Exception {
HashMap<String, Integer> nameIdRel = new HashMap<>();
//保存节点信息
for(PluginBaseNode pluginNode : nodeList){
for (PluginBaseNode pluginNode : nodeList) {
Node node = new Node();
buildNode(pluginNode, flow, node);
buildNode(pluginNode, flow, node, publicParamMap);
node.setNodeId(null);
Integer nodeId = nodeMapper.insertSelective(node);
nameIdRel.put(node.getNodeName(), node.getNodeId());
......@@ -268,12 +275,13 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* 设置节点信息
*
* @param pluginNode
* @param flow
* @param node
* @throws Exception
*/
private void buildNode(PluginBaseNode pluginNode, Flow flow, Node node) throws Exception{
private void buildNode(PluginBaseNode pluginNode, Flow flow, Node node, Map<String, String> publicParamMap) throws Exception {
node.setVersionName(flow.getVersionName());
node.setFlowId(flow.getFlowId());
......@@ -282,67 +290,73 @@ public class ApiFlowServiceImpl implements ApiFlowService {
node.setAuthor(pluginNode.getAuthor());
node.setNodeName(pluginNode.getName());
if (PluginNodeTypeEnum.FLOW.getCode().equals(pluginNode.getType())){
if (PluginNodeTypeEnum.FLOW.getCode().equals(pluginNode.getType())) {
PluginFlow pluginFlow = (PluginFlow) pluginNode;
pluginFlow.getConfig().setFlowCron(flow.getFlowCron());
//如果是内嵌工作流先保存工作流信息
Flow innerFlow = saveFlow(pluginFlow, flow.getWorkspaceId(), true);
node.setIsVirtual(NodePropertyEnum.IS_VIRTUAL.getCode());
node.setMapFlowId(innerFlow.getFlowId());
}else {
node.setHandlerName(((PluginNode)pluginNode).getHandlerName());
node.setJobType(((PluginNode)pluginNode).getJobType());
node.setRunSource(((PluginNode)pluginNode).getRunSource());
node.setRunSourceDesc(((PluginNode)pluginNode).getRunSourceDesc());
node.setRunCommand(((PluginNode)pluginNode).getRunCommand());
node.setRunParam(((PluginNode)pluginNode).getRunParam());
node.setSourcePrincipal(((PluginNode)pluginNode).getSourcePrincipal());
node.setBlockStrategy(StringUtils.isEmpty(((PluginNode)pluginNode).getConfig().getBlockStrategy()) ? "0" : ((PluginNode)pluginNode).getConfig().getBlockStrategy());
node.setGatewayToken(((PluginNode)pluginNode).getConfig().getGatewayToken());
node.setPluginToken(((PluginNode)pluginNode).getConfig().getPluginToken());
node.setNodeCron(((PluginNode)pluginNode).getConfig().getNodeCron());
node.setTriggerNextTime(StringUtils.isEmpty(((PluginNode)pluginNode).getConfig().getNodeCron()) ? null : new CronExpression(((PluginNode)pluginNode).getConfig().getNodeCron()).getNextValidTimeAfter(new Date()).getTime());
node.setPluginUrls(((PluginNode)pluginNode).getConfig().getPluginUrls());
node.setRoutingStrategy(StringUtils.isEmpty(((PluginNode)pluginNode).getConfig().getRoutingStrategy()) ? "RANDOM" : ((PluginNode)pluginNode).getConfig().getRoutingStrategy());
node.setPriority(StringUtils.isEmpty(((PluginNode)pluginNode).getConfig().getPriority()) || !"2".equals(((PluginNode)pluginNode).getConfig().getPriority()) ? "1" : ((PluginNode)pluginNode).getConfig().getPriority());
node.setScriptUrls(((PluginNode)pluginNode).getScriptUrls());
} else {
node.setHandlerName(((PluginNode) pluginNode).getHandlerName());
node.setJobType(((PluginNode) pluginNode).getJobType());
node.setRunSource(((PluginNode) pluginNode).getRunSource());
node.setRunSourceDesc(((PluginNode) pluginNode).getRunSourceDesc());
node.setRunCommand(((PluginNode) pluginNode).getRunCommand());
node.setRunParam(((PluginNode) pluginNode).getRunParam());
node.setSourcePrincipal(((PluginNode) pluginNode).getSourcePrincipal());
node.setBlockStrategy(StringUtils.isEmpty(((PluginNode) pluginNode).getConfig().getBlockStrategy()) ? "0" : ((PluginNode) pluginNode).getConfig().getBlockStrategy());
node.setGatewayToken(((PluginNode) pluginNode).getConfig().getGatewayToken());
node.setPluginToken(((PluginNode) pluginNode).getConfig().getPluginToken());
node.setNodeCron(((PluginNode) pluginNode).getConfig().getNodeCron());
node.setTriggerNextTime(StringUtils.isEmpty(((PluginNode) pluginNode).getConfig().getNodeCron()) ? null : new CronExpression(((PluginNode) pluginNode).getConfig().getNodeCron()).getNextValidTimeAfter(new Date()).getTime());
node.setPluginUrls(((PluginNode) pluginNode).getConfig().getPluginUrls());
node.setRoutingStrategy(StringUtils.isEmpty(((PluginNode) pluginNode).getConfig().getRoutingStrategy()) ? "RANDOM" : ((PluginNode) pluginNode).getConfig().getRoutingStrategy());
node.setPriority(StringUtils.isEmpty(((PluginNode) pluginNode).getConfig().getPriority()) || !"2".equals(((PluginNode) pluginNode).getConfig().getPriority()) ? "1" : ((PluginNode) pluginNode).getConfig().getPriority());
node.setScriptUrls(((PluginNode) pluginNode).getScriptUrls());
//设置失败重试
if (null != ((PluginNode)pluginNode).getConfig().getFailedRetryCount()){
node.setFailedRetryCount(((PluginNode)pluginNode).getConfig().getFailedRetryCount());
node.setFailedRetryInterval(null == ((PluginNode)pluginNode).getConfig().getFailedRetryInterval() ? 3000 : ((PluginNode)pluginNode).getConfig().getFailedRetryInterval());
}else {
if (null != ((PluginNode) pluginNode).getConfig().getFailedRetryCount()) {
node.setFailedRetryCount(((PluginNode) pluginNode).getConfig().getFailedRetryCount());
node.setFailedRetryInterval(null == ((PluginNode) pluginNode).getConfig().getFailedRetryInterval() ? 3000 : ((PluginNode) pluginNode).getConfig().getFailedRetryInterval());
} else {
node.setFailedRetryCount(0);
}
//设置剩余执行次数
if (null != ((PluginNode)pluginNode).getConfig().getRepeatCount() && !"-1".equals(((PluginNode)pluginNode).getConfig().getRepeatCount())){
node.setRemainingCount(((PluginNode)pluginNode).getConfig().getRepeatCount());
node.setRepeatCount(((PluginNode)pluginNode).getConfig().getRepeatCount());
}else {
if (null != ((PluginNode) pluginNode).getConfig().getRepeatCount() && !"-1".equals(((PluginNode) pluginNode).getConfig().getRepeatCount())) {
node.setRemainingCount(((PluginNode) pluginNode).getConfig().getRepeatCount());
node.setRepeatCount(((PluginNode) pluginNode).getConfig().getRepeatCount());
} else {
node.setRemainingCount(-1);
}
//判断是否在上级成功是才执行下级作业
if (null != ((PluginNode) pluginNode).getSuperSuccessRun()){
if (null != ((PluginNode) pluginNode).getSuperSuccessRun()) {
node.setSuperSuccessRun(((PluginNode) pluginNode).getSuperSuccessRun());
}else {
} else {
node.setSuperSuccessRun("0");
}
node.setIsVirtual(NodePropertyEnum.ISNOT_VIRTUAL.getCode());
RunParamWrapped runParamWrapped = new RunParamWrapped();
runParamWrapped.setPrivateParam(node.getRunParam());
runParamWrapped.setPublicParamMap(publicParamMap);
node.setRunParam(JSON.toJSONString(runParamWrapped));
}
}
/**
* 重新组织节点依赖关系
*
* @param nodeList
* @param nameIdRel
*/
public void buildDepend(List<PluginBaseNode> nodeList, HashMap<String, Integer> nameIdRel){
for (PluginBaseNode pluginNode : nodeList){
if (null != pluginNode.getDependNodeNameList() && pluginNode.getDependNodeNameList().size() > 0){
public void buildDepend(List<PluginBaseNode> nodeList, HashMap<String, Integer> nameIdRel) {
for (PluginBaseNode pluginNode : nodeList) {
if (null != pluginNode.getDependNodeNameList() && pluginNode.getDependNodeNameList().size() > 0) {
pluginNode.getDependNodeNameList().forEach(dependNodeName -> {
Integer nodeId = nameIdRel.get(pluginNode.getName());
Integer dependNodeId = nameIdRel.get(dependNodeName);
ValidationUtil.isTrueValidation(null == nodeId , pluginNode.getName() + "节点不存在！");
ValidationUtil.isTrueValidation(null == dependNodeId , "依赖的节点" + dependNodeName + "不存在！");
ValidationUtil.isTrueValidation(null == nodeId, pluginNode.getName() + "节点不存在！");
ValidationUtil.isTrueValidation(null == dependNodeId, "依赖的节点" + dependNodeName + "不存在！");
nodeDependencyMapper.insert(nodeId, dependNodeId);
});
}
......@@ -351,6 +365,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* 校验插件端请求参数是否合法
*
* @param pluginPackage
*/
private Workspace validate(PluginPackage pluginPackage) {
......@@ -384,6 +399,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* 校验工作流
*
* @param workspace
* @param flow
* @param flowNameSet
......@@ -393,7 +409,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
*/
private void checkFlow(Workspace workspace, PluginFlow flow, HashSet<String> flowNameSet, HashSet<String> fromFlowSet, HashSet<String> toFlowSet, HashMap<String, Integer> lineMap) {
//校验是否存在重名工作流
ValidationUtil.isTrueValidation(!flowNameSet.add(flow.getName()), "【"+ flow.getName() +"】工作流重复，同一个工作空间内工作流名称不允许重复！");
ValidationUtil.isTrueValidation(!flowNameSet.add(flow.getName()), "【" + flow.getName() + "】工作流重复，同一个工作空间内工作流名称不允许重复！");
fromFlowSet.add(flow.getName());
List<PluginBaseNode> nodeList = flow.getNodeList();
//用来校验同一工作流下是否有重复的节点
......@@ -402,24 +418,24 @@ public class ApiFlowServiceImpl implements ApiFlowService {
ValidationUtil.dataNotBank(node.getName(), "节点名称不允许为空！");
ValidationUtil.isTrueValidation(!nodeSet.add(node.getName()), "同一个工作流下任务节点不允许重复！");
//判断节点类型
if (PluginNodeTypeEnum.FLOW.getCode().equals(node.getType())){
if (PluginNodeTypeEnum.FLOW.getCode().equals(node.getType())) {
//校验node的类别
ValidationUtil.isTrueValidation(!(node instanceof PluginFlow), "工作流的type设置错误，只有当节点为内嵌工作流是type才允许设置为flow");
//校验父工作流和内嵌工作流的执行类型是否一致
ValidationUtil.isTrueValidation(!(flow.getConfig().getExecType().equals(((PluginFlow) node).getConfig().getExecType())), "内嵌工作流的调度配置必须和父工作流配置保持一致！");
validateFlow(workspace.getWorkspaceId(), (PluginFlow) node, true);
if(FlowPropertyEnum.NO_SCHEDULE.getCode().equals(flow.getConfig().getScheduleFollow())){
ValidationUtil.dataNotBank( ((PluginFlow) node).getConfig().getFlowCron(), "工作流设置为不跟随调度时节点必须设置调度时间！");
if (FlowPropertyEnum.NO_SCHEDULE.getCode().equals(flow.getConfig().getScheduleFollow())) {
ValidationUtil.dataNotBank(((PluginFlow) node).getConfig().getFlowCron(), "工作流设置为不跟随调度时节点必须设置调度时间！");
}
toFlowSet.add(node.getName());
lineMap.put(flow.getName() + "_" + node.getName(), 1);
checkFlow(workspace, (PluginFlow) node, flowNameSet, fromFlowSet, toFlowSet, lineMap);
}else {
} else {
ValidationUtil.dataNotBank(((PluginNode) node).getJobType(), "节点类型不允许为空！");
if ("start".equals(node.getName()) || "end".equals(node.getName())){
if ("start".equals(node.getName()) || "end".equals(node.getName())) {
((PluginNode) node).getConfig().setNodeCron(flow.getConfig().getFlowCron());
}
if(FlowPropertyEnum.NO_SCHEDULE.getCode().equals(flow.getConfig().getScheduleFollow())) {
if (FlowPropertyEnum.NO_SCHEDULE.getCode().equals(flow.getConfig().getScheduleFollow())) {
ValidationUtil.dataNotNull(((PluginNode) node).getConfig(), "节点配置信息不允许为空！");
ValidationUtil.dataNotBank(((PluginNode) node).getConfig().getNodeCron(), "工作流设置为不跟随调度时节点必须设置调度时间！");
}
......@@ -448,27 +464,28 @@ public class ApiFlowServiceImpl implements ApiFlowService {
ValidationUtil.dataNotNull(pluginFlow.getNodeList(), "工作流下属节点不允许为空！");
ValidationUtil.dataNotNull(pluginFlow.getConfig(), "工作流配置不允许为空！");
ValidationUtil.dataNotBank(pluginFlow.getConfig().getExecType(), "工作流的调度类型不允许为空！");
if (FlowPropertyEnum.SCHEDULE_MODE.getCode().equals(pluginFlow.getConfig().getExecType())){
if (FlowPropertyEnum.SCHEDULE_MODE.getCode().equals(pluginFlow.getConfig().getExecType())) {
ValidationUtil.dataNotBank(pluginFlow.getConfig().getFlowCron(), "工作流cron表达式不允许为空！");
ValidationUtil.isTrueValidation(!CronExpression.isValidExpression(pluginFlow.getConfig().getFlowCron()), "工作流cron表达式不符合规范！");
}
if (!FlowPropertyEnum.NO_ALARML.getCode().equals(pluginFlow.getConfig().getAlarmlAction())){
if (!FlowPropertyEnum.NO_ALARML.getCode().equals(pluginFlow.getConfig().getAlarmlAction())) {
ValidationUtil.dataNotBank(pluginFlow.getConfig().getAlarmEmail(), "设置告警时机时告警邮箱不允许为空！");
}
Flow flow = flowMapper.getByWorkSpaceAndName(workspaceId, pluginFlow.getName());
//判断是否是重发
//是重发
if (!isInner){
if (pluginFlow.isRePublish()){
ValidationUtil.dataNotNull(flow, "工作流【"+ pluginFlow.getName() + "】不存在！");
}else {//不是重发
ValidationUtil.isTrueValidation(null != flow, "工作流【"+ pluginFlow.getName() + "】已存在！");
if (!isInner) {
if (pluginFlow.isRePublish()) {
ValidationUtil.dataNotNull(flow, "工作流【" + pluginFlow.getName() + "】不存在！");
} else {//不是重发
ValidationUtil.isTrueValidation(null != flow, "工作流【" + pluginFlow.getName() + "】已存在！");
}
}
}
/**
* 重跑任务
*
* @param param
*/
@Override
......@@ -495,7 +512,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
RunRecording runRecording = runRecordingMapper.findRunRecordingByFlowIdAndRunId(flow.getFlowId(), runInfo.getRunId());
ValidationUtil.dataNotNull(runRecording, "查无此运行记录");
//TODO 这里应该查询的是版本表 与版本号挂钩
NodeVersion nodeVersion = nodeVersionMapper.findNodeByNodeNameAndVersionNameAndFlowId(runInfo.getNodeName(),runRecording.getFlowVersionName(),runRecording.getFlowId());
NodeVersion nodeVersion = nodeVersionMapper.findNodeByNodeNameAndVersionNameAndFlowId(runInfo.getNodeName(), runRecording.getFlowVersionName(), runRecording.getFlowId());
// //Node node = nodeMapper.getByNameAndFlow(runInfo.getNodeName(), flow.getFlowId());
// ValidationUtil.dataNotNull(node, runInfo.getNodeName() + "节点不存在");
......@@ -507,7 +524,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
String userName = currentUserUtils.account();
//校验上级是否成功
if (StringUtils.isNotEmpty(jobTaskRunLog.getNodeDepend())){
if (StringUtils.isNotEmpty(jobTaskRunLog.getNodeDepend())) {
List<String> dependNodeList = Arrays.asList(jobTaskRunLog.getNodeDepend().split(","));
dependNodeList.forEach(dependNodeId -> {
JobTaskRunLog dependJobTaskRunLog = jobTaskRunLogMapper.findByRunIdAndNodeId(runInfo.getRunId(), Integer.valueOf(dependNodeId));
......@@ -543,16 +560,16 @@ public class ApiFlowServiceImpl implements ApiFlowService {
runNodeIdSet.add(jobTaskRunLog.getNodeId());
nodeIdSet.add(jobTaskRunLog.getNodeId());
//获取全部的需要重跑的节点
if (subNodeList != null && subNodeList.size() > 0){
if (subNodeList != null && subNodeList.size() > 0) {
Queue<Integer> queue = new LinkedList<>();
subNodeList.forEach(subNode -> {
queue.offer(subNode.getNodeVersionId());
runNodeIdSet.add(subNode.getNodeId());
});
while(!queue.isEmpty()){
List<NodeVersion> innerSubNodeList = nodeVersionMapper.findSubNodeList(queue.poll());
if (innerSubNodeList != null && innerSubNodeList.size() > 0){
innerSubNodeList.forEach(innerSubNode->{
while (!queue.isEmpty()) {
List<NodeVersion> innerSubNodeList = nodeVersionMapper.findSubNodeList(queue.poll());
if (innerSubNodeList != null && innerSubNodeList.size() > 0) {
innerSubNodeList.forEach(innerSubNode -> {
queue.offer(innerSubNode.getNodeVersionId());
runNodeIdSet.add(innerSubNode.getNodeId());
});
......@@ -568,7 +585,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
BeanUtils.copyProperties(newRunRecording, waitingRecord);
Integer order = waitingRecordMapper.findOrderByFlowId(flow.getFlowId());
if (order == null){
if (order == null) {
order = 0;
}
waitingRecord.setWaitOrder(++order);
......@@ -584,11 +601,12 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* 构建运行实例
*
* @param runRecording 旧的运行实例
* @param triggerTime 本次触发时间
* @param reRunId 重跑的实例id
* @param scheduleNodeCount 本次实例的重跑节点数目
* @param userName 重跑操作人
* @param reRunId 重跑的实例id
* @param scheduleNodeCount 本次实例的重跑节点数目
* @param userName 重跑操作人
* @return
*/
private RunRecording buildRunRecording(RunRecording runRecording, Long triggerTime, String reRunId, int scheduleNodeCount, String userName) {
......@@ -616,7 +634,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
@Override
public void reRunFlow(String param){
public void reRunFlow(String param) {
ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
RunInfo runInfo = JSON.parseObject(param, RunInfo.class);
//获取工作空间名称
......@@ -635,12 +653,12 @@ public class ApiFlowServiceImpl implements ApiFlowService {
RunRecording runRecording = runRecordingMapper.findRunRecordingByFlowIdAndRunId(flow.getFlowId(), runInfo.getRunId());
ValidationUtil.dataNotNull(runRecording, "没有找到对应的运行记录");
//判断是否是内嵌工作流
if (FlowPropertyEnum.IS_INNER.getCode().equals(flow.getIsInner())){//是内嵌工作流
if (FlowPropertyEnum.IS_INNER.getCode().equals(flow.getIsInner())) {//是内嵌工作流
Node node = nodeMapper.getByMapFlowId(flow.getFlowId());
RunInfo nodeRun = RunInfo.builder().workspaceName(runInfo.getWorkspaceName()).flowName(runInfo.getFlowName())
.nodeName(node.getNodeName()).runId(runInfo.getRunId()).runState("1").build();
reRunJob(JSON.toJSONString(nodeRun, WriteClassName));
}else {
} else {
//不是内嵌工作流
Long triggerTime = System.currentTimeMillis();
String reRunId = IDGenerationStrategy.runIdGenerationStrategy(serverPort);
......@@ -654,7 +672,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
RunRecording newRunRecording = buildRunRecording(runRecording, triggerTime, reRunId, jobTaskRunLogList.size(), userName);
BeanUtils.copyProperties(newRunRecording, waitingRecord);
Integer order = waitingRecordMapper.findOrderByFlowId(flow.getFlowId());
if (order == null){
if (order == null) {
order = 0;
}
waitingRecord.setWaitOrder(++order);
......@@ -675,7 +693,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
//查询当前节点的依赖节点
List<Integer> dependNodeIdList = nodeDependencyMapper.findDependIdByNodeId(jobTaskRunLog.getNodeId());
if (dependNodeIdList != null && dependNodeIdList.size() > 0){
if (dependNodeIdList != null && dependNodeIdList.size() > 0) {
waitingTask.setNodeDepend(Joiner.on(",").join(dependNodeIdList));
}
waitingTask.setWaitId(waitingRecord.getWaitId());
......@@ -697,22 +715,22 @@ public class ApiFlowServiceImpl implements ApiFlowService {
waitingTask.setRunParam(jobTaskRunLog.getRunParams());
//查询当前节点的依赖节点
if (StringUtils.isNotEmpty(jobTaskRunLog.getNodeDepend())){
if (StringUtils.isNotEmpty(jobTaskRunLog.getNodeDepend())) {
List<String> dependNodeIdList = Arrays.asList(jobTaskRunLog.getNodeDepend().split(","));
List<Integer> realDependNodeIdList = new ArrayList<>();
for (String dependNodeId : dependNodeIdList) {
if (!runNodeIdSet.add(Integer.valueOf(dependNodeId))){
if (!runNodeIdSet.add(Integer.valueOf(dependNodeId))) {
realDependNodeIdList.add(Integer.valueOf(dependNodeId));
}
}
waitingTask.setNodeDepend(Joiner.on(",").join(realDependNodeIdList));
}
if (nodeIdSet.add(childNode.getNodeId())){
if (nodeIdSet.add(childNode.getNodeId())) {
waitingTaskList.add(waitingTask);
}
//查询依赖于当前节点的下级节点
List<NodeVersion> childNodeList = nodeVersionMapper.findSubNodeList(childNode.getNodeVersionId());
if (childNodeList != null && childNodeList.size() > 0){
if (childNodeList != null && childNodeList.size() > 0) {
addDependNode(runId, triggerTime, waitingTaskList, childNodeList, userName, nodeIdSet, runNodeIdSet);
}
});
......@@ -720,10 +738,11 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* 重跑任务
*
* @param param
*/
@Override
public void madeSuccess(String param){
public void madeSuccess(String param) {
ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
RunInfo runInfo = JSON.parseObject(param, RunInfo.class);
......@@ -749,7 +768,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
}
@Override
public List<RunRecording> loadScheduleResult(String param){
public List<RunRecording> loadScheduleResult(String param) {
ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
JSONObject jsonObject = JSON.parseObject(param);
Long startTime = jsonObject.getLong("startTime");
......@@ -758,7 +777,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
Long endTime = jsonObject.getLong("endTime");
ValidationUtil.dataNotNull(endTime, "结束时间不允许为空！");
ValidationUtil.isTrueValidation(String.valueOf(startTime).length() < 13 || String.valueOf(endTime).length() < 13 , "时间格式必须是毫秒");
ValidationUtil.isTrueValidation(String.valueOf(startTime).length() < 13 || String.valueOf(endTime).length() < 13, "时间格式必须是毫秒");
Date startDate = new Date(startTime);
Date endtDate = new Date(endTime);
long between = DateUtil.between(startDate, endtDate, DateUnit.DAY);
......@@ -769,16 +788,16 @@ public class ApiFlowServiceImpl implements ApiFlowService {
String workspaceName = jsonObject.getString("workspaceName");
String flowName = jsonObject.getString("flowName");
List<Integer> flowIds = new ArrayList<>();
if (StringUtils.isNotBlank(workspaceName)){
if (StringUtils.isNotBlank(workspaceName)) {
Workspace workspace = workspaceMapper.getByName(workspaceName);
ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
if (StringUtils.isNotBlank(flowName)){
if (StringUtils.isNotBlank(flowName)) {
Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
flowIds.add(flow.getFlowId());
}else {
} else {
List<Flow> flowList = flowMapper.findByWorkspace(workspace.getWorkspaceId());
if (flowList != null && flowList.size() > 0){
if (flowList != null && flowList.size() > 0) {
flowList.forEach(flow -> flowIds.add(flow.getFlowId()));
}
}
......@@ -786,12 +805,12 @@ public class ApiFlowServiceImpl implements ApiFlowService {
String scheduleStatus = jsonObject.getString("scheduleStatus");
List<String> scheduleStatusList = null;
if(StringUtils.isNotEmpty(scheduleStatus)){
if (StringUtils.isNotEmpty(scheduleStatus)) {
scheduleStatusList = Arrays.asList(scheduleStatus.split(","));
}
String executeStatus = jsonObject.getString("executeStatus");
List<String> executeStatusList = null;
if(StringUtils.isNotEmpty(executeStatus)){
if (StringUtils.isNotEmpty(executeStatus)) {
executeStatusList = Arrays.asList(executeStatus.split(","));
}
List<RunRecording> runRecordList = runRecordingMapper.findByStartAndEndTime(startTime, endTime, flowIds, scheduleStatusList, executeStatusList);
......@@ -819,7 +838,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
}
@Override
public CollectData loadStatisticData(String param){
public CollectData loadStatisticData(String param) {
ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
JSONObject jsonObject = JSON.parseObject(param);
//获取工作空间名称
......@@ -835,22 +854,22 @@ public class ApiFlowServiceImpl implements ApiFlowService {
Long endTime = jsonObject.getLong("endTime");
ValidationUtil.dataNotNull(endTime, "结束时间不允许为空！");
ValidationUtil.isTrueValidation(String.valueOf(startTime).length() < 13 || String.valueOf(endTime).length() < 13 , "时间格式必须是毫秒");
ValidationUtil.isTrueValidation(String.valueOf(startTime).length() < 13 || String.valueOf(endTime).length() < 13, "时间格式必须是毫秒");
List<Integer> flowIdList = new ArrayList<>();
List<Flow> flowList = flowMapper.findByWorkspace(workspace.getWorkspaceId());
if (flowList != null && flowList.size() > 0){
if (flowList != null && flowList.size() > 0) {
flowList.forEach(flow -> flowIdList.add(flow.getFlowId()));
}
//如果工作空间下面有数据
if (flowIdList.size() > 0){
return buildStatisticAllDate(startTime, endTime, workspace.getWorkspaceId(),null);
if (flowIdList.size() > 0) {
return buildStatisticAllDate(startTime, endTime, workspace.getWorkspaceId(), null);
}
return null;
}
//--------------------------------------------------------------------
private CollectData buildStatisticAllDate (Long startTime, Long endTime, Integer workspaceId, String flowName) {
private CollectData buildStatisticAllDate(Long startTime, Long endTime, Integer workspaceId, String flowName) {
StatisticsConditionDto statisticsConditionDto = new StatisticsConditionDto();
statisticsConditionDto.setStartTime(startTime);
statisticsConditionDto.setEndTime(endTime);
......@@ -858,7 +877,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
statisticsConditionDto.setFlowName(flowName);
//查询固定工作空间下 时间范围内的的所由工作流实例数目
List<RunRecording> betweenRunRecording = runRecordingMapper.findThisDayRunRecording(statisticsConditionDto);
Map<String, Long> resultMap = betweenRunRecording.stream().collect(Collectors.groupingBy(e ->{
Map<String, Long> resultMap = betweenRunRecording.stream().collect(Collectors.groupingBy(e -> {
String flowRunResult = e.getFlowRunResult();
if (StringUtils.isBlank(flowRunResult)) {
flowRunResult = "-1";
......@@ -868,29 +887,29 @@ public class ApiFlowServiceImpl implements ApiFlowService {
Map<String, Long> statusMap = betweenRunRecording.stream().collect(Collectors.groupingBy(RunRecording::getFlowStatus, Collectors.counting()));
//成功的
long successCount = resultMap.get("1") == null?0:resultMap.get("1");
long reSuccessCount = resultMap.get("3") == null?0:resultMap.get("3");
long successTotal = successCount+reSuccessCount;
long successCount = resultMap.get("1") == null ? 0 : resultMap.get("1");
long reSuccessCount = resultMap.get("3") == null ? 0 : resultMap.get("3");
long successTotal = successCount + reSuccessCount;
//失败的
long errorCount = resultMap.get("2") == null?0:resultMap.get("2");
long reErrorCount = resultMap.get("4") == null?0:resultMap.get("4");
long killCount = resultMap.get("5") == null?0:resultMap.get("5");
long errorTotal = errorCount+reErrorCount +killCount;
long errorCount = resultMap.get("2") == null ? 0 : resultMap.get("2");
long reErrorCount = resultMap.get("4") == null ? 0 : resultMap.get("4");
long killCount = resultMap.get("5") == null ? 0 : resultMap.get("5");
long errorTotal = errorCount + reErrorCount + killCount;
//运行中的
long runingFlow = statusMap.get("2") == null?0:statusMap.get("2");
long stopFlow = statusMap.get("3") == null?0:statusMap.get("3");
long runTotal = runingFlow+stopFlow;
long runingFlow = statusMap.get("2") == null ? 0 : statusMap.get("2");
long stopFlow = statusMap.get("3") == null ? 0 : statusMap.get("3");
long runTotal = runingFlow + stopFlow;
//统计待运行的实例
//统计固定时间内所有flow表的数据 查询所有待运行的工作流
List<Flow> allFlow = flowMapper.findAllByThisDayFlow(statisticsConditionDto);
//统计视力表啊的数据
long notStartFlowCount = statusMap.get("1") == null?0:statusMap.get("1");
long notStartFlowCount = statusMap.get("1") == null ? 0 : statusMap.get("1");
//统计等待表哦数据
//List<WaitingRecord> allByCondition = waitingRecordMapper.findAllByCondition(statisticsConditionDto);
int waitFlowCount = 0;
if(CollectionUtil.isNotEmpty(allFlow)) {
if (CollectionUtil.isNotEmpty(allFlow)) {
waitFlowCount = allFlow.size();
}
/*int allWaitRecount = 0;
......@@ -903,16 +922,16 @@ public class ApiFlowServiceImpl implements ApiFlowService {
CollectData collectData = new CollectData();
//工作流数据灌输
StatisticData statisticData = new StatisticData();
statisticData.setUnstart((int)waitTotal);
statisticData.setUnstart((int) waitTotal);
statisticData.setFail((int) errorTotal);
statisticData.setRunIng((int)runTotal);
statisticData.setSuccess((int)successTotal);
statisticData.setRunIng((int) runTotal);
statisticData.setSuccess((int) successTotal);
collectData.setFlowData(statisticData);
//节点数据灌输
StatisticData nodeStatisticData = statisticsNodeData(betweenRunRecording);
//增加工作流里面的节点 待运行 即扫描标位为未扫描的 就是今天的待运行
allFlow.forEach(flow ->{
nodeStatisticData.setUnstart(nodeStatisticData.getUnstart()+flow.getFlowNodeCount());
allFlow.forEach(flow -> {
nodeStatisticData.setUnstart(nodeStatisticData.getUnstart() + flow.getFlowNodeCount());
});
collectData.setNodeData(nodeStatisticData);
......@@ -922,27 +941,27 @@ public class ApiFlowServiceImpl implements ApiFlowService {
Map<String, StatisticData> nodeCollectMap = new HashMap<>(8);
Map<String, List<FlowStatusSnapshoot>> collect = aallByCon.stream().collect(Collectors.groupingBy(flowStatusSnapshoot -> flowStatusSnapshoot.getDay() + "/" + flowStatusSnapshoot.getHour()));
collect.forEach((key,value) ->{
collect.forEach((key, value) -> {
List<FlowStatusSnapshoot> flowStatusSnapshoots = collect.get(key);
StatisticData figFlowStatisticData = new StatisticData();
Map<Integer, Long> integerLongMap = flowStatusSnapshoots.stream().collect(Collectors.groupingBy(FlowStatusSnapshoot::getFlowStatus, Collectors.counting()));
//未运行的
long waitFigFlowCount = integerLongMap.get(1) == null?0:integerLongMap.get(1);
long waitFigFlowCount = integerLongMap.get(1) == null ? 0 : integerLongMap.get(1);
//运行中的 暂停的
long runFigFlowCount = integerLongMap.get(2) == null?0:integerLongMap.get(2);
long stopFigFlowCount = integerLongMap.get(3) == null?0:integerLongMap.get(3);
long runFigFlowCount = integerLongMap.get(2) == null ? 0 : integerLongMap.get(2);
long stopFigFlowCount = integerLongMap.get(3) == null ? 0 : integerLongMap.get(3);
long figRunTotal = runFigFlowCount + stopFigFlowCount;
//成功的
long successFigFlowCount = integerLongMap.get(4) == null?0:integerLongMap.get(4);
long successFigFlowCount = integerLongMap.get(4) == null ? 0 : integerLongMap.get(4);
//失败的
long errorFigFlowCount = integerLongMap.get(5) == null?0:integerLongMap.get(5);
long killFigFlowCount = integerLongMap.get(6) == null?0:integerLongMap.get(6);
long errorFigFlowCount = integerLongMap.get(5) == null ? 0 : integerLongMap.get(5);
long killFigFlowCount = integerLongMap.get(6) == null ? 0 : integerLongMap.get(6);
long figErrorTotal = errorFigFlowCount + killFigFlowCount;
figFlowStatisticData.setUnstart((int)waitFigFlowCount);
figFlowStatisticData.setFail((int)figErrorTotal);
figFlowStatisticData.setSuccess((int)successFigFlowCount);
figFlowStatisticData.setRunIng((int)figRunTotal);
flowCollectMap.put(key,figFlowStatisticData);
figFlowStatisticData.setUnstart((int) waitFigFlowCount);
figFlowStatisticData.setFail((int) figErrorTotal);
figFlowStatisticData.setSuccess((int) successFigFlowCount);
figFlowStatisticData.setRunIng((int) figRunTotal);
flowCollectMap.put(key, figFlowStatisticData);
long nodeError = 0;
......@@ -952,15 +971,15 @@ public class ApiFlowServiceImpl implements ApiFlowService {
for (FlowStatusSnapshoot flowStatusSnapshoot : flowStatusSnapshoots) {
nodeError += flowStatusSnapshoot.getFailNode() + flowStatusSnapshoot.getKillNode();
nodeRun += flowStatusSnapshoot.getRuningNode();
nodeWaitRun+=flowStatusSnapshoot.getUnstartNode();
nodeSuccess+=flowStatusSnapshoot.getSuccessNode();
nodeWaitRun += flowStatusSnapshoot.getUnstartNode();
nodeSuccess += flowStatusSnapshoot.getSuccessNode();
}
StatisticData nodeFigStatisticData = new StatisticData();
nodeFigStatisticData.setSuccess((int)nodeSuccess);
nodeFigStatisticData.setFail((int)nodeError);
nodeFigStatisticData.setRunIng((int)nodeRun);
nodeFigStatisticData.setUnstart((int)nodeWaitRun);
nodeCollectMap.put(key,nodeFigStatisticData);
nodeFigStatisticData.setSuccess((int) nodeSuccess);
nodeFigStatisticData.setFail((int) nodeError);
nodeFigStatisticData.setRunIng((int) nodeRun);
nodeFigStatisticData.setUnstart((int) nodeWaitRun);
nodeCollectMap.put(key, nodeFigStatisticData);
});
collectData.setFlowCollectMap(flowCollectMap);
collectData.setNodeCollectMap(nodeCollectMap);
......@@ -971,10 +990,11 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* 统计非未运行实例的节点数据
*
* @param betweenRunRecording
* @return
*/
public StatisticData statisticsNodeData (List<RunRecording> betweenRunRecording) {
public StatisticData statisticsNodeData(List<RunRecording> betweenRunRecording) {
//剔除未运行状态的实例
//List<RunRecording> runRecordings = betweenRunRecording.stream().filter(recording -> !recording.getFlowStatus().equals("1")).collect(Collectors.toList());
List<JobTaskRunLogWithBLOBs> logWithBLOBs = new ArrayList<>(32);
......@@ -992,39 +1012,39 @@ public class ApiFlowServiceImpl implements ApiFlowService {
StatisticData statisticData = new StatisticData();
Map<String, Long> stringLongMap = logWithBLOBs.stream().collect(Collectors.groupingBy(JobTaskRunLogWithBLOBs::getRunCode, Collectors.counting()));
//成功的
long successCount = stringLongMap.get("1") == null?0:stringLongMap.get("1");
long reSuccessCount = stringLongMap.get("3") == null?0:stringLongMap.get("3");
long successCount = stringLongMap.get("1") == null ? 0 : stringLongMap.get("1");
long reSuccessCount = stringLongMap.get("3") == null ? 0 : stringLongMap.get("3");
long successTotal = successCount + reSuccessCount;
//执行中的
long runIngCount = stringLongMap.get("0") == null?0:stringLongMap.get("0");
long runIngCount = stringLongMap.get("0") == null ? 0 : stringLongMap.get("0");
//失败的
long errorCount = stringLongMap.get("2") == null?0:stringLongMap.get("2");
long reErorCount = stringLongMap.get("4") == null?0:stringLongMap.get("4");
long killCount = stringLongMap.get("5") == null?0:stringLongMap.get("5");
long upErrorCount = stringLongMap.get("6") == null?0:stringLongMap.get("6");
long errorCount = stringLongMap.get("2") == null ? 0 : stringLongMap.get("2");
long reErorCount = stringLongMap.get("4") == null ? 0 : stringLongMap.get("4");
long killCount = stringLongMap.get("5") == null ? 0 : stringLongMap.get("5");
long upErrorCount = stringLongMap.get("6") == null ? 0 : stringLongMap.get("6");
long errorTotal = errorCount + reErorCount + upErrorCount + killCount;
//待运行的
long waitTotal = total - (successTotal + runIngCount + errorTotal);
statisticData.setSuccess((int)successTotal);
statisticData.setRunIng((int)runIngCount);
statisticData.setFail((int)errorTotal);
statisticData.setUnstart((int)waitTotal);
statisticData.setSuccess((int) successTotal);
statisticData.setRunIng((int) runIngCount);
statisticData.setFail((int) errorTotal);
statisticData.setUnstart((int) waitTotal);
return statisticData;
}
//--------------------------------------------------------------------
private CollectData buildStatisticDate(Long startTime, Long endTime, List<Integer> flowIdList){
private CollectData buildStatisticDate(Long startTime, Long endTime, List<Integer> flowIdList) {
Date startDate = new Date(startTime);
Date endDate = new Date(endTime);
StatisticData flowStatisticData = runRecordingMapper.findStatusByStartAndEndTime(startDate, endDate, flowIdList);
if (null == flowStatisticData){
if (null == flowStatisticData) {
flowStatisticData = new StatisticData();
flowStatisticData.setUnstart(flowIdList.size());
}else {
} else {
Integer sum = runRecordingMapper.findFlowNum(new Date(startTime), new Date(endTime), flowIdList);
if (flowIdList.size() > sum){
if (flowIdList.size() > sum) {
//如果工作流
flowStatisticData.setUnstart(flowIdList.size() - sum + flowStatisticData.getUnstart());
}
......@@ -1032,12 +1052,12 @@ public class ApiFlowServiceImpl implements ApiFlowService {
StatisticData nodeStatisticData = jobTaskRunLogMapper.findStatusByStartAndEndTime(startDate, endDate, flowIdList);
List<Node> nodeList = nodeMapper.findbyFlowIds(flowIdList);
if (null == nodeStatisticData){
if (null == nodeStatisticData) {
nodeStatisticData = new StatisticData();
nodeStatisticData.setUnstart(nodeList.size());
}else {
} else {
int sum = nodeStatisticData.getFail() + nodeStatisticData.getStop() + nodeStatisticData.getRunIng() + nodeStatisticData.getSuccess() + nodeStatisticData.getKill();
nodeStatisticData.setUnstart(nodeList.size() > sum ? nodeList.size()- sum : 0 );
nodeStatisticData.setUnstart(nodeList.size() > sum ? nodeList.size() - sum : 0);
}
List<FlowStatusSnapshoot> flowStatusSnapshootList = flowStatusSnapshootMapper.findByTime(flowIdList, startTime, endTime);
......@@ -1047,7 +1067,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
String key = flowStatusSnapshoot.getDay() + "/" + flowStatusSnapshoot.getHour();
StatisticData nodeHourStatisticData = nodeCollectMap.get(key);
StatisticData flowHourStatisticData = flowCollectMap.get(key);
if (null == nodeHourStatisticData){
if (null == nodeHourStatisticData) {
flowHourStatisticData = new StatisticData();
buildFlowHourStatisticData(flowHourStatisticData, flowStatusSnapshoot);
nodeHourStatisticData = new StatisticData();
......@@ -1056,7 +1076,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
nodeHourStatisticData.setSuccess(flowStatusSnapshoot.getSuccessNode());
nodeHourStatisticData.setFail(flowStatusSnapshoot.getFailNode());
nodeHourStatisticData.setKill(flowStatusSnapshoot.getKillNode());
}else {
} else {
buildFlowHourStatisticData(flowHourStatisticData, flowStatusSnapshoot);
nodeHourStatisticData.setUnstart(nodeHourStatisticData.getUnstart() + flowStatusSnapshoot.getUnstartNode());
nodeHourStatisticData.setRunIng(nodeHourStatisticData.getRunIng() + flowStatusSnapshoot.getRuningNode());
......@@ -1078,23 +1098,37 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* 构建统计数据
*
* @param flowHourStatisticData
* @param flowStatusSnapshoot
*/
private void buildFlowHourStatisticData(StatisticData flowHourStatisticData, FlowStatusSnapshoot flowStatusSnapshoot){
switch (flowStatusSnapshoot.getFlowStatus()){
case 1 : flowHourStatisticData.setUnstart(flowHourStatisticData.getUnstart() + 1); break;
case 2 : flowHourStatisticData.setRunIng(flowHourStatisticData.getRunIng() + 1); break;
case 3 : flowHourStatisticData.setStop(flowHourStatisticData.getStop() + 1); break;
case 4 : flowHourStatisticData.setSuccess(flowHourStatisticData.getSuccess() + 1); break;
case 5 : flowHourStatisticData.setFail(flowHourStatisticData.getFail() + 1); break;
case 6 : flowHourStatisticData.setKill(flowHourStatisticData.getFail() + 1); break;
default: break;
private void buildFlowHourStatisticData(StatisticData flowHourStatisticData, FlowStatusSnapshoot flowStatusSnapshoot) {
switch (flowStatusSnapshoot.getFlowStatus()) {
case 1:
flowHourStatisticData.setUnstart(flowHourStatisticData.getUnstart() + 1);
break;
case 2:
flowHourStatisticData.setRunIng(flowHourStatisticData.getRunIng() + 1);
break;
case 3:
flowHourStatisticData.setStop(flowHourStatisticData.getStop() + 1);
break;
case 4:
flowHourStatisticData.setSuccess(flowHourStatisticData.getSuccess() + 1);
break;
case 5:
flowHourStatisticData.setFail(flowHourStatisticData.getFail() + 1);
break;
case 6:
flowHourStatisticData.setKill(flowHourStatisticData.getFail() + 1);
break;
default:
break;
}
}
@Override
public CollectData loadNodeStatisticData(String param){
public CollectData loadNodeStatisticData(String param) {
ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
JSONObject jsonObject = JSON.parseObject(param);
//获取工作空间名称
......@@ -1110,7 +1144,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
Long endTime = jsonObject.getLong("endTime");
ValidationUtil.dataNotNull(endTime, "结束时间不允许为空！");
//校验时间间隔
ValidationUtil.isTrueValidation(String.valueOf(startTime).length() < 13 || String.valueOf(endTime).length() < 13 , "时间格式必须是毫秒");
ValidationUtil.isTrueValidation(String.valueOf(startTime).length() < 13 || String.valueOf(endTime).length() < 13, "时间格式必须是毫秒");
//获取工作流名称
String flowName = jsonObject.getString("flowName");
ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空！");
......@@ -1131,7 +1165,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
}
return buildStatisticDate(startTime, endTime, flowIdList);*/
return buildStatisticAllDate(startTime, endTime, workspace.getWorkspaceId(),flowName);
return buildStatisticAllDate(startTime, endTime, workspace.getWorkspaceId(), flowName);
}
@Override
......@@ -1148,7 +1182,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
ValidationUtil.dataNotBank(flowNames, "工作流名称不允许为空！");
List<String> flowNameList = Arrays.asList(flowNames.split(","));
List<Flow> flowList = new ArrayList<>();
flowNameList.forEach(flowName->{
flowNameList.forEach(flowName -> {
Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
flowList.add(flow);
......@@ -1177,7 +1211,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
String flowName = jsonObject.getString("flowName");
ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空！");
Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
if (null == flow){
if (null == flow) {
return false;
}
return true;
......@@ -1186,11 +1220,11 @@ public class ApiFlowServiceImpl implements ApiFlowService {
@Override
public Boolean stopScheduleByRunId(String runId) {
List<RunRecording> recordingList = runRecordingMapper.findByRunID(runId);
ValidationUtil.dataNotNull(recordingList , "查无此运行实例！");
ValidationUtil.dataNotNull(recordingList, "查无此运行实例！");
AtomicReference<Boolean> result = new AtomicReference<>(true);
recordingList.forEach(runRecording -> {
if (("3".equals(runRecording.getFlowStatus()) || "4".equals(runRecording.getFlowStatus())) //如果不是正在运行
&& FlowPropertyEnum.ISNOT_INNER.getCode().equals(runRecording.getIsInner())){ //并且不是内嵌工作流
&& FlowPropertyEnum.ISNOT_INNER.getCode().equals(runRecording.getIsInner())) { //并且不是内嵌工作流
result.set(false);
}
});
......@@ -1205,6 +1239,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* runState 补批机制 1 补批当前节点 2 补批当前节点及以下节点
*
* @param param
*/
@Override
......@@ -1234,7 +1269,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
Node node = nodeMapper.getByNameAndFlow(nodeName, flow.getFlowId());
ValidationUtil.dataNotNull(node, nodeName + "节点不存在");
if (StringUtils.isNotEmpty(node.getRunParam())){
if (StringUtils.isNotEmpty(node.getRunParam())) {
ValidationUtil.dataNotBank(runParam, "运行参数不允许为空！");
}
}
......@@ -1267,7 +1302,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
List<Node> nodeList = nodeMapper.findByFlowIdAndName(flow.getFlowId(), nodeNameList);
try{
try {
RedissLockUtil.trlock(flow.getFlowId().toString(), 5);
//设置触发时间
Long triggerTime = System.currentTimeMillis();
......@@ -1277,11 +1312,11 @@ public class ApiFlowServiceImpl implements ApiFlowService {
Collections.sort(repairTimeList);
//查看当前排队的工作流最大排队序号
Integer order = waitingRecordMapper.findOrderByFlowId(flow.getFlowId());
if (order == null){
if (order == null) {
order = 0;
}
String userName = currentUserUtils.account();
for (String repairTime : repairTimeList){
for (String repairTime : repairTimeList) {
WaitingRecord waitingRecord = new WaitingRecord();
RunRecording runRecording = new RunRecording();
String runId = IDGenerationStrategy.runIdGenerationStrategy(serverPort);
......@@ -1311,15 +1346,15 @@ public class ApiFlowServiceImpl implements ApiFlowService {
waitingTaskMapper.insertSelective(waitingTask);
});
}
}catch (Exception e) {
} catch (Exception e) {
e.printStackTrace();
}finally {
} finally {
RedissLockUtil.unlock(flow.getFlowId().toString());
}
}
private Map<String, List<WaitingTask>> buildTask(List<Node> nodeList, List<String> nodeNameList, List<String> repairTimeList, Long triggerTime, String flowName){
private Map<String, List<WaitingTask>> buildTask(List<Node> nodeList, List<String> nodeNameList, List<String> repairTimeList, Long triggerTime, String flowName) {
Map<String, List<WaitingTask>> result = new HashMap<>();
String usernName = currentUserUtils.account();
List<WaitingTask> waitingTaskList = new ArrayList<>();
......@@ -1329,10 +1364,10 @@ public class ApiFlowServiceImpl implements ApiFlowService {
WaitingTask waitingTask = new WaitingTask();
BeanUtils.copyProperties(node, waitingTask);
List<String> dependList = nodeDependencyMapper.findNodeInfoByNodeId(node.getNodeId());
if (dependList != null && dependList.size() > 0){
if (dependList != null && dependList.size() > 0) {
List<Integer> dependNodeIdList = new ArrayList<>();
dependList.forEach(nodeName -> {
if (nodeNameList.contains(nodeName)){
if (nodeNameList.contains(nodeName)) {
dependNodeIdList.add(idNameMap.get(nodeName));
}
});
......@@ -1347,7 +1382,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
});
//校验时间
for (String repairTime : repairTimeList){
for (String repairTime : repairTimeList) {
List<WaitingTask> waitingTaskDateList = new ArrayList<>();
try {
Date repairDate = sdf.parse(repairTime);
......@@ -1359,11 +1394,16 @@ public class ApiFlowServiceImpl implements ApiFlowService {
//TODO 替换参数
waitingTaskList.forEach(waitingTask -> {
//补批只替换不是java的节点
if(!NodeTypeEnum.JAVA.getCode().equals( waitingTask.getJobType())){
if (StringUtils.isNotEmpty(waitingTask.getRunParam())){
String param = PlaceholderUtils.paramPlaceholder(waitingTask.getRunParam(), repairTime);
if (!NodeTypeEnum.JAVA.getCode().equals(waitingTask.getJobType())) {
if (StringUtils.isNotEmpty(waitingTask.getRunParam())) {
String runParamSource = waitingTask.getRunParam();
RunParamWrapped runParamWrapped = JSON.parseObject(runParamSource, RunParamWrapped.class);
//替换私有参数
String param = PlaceholderUtils.paramPlaceholder(runParamWrapped.getPrivateParam(), repairTime);
runParamWrapped.setPrivateParam(param);
param = JSON.toJSONString(runParamWrapped);
String runCommand = waitingTask.getRunCommand();
runCommand.replace(PRE +PlaceholderEnum.DATE_PLACEHOLDER.getName()+ SUFFER,repairTime);
runCommand.replace(PRE + PlaceholderEnum.DATE_PLACEHOLDER.getName() + SUFFER, repairTime);
waitingTask.setRunCommand(runCommand);
waitingTask.setRunParam(param);
}
......@@ -1380,6 +1420,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* 工作流生成新版本
*
* @param flow
* @return
*/
......@@ -1391,10 +1432,10 @@ public class ApiFlowServiceImpl implements ApiFlowService {
FlowVersion flowVersion = new FlowVersion();
BeanUtils.copyProperties(flow, flowVersion);
if (!"V.1".equals(flow.getVersionName())){
if (!"V.1".equals(flow.getVersionName())) {
//查找当前版本的工作流，如果有，则设置为不是，若没有跳过
FlowVersion oldFlowVersion = flowVersionMapper.getByFlowIdAndThis(flow.getFlowId());
if (null != oldFlowVersion){
if (null != oldFlowVersion) {
oldFlowVersion.setVersionMark(FlowPropertyEnum.ISNOT_CURRENTVERSION.getCode());
flowVersionMapper.updateByIdSelective(oldFlowVersion);
}
......@@ -1411,7 +1452,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
nodeList.forEach(node -> {
//查询本节点的依赖关系，若不为空则添加到总的依赖集合中
List<NodeDependencyKey> nodeDependList = nodeDependencyMapper.findByNodeId(node.getNodeId());
if (null != nodeDependList && nodeDependList.size() > 0){
if (null != nodeDependList && nodeDependList.size() > 0) {
dependencyKeyList.addAll(nodeDependList);
}
NodeVersion nodeVersion = new NodeVersion();
......@@ -1419,7 +1460,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
nodeVersion.setAddTime(currentDate);
nodeVersion.setFlowVersionId(flowVersion.getFlowVersionId());
nodeVersion.setVersionMark(FlowPropertyEnum.IS_CURRENTVERSION.getCode());
if (null != node.getIsVirtual() && NodePropertyEnum.IS_VIRTUAL.getCode().equals(node.getIsVirtual())){
if (null != node.getIsVirtual() && NodePropertyEnum.IS_VIRTUAL.getCode().equals(node.getIsVirtual())) {
Flow innerFlow = flowMapper.getById(node.getMapFlowId());
FlowVersion innerFlowVersion = saveFlowVersion(innerFlow);
nodeVersion.setMapFlowId(innerFlowVersion.getFlowVersionId());
......@@ -1432,8 +1473,8 @@ public class ApiFlowServiceImpl implements ApiFlowService {
dependencyKeyList.forEach(nodeDependencyKey -> {
Integer nodeVersionId = idVersionIdRel.get(nodeDependencyKey.getNodeId());
Integer dependVersionId = idVersionIdRel.get(nodeDependencyKey.getDependencyId());
ValidationUtil.dataNotNull(nodeVersionId , "没有找到对应的节点");
ValidationUtil.dataNotNull(dependVersionId , "没有找到对应的依赖的节点");
ValidationUtil.dataNotNull(nodeVersionId, "没有找到对应的节点");
ValidationUtil.dataNotNull(dependVersionId, "没有找到对应的依赖的节点");
nodeVersionDependencyMapper.insert(nodeVersionId, dependVersionId);
});
......@@ -1445,7 +1486,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
public void deleteFlow(String param) {
DeleteDto deleteDto = new DeleteDto();
parseParam(param,deleteDto);
parseParam(param, deleteDto);
String workspaceName = deleteDto.getWorkspaceName();
String flowName = deleteDto.getFlowName();
log.debug("删除工作空间【{}】---工作流【{}】调度", workspaceName, flowName);
......@@ -1456,11 +1497,11 @@ public class ApiFlowServiceImpl implements ApiFlowService {
ValidationUtil.isTrueValidation(FlowPropertyEnum.IS_INNER.getCode().equals(flow.getIsInner()), "内嵌工作流不允许删除！");
//判断是否在调度中
List<RunRecording> recordingList = runRecordingMapper.findOnScheduleByFlowId(flow.getFlowId());
ValidationUtil.isTrueValidation(null != recordingList && recordingList.size() > 0 , "工作流已在调度中不允许撤销调度！");
ValidationUtil.isTrueValidation(null != recordingList && recordingList.size() > 0, "工作流已在调度中不允许撤销调度！");
List<RunRecording> unStartRecordingList = runRecordingMapper.findUnStartByFlowId(flow.getFlowId());
//删除对应的task记录
unStartRecordingList.forEach(runRecording -> {
ValidationUtil.isTrueValidation((runRecording.getTriggerTime() - System.currentTimeMillis()) < 10000 , "工作流已在调度中不允许删除");
ValidationUtil.isTrueValidation((runRecording.getTriggerTime() - System.currentTimeMillis()) < 10000, "工作流已在调度中不允许删除");
runRecordingMapper.deleteByRunId(runRecording.getRunId());
jobTaskMapper.deleteByRunId(runRecording.getRunId());
});
......@@ -1470,15 +1511,16 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* 删除工作流及下属节点
*
* @param flow
*/
private void deleteFlowAndNode(Flow flow) {
List<Node> nodeList = nodeMapper.findNodeByFlowId(flow.getFlowId());
if (null != nodeList && !nodeList.isEmpty()){
if (null != nodeList && !nodeList.isEmpty()) {
//删除节点的依赖
nodeList.forEach(node -> {
nodeDependencyMapper.deleteByNodeId(node.getNodeId());
if (StringUtils.isNotEmpty(node.getIsVirtual()) && NodePropertyEnum.IS_VIRTUAL.getCode().equals(node.getIsVirtual())){
if (StringUtils.isNotEmpty(node.getIsVirtual()) && NodePropertyEnum.IS_VIRTUAL.getCode().equals(node.getIsVirtual())) {
Flow innerFlow = new Flow();
innerFlow.setFlowId(node.getMapFlowId());
deleteFlowAndNode(innerFlow);
......@@ -1498,7 +1540,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
@Transactional(rollbackFor = Exception.class)
public void start(String param) throws ParseException {
DeleteDto deleteDto = new DeleteDto();
parseParam(param,deleteDto);
parseParam(param, deleteDto);
String workspaceName = deleteDto.getWorkspaceName();
String flowName = deleteDto.getFlowName();
log.debug("开始工作空间【{}】---工作流【{}】调度", workspaceName, flowName);
......@@ -1516,7 +1558,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
@Transactional(rollbackFor = Exception.class)
public void repealSchedule(String param) throws ParseException {
DeleteDto deleteDto = new DeleteDto();
parseParam(param,deleteDto);
parseParam(param, deleteDto);
String workspaceName = deleteDto.getWorkspaceName();
String flowName = deleteDto.getFlowName();
log.debug("撤销工作空间【{}】---工作流【{}】调度", workspaceName, flowName);
......@@ -1527,12 +1569,12 @@ public class ApiFlowServiceImpl implements ApiFlowService {
ValidationUtil.isTrueValidation(FlowPropertyEnum.IS_INNER.getCode().equals(flow.getIsInner()), "内嵌工作流不允许撤销调度！");
//判断是否在调度中
List<RunRecording> recordingList = runRecordingMapper.findOnScheduleByFlowId(flow.getFlowId());
ValidationUtil.isTrueValidation(null != recordingList && recordingList.size() > 0 , "工作流已在调度中不允许撤销调度！");
ValidationUtil.isTrueValidation(null != recordingList && recordingList.size() > 0, "工作流已在调度中不允许撤销调度！");
List<RunRecording> unStartRecordingList = runRecordingMapper.findUnStartByFlowId(flow.getFlowId());
if (null != unStartRecordingList && unStartRecordingList.size() > 0){
if (null != unStartRecordingList && unStartRecordingList.size() > 0) {
//删除对应的task记录
unStartRecordingList.forEach(runRecording -> {
ValidationUtil.isTrueValidation((runRecording.getTriggerTime() - System.currentTimeMillis()) < 10000 , "工作流已在调度中不允许撤销调度");
ValidationUtil.isTrueValidation((runRecording.getTriggerTime() - System.currentTimeMillis()) < 10000, "工作流已在调度中不允许撤销调度");
runRecordingMapper.deleteByRunId(runRecording.getRunId());
jobTaskMapper.deleteByRunId(runRecording.getRunId());
});
......@@ -1546,23 +1588,24 @@ public class ApiFlowServiceImpl implements ApiFlowService {
/**
* 更改工作流的启动状态
* @param flow 工作流
* @param isStart 是否启动
*
* @param flow 工作流
* @param isStart 是否启动
*/
private void updateFlowStart(Flow flow, boolean isStart) throws ParseException {
if (isStart){
if (isStart) {
flow.setStartUp(FlowPropertyEnum.IS_START.getCode());
//如果是周期调度修改下次执行时间
if (FlowPropertyEnum.SCHEDULE_MODE.getCode().equals(flow.getExecType())){
if (FlowPropertyEnum.SCHEDULE_MODE.getCode().equals(flow.getExecType())) {
flow.setTriggerNextTime(new CronExpression(flow.getFlowCron()).getNextValidTimeAfter(new Date()).getTime());
}
}else {
} else {
flow.setStartUp(FlowPropertyEnum.NO_START.getCode());
}
flowMapper.updateByIdSelective(flow);
List<Node> nodeList = nodeMapper.findVirtualByFlowId(flow.getFlowId());
if (null != nodeList && nodeList.size() > 0 ){
for (Node node : nodeList){
if (null != nodeList && nodeList.size() > 0) {
for (Node node : nodeList) {
Flow innerFlow = flowMapper.getById(node.getMapFlowId());
updateFlowStart(innerFlow, isStart);
}
......@@ -1578,7 +1621,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
@Transactional(rollbackFor = Exception.class)
public String stopSchedule(String param) {
DeleteDto deleteDto = new DeleteDto();
parseParam(param,deleteDto);
parseParam(param, deleteDto);
String workspaceName = deleteDto.getWorkspaceName();
String flowName = deleteDto.getFlowName();
log.debug("停止工作空间【{}】---工作流【{}】调度", workspaceName, flowName);
......@@ -1589,7 +1632,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
ValidationUtil.isTrueValidation(FlowPropertyEnum.IS_INNER.getCode().equals(flow.getIsInner()), "内嵌工作流不允许停止调度！");
//判断是否在调度中
List<RunRecording> recordingList = runRecordingMapper.findUnFinishByFlowId(flow.getFlowId());
ValidationUtil.isTrueValidation(null == recordingList || recordingList.size() == 0 , flowName + "工作流没有正在运行的调度！");
ValidationUtil.isTrueValidation(null == recordingList || recordingList.size() == 0, flowName + "工作流没有正在运行的调度！");
StringBuffer runids = new StringBuffer();
//暂停工作流调度
recordingList.forEach(runRecording -> {
......@@ -1607,8 +1650,8 @@ public class ApiFlowServiceImpl implements ApiFlowService {
List<String> runIdList = Arrays.asList(runIds.split(","));
runIdList.forEach(runId -> {
List<RunRecording> runRecordList = runRecordingMapper.findStopRunCordByRunId(runId);
log.warn("---------运行记录{}，不存在暂停记录中！---",runId);
List<RunRecording> runRecordList = runRecordingMapper.findStopRunCordByRunId(runId);
log.warn("---------运行记录{}，不存在暂停记录中！---", runId);
ValidationUtil.isTrueValidation(null == runRecordList || runRecordList.size() == 0, "该工作流不是暂停状态！");
runRecordingMapper.startByRunId(runId);
jobTaskMapper.startByRunId(runId);
......
byit-myth-admin/src/main/resources/config/application-local.yml
spring:
datasource:
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://10.0.10.118:3306/myth-job?Unicode=true&characterEncoding=UTF-8&useSSL=true
url: jdbc:mysql://10.0.120.30:3307/myth-job?Unicode=true&characterEncoding=UTF-8&useSSL=true
username: root
password: 123456
password: root
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/dto/NodeRelyDto.java 0 → 100644
package com.byit.dto;
import com.byit.model.Node;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
/**
* 节点依赖数据载体
*
* @author huangfu
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
@EqualsAndHashCode
public class NodeRelyDto {
private String relyId;
private Node node;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/NodeDependencyMapper.java
package com.byit.mapper;
import com.byit.model.Node;
import com.byit.model.NodeDependencyKey;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;
@Repository
public interface NodeDependencyMapper {
int deleteById(NodeDependencyKey key);
......@@ -15,6 +16,7 @@ public interface NodeDependencyMapper {
/**
* 根据节点id删除依赖关系
*
* @param nodeId
* @return
*/
......@@ -22,6 +24,7 @@ public interface NodeDependencyMapper {
/**
* 根据节点id查询本节点依赖的节点
*
* @param nodeId
* @return
*/
......@@ -31,10 +34,18 @@ public interface NodeDependencyMapper {
/**
* 查询依赖于当前nodeid的节点id
*
* @param nodeId
* @return
*/
List<Integer> findSubNodeList(Integer nodeId);
List<String> findNodeInfoByNodeId(Integer nodeId);
/**
* 查询所有节点的依赖节点
* @param nodeIds 所有的节点id
* @return 所有节点的依赖关系
*/
List<NodeDependencyKey> findAllNodeDependencyKey(@Param("nodeIds") List<Integer> nodeIds);
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/Flow.java
......@@ -46,7 +46,7 @@ public class Flow implements Serializable {
/**
* 当前工作流的介绍
*/
@ApiModelProperty("当前工作流的介绍")
@ApiModelProperty("当前工作流的公共参数")
private String flowDesc;
/**
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/FlowVersion.java
......@@ -46,7 +46,7 @@ public class FlowVersion implements Serializable {
/**
* 工作流的介绍
*/
@ApiModelProperty("工作流的介绍")
@ApiModelProperty("工作流的公共参数")
private String flowDesc;
/**
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/FlowService.java
......@@ -18,6 +18,14 @@ import java.util.List;
* @create: 2019-12-23 17:33
*/
public interface FlowService {
/**
* 查询工作流名称 基于工作流名称查询
*
* @param flowName 工作流名称
* @param workspaceId 工作空间名称ID
* @return 返回对应的
*/
Flow findFlowByName(String flowName, Integer workspaceId);
/**
* 查询当天将要运行的工作流
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/NodeDependencyService.java
package com.byit.service;
import com.byit.model.NodeDependencyKey;
import java.util.List;
/**
......@@ -14,4 +16,11 @@ public interface NodeDependencyService {
*/
List<Integer> findDependIdByNodeId(Integer nodeId);
/**
* 查询所有节点的依赖节点
* @param nodeIds 所有的节点id
* @return 所有节点的依赖关系
*/
List<NodeDependencyKey> findAllNodeDependencyKey(List<Integer> nodeIds);
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/FlowServiceImpl.java
......@@ -57,8 +57,14 @@ public class FlowServiceImpl implements FlowService {
private JobTaskService jobTaskService;
@Override
public Flow findFlowByName(String flowName, Integer workspaceId) {
return flowMapper.getByWorkSpaceAndName(workspaceId, flowName);
}
/**
* 查询当天所有待运行的任务流
*
* @param triggerNextTime 下次运行中
* @return
*/
......@@ -73,7 +79,7 @@ public class FlowServiceImpl implements FlowService {
@Override
public List<FlowViewVo> findAllFlowViewVo(FlowConditionDto flowConditionDto) {
List<Flow> allFlow = findAllFlow(flowConditionDto);
return allFlow.stream().map(flow -> {
return allFlow.stream().map(flow -> {
FlowViewVo flowViewVo = new FlowViewVo();
flowViewVo.setFlowId(flow.getFlowId());
flowViewVo.setFlowName(flow.getFlowName());
......@@ -89,7 +95,7 @@ public class FlowServiceImpl implements FlowService {
List<Flow> all = flowMapper.findAllByCondition(flowConditionDto);
return all.stream().map(flow -> {
FlowImportantAllVo flowImportantAllVo = new FlowImportantAllVo();
flowImportantAllVo.setRowKey(UUID.randomUUID().toString().replace("-",""));
flowImportantAllVo.setRowKey(UUID.randomUUID().toString().replace("-", ""));
flowImportantAllVo.setHasChildren(true);
flowImportantAllVo.setFlowName(flow.getFlowName());
flowImportantAllVo.setFlowId(flow.getFlowId());
......@@ -144,6 +150,7 @@ public class FlowServiceImpl implements FlowService {
/**
* 重新组织节点的依赖和调度关系
*
* @param flowVo
* @param nodeVoList
*/
......@@ -160,7 +167,7 @@ public class FlowServiceImpl implements FlowService {
//查找并删除工作流下的虚节点
List<Node> virtualNodeList = nodeMapper.findVirtualByFlowId(flowVo.getFlowId());
if (null != virtualNodeList && virtualNodeList.size() > 0){
if (null != virtualNodeList && virtualNodeList.size() > 0) {
for (Node node : virtualNodeList) {
Flow virtualFlow = new Flow();
virtualFlow.setFlowId(node.getMapFlowId());
......@@ -177,16 +184,16 @@ public class FlowServiceImpl implements FlowService {
log.info("给工作流【{}】添加节点调度调度", flowVo.getFlowName());
for (NodeVo nodeVo : nodeVoList) {
//判断是否是虚节点
if (StringUtils.isNotEmpty(nodeVo.getIsVirtual()) && NodePropertyEnum.IS_VIRTUAL.getCode().equals(nodeVo.getIsVirtual())){
if (StringUtils.isNotEmpty(nodeVo.getIsVirtual()) && NodePropertyEnum.IS_VIRTUAL.getCode().equals(nodeVo.getIsVirtual())) {
includeFlow(flowVo, nodeVo);
}else {
} else {
nodeMapper.updateOnforkUp(nodeVo.getNodeId());
}
//重新组建节点的依赖关系
if (null != nodeVo.getDependNodeId() && nodeVo.getDependNodeId().size() > 0){
if (null != nodeVo.getDependNodeId() && nodeVo.getDependNodeId().size() > 0) {
List<Integer> dependNodeList = nodeVo.getDependNodeId();
dependNodeList.forEach(dependNodeId-> nodeDependencyMapper.insert(nodeVo.getNodeId(), dependNodeId));
dependNodeList.forEach(dependNodeId -> nodeDependencyMapper.insert(nodeVo.getNodeId(), dependNodeId));
}
}
......@@ -244,14 +251,14 @@ public class FlowServiceImpl implements FlowService {
flow.setVersionName("V.1");
//新建的工作流设置为不启动
flow.setStartUp(FlowPropertyEnum.NO_START.getCode());
if(flow.getRemainingCount() != null && flow.getRemainingCount() > 0){
if (flow.getRemainingCount() != null && flow.getRemainingCount() > 0) {
flow.setRepeatCount(flow.getRemainingCount());
}
//获取工作流id
int flowId = flowMapper.insertSelective(flow);
//初始创建两个节点（开始和结束节点）
List<Node> nodeList = FlowInitUtil.InitNode(flow);
nodeList.forEach(node-> nodeMapper.insertSelective(node));
nodeList.forEach(node -> nodeMapper.insertSelective(node));
//将工作流的节点数改为2
flow.setFlowNodeCount(2);
......@@ -284,7 +291,7 @@ public class FlowServiceImpl implements FlowService {
BeanUtils.copyProperties(flow, flowVo);
//将工作流下在调度上的节点添加进来
List<Node> nodeList = nodeMapper.findOnforkByFlowId(flowId);
if (null != nodeList && nodeList.size() > 0){
if (null != nodeList && nodeList.size() > 0) {
List<NodeVo> nodeVoList = new ArrayList<>();
nodeList.forEach(node -> {
NodeVo nodeVo = new NodeVo();
......@@ -304,11 +311,11 @@ public class FlowServiceImpl implements FlowService {
BeanUtils.copyProperties(flowVo, flow);
//校验cron表达式
boolean isValid = CronExpression.isValidExpression(flow.getFlowCron());
if (!isValid){
if (!isValid) {
log.error("工作流【{}】cron表达式不符合规范", flow.getFlowName());
return false;
}
if(flow.getRemainingCount() != null && flow.getRemainingCount() > 0){
if (flow.getRemainingCount() != null && flow.getRemainingCount() > 0) {
log.info("设置工作流的剩余次数");
flow.setRepeatCount(flow.getRemainingCount());
}
......@@ -317,8 +324,9 @@ public class FlowServiceImpl implements FlowService {
/**
* 判断是否允许依赖这个工作流
* @param flowVo 当前工作流
* @param dependFlowId 要依赖的工作流
*
* @param flowVo 当前工作流
* @param dependFlowId 要依赖的工作流
* @return
*/
@Override
......@@ -372,41 +380,41 @@ public class FlowServiceImpl implements FlowService {
JobTaskRunLogWithBLOBs jobTaskRunLog = jobTaskRunLogMapper.findByRunIdAndFlowAndNode(runId, flowName, nodeName);
ValidationUtil.dataNotNull(jobTaskRunLog, "不存在【" + flowName + "】下节点【" + nodeName + "】的运行日志,或尚未开始调度!");
ValidationUtil.isTrueValidation(!"0".equals(jobTaskRunLog.getRunCode()), "该任务已经运行结束！");
if (FlowPropertyEnum.IS_INNER.getCode().equals(jobTaskRunLog.getIsVirtual())){
if (FlowPropertyEnum.IS_INNER.getCode().equals(jobTaskRunLog.getIsVirtual())) {
runRecordingMapper.killInnerFlow(runId, flowName);
List<JobTaskRunLogWithBLOBs> jobTaskRunLogList = jobTaskRunLogMapper.findByRunIdAndFlowName(runId, flowName);
ValidationUtil.dataNotNull(jobTaskRunLogList, "该任务尚未开始调度");
for (JobTaskRunLogWithBLOBs innerJobTaskRunLog : jobTaskRunLogList){
if (!"0".equals(innerJobTaskRunLog.getRunCode())){
if (StringUtils.isEmpty(innerJobTaskRunLog.getJobGroupIp())){
for (JobTaskRunLogWithBLOBs innerJobTaskRunLog : jobTaskRunLogList) {
if (!"0".equals(innerJobTaskRunLog.getRunCode())) {
if (StringUtils.isEmpty(innerJobTaskRunLog.getJobGroupIp())) {
log.warn("已经调度成功但是还未返回具体的调用机器的ip地址");
Thread.sleep(3000);
innerJobTaskRunLog = jobTaskRunLogMapper.findByRunIdAndFlowAndNode(runId, flowName, nodeName);
if(!"0".equals(jobTaskRunLog.getRunCode())){
if (!"0".equals(jobTaskRunLog.getRunCode())) {
log.info("该任务已经运行结束,无需停止！");
continue;
}
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())) {
errorMsg.append("【").append(jobTaskRunLog.getNodeName()).append("】节点尚未分配执行机请稍后再试\n");
result = false;
continue;
}
}
if (!killJob(innerJobTaskRunLog.getLogId(), innerJobTaskRunLog.getJobGroupIp(), innerJobTaskRunLog.getNodeName(), errorMsg)){
if (!killJob(innerJobTaskRunLog.getLogId(), innerJobTaskRunLog.getJobGroupIp(), innerJobTaskRunLog.getNodeName(), errorMsg)) {
result = false;
}
}
}
}else {
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
} else {
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())) {
log.warn("已经调度成功但是还未返回具体的调用机器的ip地址");
Thread.sleep(3000);
jobTaskRunLog = jobTaskRunLogMapper.findByRunIdAndFlowAndNode(runId, flowName, nodeName);
if(!"0".equals(jobTaskRunLog.getRunCode())){
if (!"0".equals(jobTaskRunLog.getRunCode())) {
log.info("该任务已经运行结束,无需停止！");
return new KillDto(result, errorMsg.toString());
}
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())) {
errorMsg.append("【").append(jobTaskRunLog.getNodeName()).append("】节点尚未分配执行机请稍后再试\n");
return new KillDto(false, errorMsg.toString());
}
......@@ -436,9 +444,9 @@ public class FlowServiceImpl implements FlowService {
Queue<Integer> queue = new LinkedList<>();
flowIdList.add(flow.getFlowId());
queue.offer(flow.getFlowId());
while(!queue.isEmpty()){
while (!queue.isEmpty()) {
List<Node> innerFlowList = nodeMapper.findVirtualByFlowId(queue.poll());
if (null != innerFlowList && innerFlowList.size() > 0){
if (null != innerFlowList && innerFlowList.size() > 0) {
innerFlowList.forEach(node -> {
flowIdList.add(node.getMapFlowId());
queue.offer(node.getMapFlowId());
......@@ -460,29 +468,29 @@ public class FlowServiceImpl implements FlowService {
//杀死所有的任务
StringBuffer errorMsg = new StringBuffer();
List<JobTaskRunLog> jobTaskRunLogList = jobTaskRunLogMapper.findbyRunIdAndFlowIdList(runId, flowIdList);
if (null != jobTaskRunLogList && jobTaskRunLogList.size() > 0){
for (JobTaskRunLog jobTaskRunLog : jobTaskRunLogList){
if ("0".equals(jobTaskRunLog.getRunCode()) && ! NodePropertyEnum.IS_VIRTUAL.getCode().equals(jobTaskRunLog.getIsVirtual())){
if (NodeTypeEnum.JAVA.getCode().equals(jobTaskRunLog.getJobType())){
if (null != jobTaskRunLogList && jobTaskRunLogList.size() > 0) {
for (JobTaskRunLog jobTaskRunLog : jobTaskRunLogList) {
if ("0".equals(jobTaskRunLog.getRunCode()) && !NodePropertyEnum.IS_VIRTUAL.getCode().equals(jobTaskRunLog.getIsVirtual())) {
if (NodeTypeEnum.JAVA.getCode().equals(jobTaskRunLog.getJobType())) {
errorMsg.append("【").append(jobTaskRunLog.getNodeName()).append("】节点类型为JAVA，无法杀死！\n");
result = false;
continue;
}
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())) {
log.warn("已经调度成功但是还未返回具体的调用机器的ip地址");
Thread.sleep(3000);
jobTaskRunLog = jobTaskRunLogMapper.findJobTaskRunLogByLogId(jobTaskRunLog.getLogId());
if(!"0".equals(jobTaskRunLog.getRunCode())){
if (!"0".equals(jobTaskRunLog.getRunCode())) {
log.info("该任务已经运行结束,无需停止！");
continue;
}
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())) {
errorMsg.append("【").append(jobTaskRunLog.getNodeName()).append("】节点尚未分配执行机请稍后再试\n");
result = false;
continue;
}
}
if (! killJob(jobTaskRunLog.getLogId(), jobTaskRunLog.getJobGroupIp(), jobTaskRunLog.getNodeName(), errorMsg)){
if (!killJob(jobTaskRunLog.getLogId(), jobTaskRunLog.getJobGroupIp(), jobTaskRunLog.getNodeName(), errorMsg)) {
result = false;
}
}
......@@ -493,7 +501,7 @@ public class FlowServiceImpl implements FlowService {
String runIdS = runRecording.getRunId();
List<JobTask> byRunId = jobTaskService.findByRunId(runIdS);
List<JobTask> stopTask = byRunId.stream().filter(task -> JobTriggerStatusEnums.STOP.getCode().equals(task.getTriggerStatus())).collect(Collectors.toList());
stopTask.forEach(task ->{
stopTask.forEach(task -> {
task.setTriggerStatus(JobTriggerStatusEnums.START.getCode());
jobTaskService.update(task);
});
......@@ -504,7 +512,7 @@ public class FlowServiceImpl implements FlowService {
return new KillDto(result, errorMsg.toString());
}
private synchronized Boolean killJob(Integer logId, String exectUrl, String nodeName, StringBuffer errorMsg){
private synchronized Boolean killJob(Integer logId, String exectUrl, String nodeName, StringBuffer errorMsg) {
//具体访问的URL
String killUrl = "http://" + exectUrl + "/myth-executor-server/processManager/killJob";
Map<String, Object> requestMap = new HashMap<>(2);
......@@ -512,7 +520,7 @@ public class FlowServiceImpl implements FlowService {
String killResult = HttpUtil.post(killUrl, requestMap);
log.info("请求结果{}", killResult);
ResponseResult responseResult = JSON.parseObject(killResult, ResponseResult.class);
if (!"SUCCESS".equals(responseResult.getResult())){
if (!"SUCCESS".equals(responseResult.getResult())) {
errorMsg.append("【").append(nodeName).append("】杀死失败，错误信息为:").append(responseResult.getMsg());
return false;
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/NodeDependencyServiceImpl.java
package com.byit.service.impl;
import com.byit.mapper.NodeDependencyMapper;
import com.byit.model.NodeDependencyKey;
import com.byit.service.NodeDependencyService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
......@@ -25,4 +26,9 @@ public class NodeDependencyServiceImpl implements NodeDependencyService {
public List<Integer> findDependIdByNodeId(Integer nodeId) {
return nodeDependencyMapper.findDependIdByNodeId(nodeId);
}
@Override
public List<NodeDependencyKey> findAllNodeDependencyKey(List<Integer> nodeIds) {
return nodeDependencyMapper.findAllNodeDependencyKey(nodeIds);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/ScheduleThreadRunHelper.java
package com.byit.thread.helper;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.executor.ScriptParamAndPlaceholderDto;
import com.byit.dto.plugin.JavaTask;
import com.byit.enums.NodeTypeEnum;
import com.byit.enums.PlaceholderEnum;
......@@ -20,12 +23,14 @@ import com.byit.thread.BaseThreadRunHelper;
import com.byit.util.SpringUtil;
import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.util.Date;
import java.util.List;
import java.util.Map;
/**
* 排期表操作
......@@ -57,16 +62,44 @@ public class ScheduleThreadRunHelper extends BaseThreadRunHelper {
log.debug("------排期表查询到有需要存在的节点--------");
//循环遍历添加任务
jobTaskSchedules.forEach(mythJobTaskSchedule ->{
//除去JAVA任务
if(!NodeTypeEnum.JAVA.getCode().equals(mythJobTaskSchedule.getJobType())){
//获取运行时参数
String runParam = mythJobTaskSchedule.getRunParam();
//获取命令
String command = mythJobTaskSchedule.getRunCommand();
//转换参数对象为参数包装体
RunParamWrapped runParamWrapped = JSON.parseObject(runParam, RunParamWrapped.class);
//当参数包装体不为空时 证明存在参数 或私有或公有
if(runParamWrapped != null) {
//获取到私有参数
String privateParam = runParamWrapped.getPrivateParam();
//获取到公有参数
Map<String,String> publicParam = runParamWrapped.getPublicParamMap();
//将私有参数转换为对应的参数DTO
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = JSON.parseObject(privateParam,ScriptParamAndPlaceholderDto.class);
//脚本参数不为空的时候
if(scriptParamAndPlaceholderDto != null) {
Map<String, String> param = scriptParamAndPlaceholderDto.getParam();
if(CollectionUtil.isNotEmpty(param)){
command = PlaceholderUtils.commandReplace(command,param );
}
}
//公共参数不为空的时候
if(CollectionUtil.isNotEmpty(publicParam)) {
command = PlaceholderUtils.commandReplace(command,publicParam );
}
//替换运行参数中的时间参数
if (ScheduleTypeEnum.NORMAL.getCode().equals(mythJobTaskSchedule.getScheduleType())) {
mythJobTaskSchedule.setRunParam(PlaceholderUtils.formatParam(mythJobTaskSchedule.getRunParam()));
mythJobTaskSchedule.setRunParam(PlaceholderUtils.formatParam(privateParam));
}
String command = mythJobTaskSchedule.getRunCommand();
}
//除去JAVA任务
if(!NodeTypeEnum.JAVA.getCode().equals(mythJobTaskSchedule.getJobType())){
assert command != null;
command = command.replace("${"+ PlaceholderEnum.DATE_PLACEHOLDER.getName()+"}", DateUtil.dateLessDayStr(new Date(),"yyyyMMdd",1));
mythJobTaskSchedule.setRunCommand(command);
}
Long triggerTime = mythJobTaskSchedule.getTriggerTime();
TimerTask timerTask = null;
if (NodeTypeEnum.JAVA.getType().equals(mythJobTaskSchedule.getJobType())) {
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/util/FlowNodeRelyParseUtil.java 0 → 100644
package com.byit.util;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.dto.NodeRelyDto;
import com.byit.model.Node;
import com.byit.model.NodeDependencyKey;
import org.apache.commons.lang3.StringUtils;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;
/**
* 工作流节点依赖解析工具类
*
* @author huangfu
*/
public class FlowNodeRelyParseUtil {
/**
* 解析当前版本的节点依赖
*
* @param nodeAllList 所有的节点信息
* @param startNodeName 开始的节点名称
* @param allNodeDependencyKey 所有的主键信息
* @return 返回解析之后的节点
*/
public static Set<NodeRelyDto> parseThisVersionNodeRely(List<Node> nodeAllList, String startNodeName, List<NodeDependencyKey> allNodeDependencyKey, boolean hasStart) {
List<NodeRelyDto> tempNodeRelyDtoList = new ArrayList<>(8);
//构建节点id与节点的字典关系
Map<Integer, List<Node>> nodeMap = nodeAllList.stream().collect(Collectors.groupingBy(Node::getNodeId));
//筛选开始节点
Node meetTheCriteriaNode = nodeAllList.stream().filter(node -> startNodeName.equals(node.getNodeName())).findFirst().orElse(null);
if (meetTheCriteriaNode != null) {
//构建一个队列
Queue<Integer> queue = new LinkedList<>();
//初始化开始节点 将开始节点加入到当前的结果集里面 注意开始节点不包含任何依赖
if (hasStart) {
NodeRelyDto nodeRelyDto = new NodeRelyDto();
nodeRelyDto.setNode(meetTheCriteriaNode);
tempNodeRelyDtoList.add(nodeRelyDto);
//构建一个队列初始化
queue.offer(meetTheCriteriaNode.getNodeId());
}else{
List<NodeDependencyKey> nodeDependencyKeys = nodeIdMatch(allNodeDependencyKey, meetTheCriteriaNode.getNodeId());
nodeDependencyKeys.forEach(e ->{
NodeRelyDto nodeRelyDto = new NodeRelyDto();
Node node = nodeMap.get(e.getNodeId()).get(0);
nodeRelyDto.setNode(node);
tempNodeRelyDtoList.add(nodeRelyDto);
//构建一个队列初始化
queue.offer(node.getNodeId());
});
}
while (queue.size() > 0) {
Integer relyNodeId = queue.poll();
//寻找依赖节点是当前节点的节点id
//基于当前节点的id 查询所有依赖该节点的节点
List<NodeDependencyKey> nodeDependencyKeys = nodeIdMatch(allNodeDependencyKey, relyNodeId);
List<NodeRelyDto> nodeRelyDtos = buildRelyNode(nodeDependencyKeys, nodeMap, relyNodeId);
//遍历查询到的所由的依赖节点的信息
if (CollectionUtil.isNotEmpty(nodeRelyDtos)) {
nodeRelyDtos.forEach(nodeRely -> {
//向队列设置一个节点id
queue.offer(nodeRely.getNode().getNodeId());
//添加到最终结果集
tempNodeRelyDtoList.add(nodeRely);
});
}
}
return buildResult(tempNodeRelyDtoList);
}
return null;
}
/**
* 节点匹配
*
* @param allNodeDependencyKey 所有的节点主键
* @param thisNodeId 当前节点的id
* @return 匹配成功的
*/
public static List<NodeDependencyKey> nodeIdMatch(List<NodeDependencyKey> allNodeDependencyKey, Integer thisNodeId) {
List<NodeDependencyKey> nodeDependencyKeys = new ArrayList<>(8);
allNodeDependencyKey.forEach(key -> {
if (key.getDependencyId().equals(thisNodeId)) {
nodeDependencyKeys.add(key);
}
});
return nodeDependencyKeys;
}
/**
* 构建最终的结果
*
* @param tempNodeRelyDtoList 临时结果集
* @return 构建完成的数据
*/
private static Set<NodeRelyDto> buildResult(List<NodeRelyDto> tempNodeRelyDtoList) {
Set<NodeRelyDto> tempNodeRelyDtoSet = new HashSet<>(tempNodeRelyDtoList);
//全部节点构建完毕后 对结果集进行分组 以相同的节点为key 依赖节点为value 构建某一节点的依赖节点
Set<NodeRelyDto> nodeRelyDtoSet = new HashSet<>(8);
Map<Node, List<NodeRelyDto>> resultMap = tempNodeRelyDtoSet.stream().collect(Collectors.groupingBy(NodeRelyDto::getNode));
resultMap.forEach((node, nodeDtoList) -> {
//抽芯构建一个新的
NodeRelyDto nodeRelyDto = new NodeRelyDto();
nodeRelyDto.setNode(node);
List<String> relIdList = nodeDtoList.stream().map(NodeRelyDto::getRelyId).collect(Collectors.toList());
String relIds = StringUtils.join(relIdList, ",");
nodeRelyDto.setRelyId(relIds);
nodeRelyDtoSet.add(nodeRelyDto);
});
return nodeRelyDtoSet;
}
/**
* 构建依赖节点
* <p>
* 将依赖节点构建为节点 包装起来
*
* @param nodeDependencyKeys 当前节点依赖的节点id
* @param nodeMap 节点字典
* @param thisNodeId 当前的节点id
* @return 构建完成的节点信息
*/
private static List<NodeRelyDto> buildRelyNode(List<NodeDependencyKey> nodeDependencyKeys, Map<Integer, List<Node>> nodeMap, Integer thisNodeId) {
return nodeDependencyKeys.stream().map(nodeDependencyKey -> {
List<Node> nodes = nodeMap.get(nodeDependencyKey.getNodeId());
Node node = nodes.get(0);
NodeRelyDto nodeRelyDto = new NodeRelyDto();
nodeRelyDto.setNode(node);
//设置当前的依赖节点
nodeRelyDto.setRelyId(String.valueOf(thisNodeId));
return nodeRelyDto;
}).collect(Collectors.toList());
}
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/resources/mapper/NodeDependencyMapper.xml
......@@ -25,6 +25,18 @@
where dependency_id = #{nodeId,jdbcType=INTEGER}
</select>
<select id="findAllNodeDependencyKey" resultMap="BaseResultMap">
select node_id, dependency_id from node_dependency
<where>
node_id in (
<foreach collection="nodeIds" item="nodeId" separator=",">
#{nodeId}
</foreach>
)
</where>
</select>
<select id="findNodeInfoByNodeId" resultType="string">
select node.node_name
from node_dependency depend
......
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/executor/RunParamWrapped.java 0 → 100644
package com.byit.dto.executor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import java.io.Serializable;
import java.util.Map;
/**
* 运行参数包装逻辑
*
* @author huangfu
* @date 2020年9月24日15:32:39
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
@EqualsAndHashCode
public class RunParamWrapped implements Serializable {
private static final long serialVersionUID = -2926047980768857364L;
/**
* 公共参数字典
*/
private Map<String,String> publicParamMap;
/**
* 私有参数
*/
private String privateParam;
}
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/plugin/PluginFlowConfig.java
......@@ -5,6 +5,8 @@ import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.Map;
/**
* @description: 插件端工作流的配置
* @author: gml
......@@ -17,6 +19,11 @@ import lombok.NoArgsConstructor;
public class PluginFlowConfig {
/**
* 公共参数
*/
private Map<String,String> publicParam;
/**
* 当前工作流版本的告警的时机(0 不告警， 1 完成时告警， 2 失败时告警， 3 成功时告警)
*/
private String alarmlAction;
......
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/specials/SpecialJobParam.java
......@@ -4,6 +4,7 @@ import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.List;
import java.util.Map;
/**
......@@ -29,7 +30,7 @@ public class SpecialJobParam {
/**
* 是否补批本节点 1：是 2：否
*/
private String supplementStatus;
private String supplementStatus = "1";
/**
* 开始节点的名称
......@@ -39,6 +40,16 @@ public class SpecialJobParam {
/**
* 任务名称 -> 任务参数载体
*/
Map<String,String> paramCarrier;
private Map<String,String> paramCarrier;
/**
* 公共参数
*/
private Map<String,String> publicParam;
/**
* 时间参数
*/
private List<String> repairTimeList;
}
byit-myth-executor/myth-executor-server/src/main/java/com/byit/node/impl/CommandAndScriptProcessingMachine.java
......@@ -7,7 +7,6 @@ import cn.hutool.core.util.ZipUtil;
import com.alibaba.fastjson.JSON;
import com.byit.dto.executor.ScriptDto;
import com.byit.dto.executor.ScriptParamAndPlaceholderDto;
import com.byit.enums.PlaceholderEnum;
import com.byit.enums.SourceType;
import com.byit.exceptions.ExecutorException;
import com.byit.filesystem.FileSystem;
......@@ -21,11 +20,14 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.io.*;
import java.nio.charset.Charset;
import java.util.*;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.UUID;
/**
* 2.命令处理节点，主要做命令的参数替换操作
*
* @author huangfu
**/
@Component
......@@ -37,6 +39,7 @@ public class CommandAndScriptProcessingMachine implements ProcessingMachine {
*/
@Value("${myth-job.script.root.path}")
private String rootScriptPath;
public CommandAndScriptProcessingMachine(FileSystem fileSystem) {
this.fileSystem = fileSystem;
}
......@@ -46,7 +49,7 @@ public class CommandAndScriptProcessingMachine implements ProcessingMachine {
try {
log.debug("-----执行到命令和脚本的处理节点，开始初始化命令和脚本------------");
String command = scriptDto.getCommand();
log.debug("-----初始命令为{}------------",command);
log.debug("-----初始命令为{}------------", command);
String param = scriptDto.getParam();
ScriptParamAndPlaceholderDto paramAndPlaceholderDto = null;
if (StringUtils.isNotBlank(param)) {
......@@ -56,27 +59,22 @@ public class CommandAndScriptProcessingMachine implements ProcessingMachine {
if (StringUtils.isNoneBlank(scriptDto.getRemotePath())) {
//这里需要解压压缩包 然后将
String scriptDirPath = byteArrayToSource(scriptDto.getRemotePath());
log.debug("-----远程地址的文件被拉取到:[{}]------------",scriptDirPath);
//初始化命令信息
//command = PlaceholderUtils.initCommand(command,scriptDirPath);
//command = "cd "+scriptDirPath + "\n" + command;
log.debug("-----执行命令初始化完成，命令为：[{}]------------",command);
resourceKeywordReplacement(scriptDirPath,paramAndPlaceholderDto);
log.debug("-----远程地址的文件被拉取到:[{}]------------", scriptDirPath);
log.debug("-----执行命令初始化完成，命令为：[{}]------------", command);
resourceKeywordReplacement(scriptDirPath, paramAndPlaceholderDto);
scriptDto.setCommand(command);
scriptDto.setTempDir(scriptDirPath);
}
if (paramAndPlaceholderDto != null) {
//执行命令参数的替换
//command = PlaceholderUtils.commandReplace(command,paramAndPlaceholderDto.getParam());
log.debug("-----命令参数替换完成，命令为：[{}]------------",command);
scriptDto.setCommand(command);
}
// if (paramAndPlaceholderDto != null) {
// log.debug("-----命令参数替换完成，命令为：[{}]------------",command);
// scriptDto.setCommand(command);
// }
//将脚本的携带的参数 拼接在命令后方 python test.py -sex 男 -name 张三 -age 14
log.info("---------命令及脚本参数处理完成，此节点处理数据为:[{}]-----------",scriptDto);
processingNodeChain.doProcessing(scriptDto,processingNodeChain);
}catch (Exception e) {
log.info("---------命令及脚本参数处理完成，此节点处理数据为:[{}]-----------", scriptDto);
processingNodeChain.doProcessing(scriptDto, processingNodeChain);
} catch (Exception e) {
e.printStackTrace();
throw new ExecutorException(e);
}
......@@ -84,16 +82,17 @@ public class CommandAndScriptProcessingMachine implements ProcessingMachine {
/**
* 资源脚本替换
*
* @param dirPath
* @return
*/
private void resourceKeywordReplacement(String dirPath, ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto){
private void resourceKeywordReplacement(String dirPath, ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto) {
File baseFile = new File(dirPath);
List<File> files = FileUtil.loopFiles(baseFile);
if(scriptParamAndPlaceholderDto != null){
files.forEach(file ->{
if (scriptParamAndPlaceholderDto != null) {
files.forEach(file -> {
String fileName = file.getName();
if(SourceType.match(fileName)){
if (SourceType.match(fileName)) {
FileInputStream fileInputStream = IoUtil.toStream(file);
byte[] bytes = IoUtil.readBytes(fileInputStream);
try {
......@@ -101,8 +100,8 @@ public class CommandAndScriptProcessingMachine implements ProcessingMachine {
} catch (IOException e) {
e.printStackTrace();
}
bytes = PlaceholderUtils.resolvePlaceholders(bytes,scriptParamAndPlaceholderDto.getPlaceholder());
FileUtil.writeBytes(bytes,file);
bytes = PlaceholderUtils.resolvePlaceholders(bytes, scriptParamAndPlaceholderDto.getPlaceholder());
FileUtil.writeBytes(bytes, file);
}
});
}
......@@ -110,30 +109,33 @@ public class CommandAndScriptProcessingMachine implements ProcessingMachine {
/**
* 获取内部所有文件
*
* @param baseFile 文件
* @param files 所有文件
* @param files 所有文件
*/
public void getFile(File baseFile, List<File> files){
public void getFile(File baseFile, List<File> files) {
if (baseFile.isDirectory()) {
File[] fileList = baseFile.listFiles();
for (File file : fileList) {
if(file.isDirectory()){
getFile(file,files);
}else{
if (file.isDirectory()) {
getFile(file, files);
} else {
files.add(file);
}
}
}
}
/**
* 拉取文件，解压文件
*
* @param remotePath 文件路径
* @return 加压后的文件夹路径
*/
private String byteArrayToSource(String remotePath){
private String byteArrayToSource(String remotePath) {
//创建目录
File rootPathMkdir = new File(rootScriptPath, DateUtil.dateFormat(new Date(),"yyyyMMdd"));
if(!rootPathMkdir.exists()){
File rootPathMkdir = new File(rootScriptPath, DateUtil.dateFormat(new Date(), "yyyyMMdd"));
if (!rootPathMkdir.exists()) {
rootPathMkdir.mkdirs();
}
......@@ -143,8 +145,8 @@ public class CommandAndScriptProcessingMachine implements ProcessingMachine {
//获取文件元信息
Map<String, String> fileMate = fileSystem.getFileMate(remotePath);
String fileName = fileMate.get("filename");
fileName = UUID.randomUUID().toString().replace("-","")+fileName;
File file = new File(rootPathMkdir,fileName);
fileName = UUID.randomUUID().toString().replace("-", "") + fileName;
File file = new File(rootPathMkdir, fileName);
FileWriter.create(file).write(scriptByteArray, 0, scriptByteArray.length);
//解压后的文件夹路径
File unzip = ZipUtil.unzip(file);
......@@ -159,12 +161,13 @@ public class CommandAndScriptProcessingMachine implements ProcessingMachine {
/**
* 将脚本字节转换成文件
*
* @return 生成文件的本地路径
*/
private String byteArrayToFile(String remotePath,ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto){
private String byteArrayToFile(String remotePath, ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto) {
//创建目录
File rootPathMkdir = new File(rootScriptPath, DateUtil.dateFormat(new Date(),"yyyyMMdd"));
if(!rootPathMkdir.exists()){
File rootPathMkdir = new File(rootScriptPath, DateUtil.dateFormat(new Date(), "yyyyMMdd"));
if (!rootPathMkdir.exists()) {
rootPathMkdir.mkdirs();
}
......@@ -175,14 +178,14 @@ public class CommandAndScriptProcessingMachine implements ProcessingMachine {
//下载脚本文件
byte[] scriptByteArray = fileSystem.downloaderFile(remotePath);
//替换脚本占位符
if(scriptParamAndPlaceholderDto != null){
scriptByteArray = PlaceholderUtils.resolvePlaceholders(scriptByteArray,scriptParamAndPlaceholderDto.getPlaceholder());
if (scriptParamAndPlaceholderDto != null) {
scriptByteArray = PlaceholderUtils.resolvePlaceholders(scriptByteArray, scriptParamAndPlaceholderDto.getPlaceholder());
}
//获取文件元信息
Map<String, String> fileMate = fileSystem.getFileMate(remotePath);
String fileName = fileMate.get("filename");
fileName = UUID.randomUUID().toString().replace("-","")+fileName;
file = new File(rootPathMkdir,fileName);
fileName = UUID.randomUUID().toString().replace("-", "") + fileName;
file = new File(rootPathMkdir, fileName);
out = new FileOutputStream(file);
//写出脚本文件
out.write(scriptByteArray);
......@@ -190,7 +193,7 @@ public class CommandAndScriptProcessingMachine implements ProcessingMachine {
} catch (IOException e) {
e.printStackTrace();
} finally {
if(out != null){
if (out != null) {
try {
out.close();
} catch (IOException e) {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment