Skip to content

B
byit-myth-job
Commit a99c4b7e by huangfusuper
Options

删除所有的公共参数替换，发布工作流的时候保存工作流的内嵌层级

parent d5ac8f22
Showing with 188 additions and 257 deletions
byit-myth-admin/src/main/java/com/byit/service/impl/ApiFlowOperatingServiceImpl.java
......@@ -4,7 +4,6 @@ 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.executor.ScriptParamAndPlaceholderDto;
import com.byit.dto.plugin.FlowExtendedConfiguration;
import com.byit.dto.specials.MakeUpReturn;
import com.byit.dto.specials.RepairFlow;
......@@ -12,8 +11,8 @@ import com.byit.dto.specials.RepairTimeParam;
import com.byit.dto.specials.SpecialJobParam;
import com.byit.enums.NodeTypeEnum;
import com.byit.enums.PlaceholderEnum;
import com.byit.enums.RedisKeyNameEnum;
import com.byit.enums.ScheduleTypeEnum;
import com.byit.job.utils.CurrentUserUtils;
import com.byit.job.utils.DateUtil;
import com.byit.job.utils.PlaceholderUtils;
import com.byit.mapper.RunRecordingMapper;
......@@ -28,6 +27,7 @@ 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.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
......@@ -55,22 +55,23 @@ public class ApiFlowOperatingServiceImpl implements ApiFlowOperatingService {
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;
private final StringRedisTemplate stringRedisTemplate;
public ApiFlowOperatingServiceImpl(FlowService flowService, WorkspaceService workspaceService, NodeService nodeService,
NodeDependencyService nodeDependencyService, CurrentUserUtils currentUserUtils,
WaitingRecordMapper waitingRecordMapper, RunRecordingMapper runRecordingMapper, WaitingTaskMapper waitingTaskMapper) {
NodeDependencyService nodeDependencyService, WaitingRecordMapper waitingRecordMapper,
RunRecordingMapper runRecordingMapper, WaitingTaskMapper waitingTaskMapper,
StringRedisTemplate stringRedisTemplate) {
this.flowService = flowService;
this.workspaceService = workspaceService;
this.nodeService = nodeService;
this.nodeDependencyService = nodeDependencyService;
this.currentUserUtils = currentUserUtils;
this.waitingRecordMapper = waitingRecordMapper;
this.runRecordingMapper = runRecordingMapper;
this.waitingTaskMapper = waitingTaskMapper;
this.stringRedisTemplate = stringRedisTemplate;
}
@Override
......@@ -117,65 +118,17 @@ public class ApiFlowOperatingServiceImpl implements ApiFlowOperatingService {
List<NodeDependencyKey> allNodeDependencyKey = nodeDependencyService.findAllNodeDependencyKey(integerList);
//解析对应的版本依赖
Set<NodeRelyDto> nodeRelyDtoSet = FlowNodeRelyParseUtil.parseThisVersionNodeRely(nodeByFlowIdAndVersionName, startNodeTaskName, allNodeDependencyKey, "1".equals(supplementStatus));
if (CollectionUtil.isNotEmpty(nodeRelyDtoSet)) {
//循环所有的结果集
nodeRelyDtoSet.forEach(nodeWrapped -> {
Node node = nodeWrapped.getNode();
String runCommand = node.getRunCommand();
//将公共参数放到替换参数里面
if(CollectionUtil.isNotEmpty(publicParam) && StringUtils.isNoneBlank(runCommand) && !runCommand.startsWith("java")){
//获取运行参数
String runParam = node.getRunParam();
//创建包装参数
RunParamWrapped runParamWrapped = new RunParamWrapped();
//如果运行参数不为空
if(StringUtils.isNoneBlank(runParam)){
//json化运行参数
runParamWrapped = JSON.parseObject(runParam, RunParamWrapped.class);
//获取私有参数
String privateParam = runParamWrapped.getPrivateParam();
//如果私有参数不为空
if(StringUtils.isNoneBlank(privateParam)) {
//转换成参数对象
ScriptParamAndPlaceholderDto paramAndPlaceholderDto = JSON.parseObject(runParamWrapped.getPrivateParam(), ScriptParamAndPlaceholderDto.class);
//获取替换参数
Map<String, String> placeholder = paramAndPlaceholderDto.getPlaceholder();
if(placeholder == null){
placeholder = new HashMap<>(8);
}
//将公共参数放到替换参数
for (Map.Entry<String, String> stringStringEntry : publicParam.entrySet()) {
placeholder.put(stringStringEntry.getKey(),stringStringEntry.getValue());
}
paramAndPlaceholderDto.setPlaceholder(placeholder);
//重新设置包装对象
runParamWrapped.setPrivateParam(JSON.toJSONString(paramAndPlaceholderDto));
//如果运行参数为空
}else{
//创建一个参数对象
ScriptParamAndPlaceholderDto paramAndPlaceholderDto = new ScriptParamAndPlaceholderDto();
//设置替换参数
paramAndPlaceholderDto.setPlaceholder(publicParam);
//设置参数包装对象的私有参数
runParamWrapped.setPrivateParam(JSON.toJSONString(paramAndPlaceholderDto));
}
//运行参数为空
}else {
//重复上述步骤
ScriptParamAndPlaceholderDto paramAndPlaceholderDto = new ScriptParamAndPlaceholderDto();
paramAndPlaceholderDto.setPlaceholder(publicParam);
runParamWrapped.setPrivateParam(JSON.toJSONString(paramAndPlaceholderDto));
}
node.setRunParam(JSON.toJSONString(runParamWrapped));
}
//TODO 原来的参数替换位置
String nodeName = node.getNodeName();
//基于节点名称替换参数
String param = parseParam(paramCarrier,nodeName);
//1. 将节点的某些参数替换掉
//2. 将节点的公共参数设置到节点中
String param = parseParam(paramCarrier, nodeName);
if (StringUtils.isNoneBlank(param)) {
//获取包装逻辑
RunParamWrapped runParamWrapped = new RunParamWrapped();
......@@ -183,34 +136,22 @@ public class ApiFlowOperatingServiceImpl implements ApiFlowOperatingService {
runParamWrapped.setPublicParamMap(publicParam);
//构建参数
node.setRunParam(JSON.toJSONString(runParamWrapped));
}else{
} else {
String runParam = node.getRunParam();
RunParamWrapped runParamWrapped = new RunParamWrapped();
if(StringUtils.isNoneBlank(runParam)) {
if (StringUtils.isNoneBlank(runParam)) {
runParamWrapped = JSON.parseObject(runParam, RunParamWrapped.class);
}
runParamWrapped.setPublicParamMap(publicParam);
//构建参数
node.setRunParam(JSON.toJSONString(runParamWrapped));
}
if(StringUtils.isNoneBlank(runCommand)){
if(runCommand.startsWith(JAVA_TYPE) && CollectionUtil.isNotEmpty(publicParam)){
for (Map.Entry<String, String> stringStringEntry : publicParam.entrySet()) {
if(!"auditTime".equals(stringStringEntry.getKey())){
String keyName = String.format("${%s}",stringStringEntry.getKey());
runCommand = runCommand.replace(keyName,stringStringEntry.getValue());
}
}
node.setRunCommand(runCommand);
}
}
//TODO 参数替换
});
}
if (CollectionUtil.isNotEmpty(nodeRelyDtoSet)) {
//构建等待队列
List<WaitingRecord> waitingRecords = buildWaitingRecordList(flowByName, nodeRelyDtoSet.size(), operator, repairTimeList,JSON.toJSONString(publicParam));
List<WaitingRecord> waitingRecords = buildWaitingRecordList(flowByName, nodeRelyDtoSet.size(), operator, repairTimeList, JSON.toJSONString(publicParam));
waitingRecords.forEach(waitingRecordMapper::insertSelective);
List<RunRecording> runRecordings = buildRunRecordingList(waitingRecords);
List<WaitingTask> waitingTaskList = buildWaitingTaskResult(waitingRecords, nodeRelyDtoSet, specialJobParam.getPublicParam(), timeFormatName, nowTimeFormatName);
......@@ -231,12 +172,13 @@ public class ApiFlowOperatingServiceImpl implements ApiFlowOperatingService {
/**
* 解析参数
*
* @param paramCarrier 参数信息
* @param nodeName 节点名称
* @return
*/
public String parseParam(Map<String, String> paramCarrier, String nodeName){
if(CollectionUtil.isNotEmpty(paramCarrier)){
public String parseParam(Map<String, String> paramCarrier, String nodeName) {
if (CollectionUtil.isNotEmpty(paramCarrier)) {
return paramCarrier.get(nodeName);
}
return null;
......@@ -264,7 +206,7 @@ public class ApiFlowOperatingServiceImpl implements ApiFlowOperatingService {
* @param nodeRelyDtoSet 节点对象
* @return 返回一个等待队的全部节点
*/
private List<WaitingTask> buildWaitingTask(WaitingRecord waitingRecord, Set<NodeRelyDto> nodeRelyDtoSet, Map<String, String> publicParam, String timeFormat, String nowDateFormat) {
protected List<WaitingTask> buildWaitingTask(WaitingRecord waitingRecord, Set<NodeRelyDto> nodeRelyDtoSet, Map<String, String> publicParam, String timeFormat, String nowDateFormat) {
return nodeRelyDtoSet.stream().map(nodeRelyDto -> {
Node node = nodeRelyDto.getNode();
String relyId = nodeRelyDto.getRelyId();
......@@ -282,7 +224,7 @@ public class ApiFlowOperatingServiceImpl implements ApiFlowOperatingService {
waitingTask.setWaitId(waitingRecord.getWaitId());
RunParamWrapped runParamWrapped = JSON.parseObject(waitingTask.getRunParam(), RunParamWrapped.class);
if(runParamWrapped == null) {
if (runParamWrapped == null) {
runParamWrapped = new RunParamWrapped();
}
runParamWrapped.setPublicParamMap(publicParam);
......@@ -366,29 +308,31 @@ public class ApiFlowOperatingServiceImpl implements ApiFlowOperatingService {
String repeatTime = repairTimeParam.getRepairTime();
String timeTypeName = repairTimeParam.getTimeTypeName();
WaitingRecord waitingRecord = new WaitingRecord();
BeanUtils.copyProperties(flow, waitingRecord);
//设置排期
BeanUtils.copyProperties(flow, waitingRecord);
String extendedConfiguration = flow.getExtendedConfiguration();
FlowExtendedConfiguration flowExtendedConfiguration = JSON.parseObject(extendedConfiguration, FlowExtendedConfiguration.class);
if(flowExtendedConfiguration != null){
if (flowExtendedConfiguration != null) {
flowExtendedConfiguration.setPublicParam(publicParam);
}else{
} else {
flowExtendedConfiguration = new FlowExtendedConfiguration();
flowExtendedConfiguration.setPublicParam(publicParam);
}
waitingRecord.setExtendedConfiguration(JSON.toJSONString(flowExtendedConfiguration));
waitingRecord.setFlowVersionName(flow.getVersionName());
waitingRecord.setRunId(IDGenerationStrategy.runIdGenerationStrategy(serverPort));
String runId = IDGenerationStrategy.runIdGenerationStrategy(serverPort);
//将公共参数保存到redis
String redisKey = String.format(RedisKeyNameEnum.REDIS_PUBLIC_PARAM_KEY.getKeyName(), runId);
stringRedisTemplate.opsForValue().set(redisKey, publicParam);
waitingRecord.setRunId(runId);
waitingRecord.setFlowNodeCount(nodeCount);
waitingRecord.setOperator(operator);
waitingRecord.setScheduleType(ScheduleTypeEnum.REPAIR.getCode());
waitingRecord.setRepeatTime(repeatTime);
waitingRecord.setWaitOrder(++order);
//需要按着时间先后来设置时间
long time = DateUtil.strFormatDate(repeatTime, timeTypeName).getTime();
waitingRecord.setTriggerTime(time);
waitingRecord.setTriggerTime(System.currentTimeMillis());
waitingRecords.add(waitingRecord);
}
......
byit-myth-admin/src/main/java/com/byit/service/impl/ApiFlowServiceImpl.java
......@@ -14,7 +14,6 @@ import com.byit.dto.specials.RepairFlow;
import com.byit.dto.specials.RepairTimeParam;
import com.byit.enums.*;
import com.byit.enums.plugin.PluginNodeTypeEnum;
import com.byit.enums.task.RunResultEnum;
import com.byit.job.utils.CronExpression;
import com.byit.job.utils.CurrentUserUtils;
import com.byit.job.utils.PlaceholderUtils;
......@@ -132,7 +131,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
Workspace workspace = validate(pluginPackage);
//保存工作流
log.debug("保存工作流和节点信息");
Flow flow = saveFlow(pluginPackage.getFlow(), workspace.getWorkspaceId(), false);
Flow flow = saveFlow(pluginPackage.getFlow(), workspace.getWorkspaceId(), false, "");
//生成工作流版本
log.debug("生成版本");
FlowVersion flowVersion = saveFlowVersion(flow);
......@@ -149,10 +148,10 @@ public class ApiFlowServiceImpl implements ApiFlowService {
* @param isInnner
* @param workspaceId
*/
private Flow saveFlow(PluginFlow pluginFlow, Integer workspaceId, boolean isInnner) throws Exception {
private Flow saveFlow(PluginFlow pluginFlow, Integer workspaceId, boolean isInnner, String upFlowName) throws Exception {
//拼装工作流名称
upFlowName = String.format("%s:%s",upFlowName,pluginFlow.getName());
Flow flow = new Flow();
flow.setExtendedConfiguration(JSON.toJSONString(pluginFlow.getConfig().getExtendedConfiguration()));
flow.setFlowName(pluginFlow.getName());
flow.setIsInner(isInnner ? "0" : "1");
flow.setFlowNodeCount(pluginFlow.getNodeList().size());
......@@ -167,8 +166,6 @@ 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())) {
......@@ -197,7 +194,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
flow.setVersionName(versionName);
}
flowMapper.updateByIdSelective(flow);
updateNodeByFlow(pluginFlow.getNodeList(), flow, publicParamMap);
updateNodeByFlow(pluginFlow.getNodeList(), flow, upFlowName);
} else {
String versionName = pluginFlow.getVersionName();
......@@ -208,9 +205,11 @@ public class ApiFlowServiceImpl implements ApiFlowService {
}
Integer flowId = flowMapper.insertSelective(flow);
saveNode(pluginFlow.getNodeList(), flow, publicParamMap);
saveNode(pluginFlow.getNodeList(), flow, upFlowName);
}
FlowExtendedConfiguration extendedConfiguration = pluginFlow.getConfig().getExtendedConfiguration();
extendedConfiguration.setFlowEmbedLogo(upFlowName);
flow.setExtendedConfiguration(JSON.toJSONString(extendedConfiguration));
return flow;
}
......@@ -220,7 +219,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
* @param pluginNodeList
* @param flow
*/
private void updateNodeByFlow(List<PluginBaseNode> pluginNodeList, Flow flow, Map<String, String> publicParamMap) throws Exception {
private void updateNodeByFlow(List<PluginBaseNode> pluginNodeList, Flow flow, String upFlowName) throws Exception {
//将工作流下的所有节点设为不在工作流调度上
log.debug("将工作流【{}】所有的节点移下调度,并删除相关的依赖关系", flow.getFlowName());
......@@ -250,7 +249,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
//保存节点信息
for (PluginBaseNode pluginNode : pluginNodeList) {
Node node = new Node();
buildNode(pluginNode, flow, node, publicParamMap);
buildNode(pluginNode, flow, node, upFlowName);
Integer nodeId = nodeMapper.insertSelective(node);
nameIdRel.put(node.getNodeName(), node.getNodeId());
......@@ -265,12 +264,12 @@ public class ApiFlowServiceImpl implements ApiFlowService {
* @param nodeList
* @param flow
*/
private void saveNode(List<PluginBaseNode> nodeList, Flow flow, Map<String, String> publicParamMap) throws Exception {
private void saveNode(List<PluginBaseNode> nodeList, Flow flow, String upFlowName) throws Exception {
HashMap<String, Integer> nameIdRel = new HashMap<>();
//保存节点信息
for (PluginBaseNode pluginNode : nodeList) {
Node node = new Node();
buildNode(pluginNode, flow, node, publicParamMap);
buildNode(pluginNode, flow, node, upFlowName);
node.setNodeId(null);
Integer nodeId = nodeMapper.insertSelective(node);
nameIdRel.put(node.getNodeName(), node.getNodeId());
......@@ -286,7 +285,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
* @param node
* @throws Exception
*/
private void buildNode(PluginBaseNode pluginNode, Flow flow, Node node, Map<String, String> publicParamMap) throws Exception {
private void buildNode(PluginBaseNode pluginNode, Flow flow, Node node, String upFlowName) throws Exception {
node.setVersionName(flow.getVersionName());
node.setFlowId(flow.getFlowId());
......@@ -299,7 +298,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
PluginFlow pluginFlow = (PluginFlow) pluginNode;
pluginFlow.getConfig().setFlowCron(flow.getFlowCron());
//如果是内嵌工作流先保存工作流信息
Flow innerFlow = saveFlow(pluginFlow, flow.getWorkspaceId(), true);
Flow innerFlow = saveFlow(pluginFlow, flow.getWorkspaceId(), true, upFlowName);
node.setIsVirtual(NodePropertyEnum.IS_VIRTUAL.getCode());
node.setMapFlowId(innerFlow.getFlowId());
node.setVersionName(pluginFlow.getVersionName());
......@@ -342,11 +341,6 @@ public class ApiFlowServiceImpl implements ApiFlowService {
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));
}
}
......@@ -809,7 +803,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
}
}
return runRecordingMapper.findByStartAndEndTime(loadScheduleCondition,flowIds);
return runRecordingMapper.findByStartAndEndTime(loadScheduleCondition, flowIds);
}
......@@ -1669,7 +1663,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
StringBuffer runids = new StringBuffer();
//暂停工作流调度
recordingList.forEach(runRecording -> {
if(RunRecordingEnum.FLOW_STATUS_IS_END.getCode().equals(runRecording.getFlowStatus())){
if (RunRecordingEnum.FLOW_STATUS_IS_END.getCode().equals(runRecording.getFlowStatus())) {
throw new RuntimeException(String.format("实例%s是完结状态", runRecording.getFlowName()));
}
//runRecordingMapper.stopByRunIdAndFlowId(runRecording.getRunId(), runRecording.getFlowId());
......@@ -1694,7 +1688,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
List<RunRecording> runRecordList = runRecordingMapper.findStopRunCordByRunId(runId);
//ValidationUtil.isTrueValidation(CollectionUtil.isEmpty(runRecordList), "该工作流不是暂停状态！");
if(CollectionUtil.isEmpty(runRecordList)){
if (CollectionUtil.isEmpty(runRecordList)) {
log.warn("---------运行记录{}，不存在暂停记录中！---", runId);
return;
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/enums/RedisKeyNameEnum.java 0 → 100644
package com.byit.enums;
/**
* @author huangfu
*/
public enum RedisKeyNameEnum {
/**
* 调度中心公共参数
*/
REDIS_PUBLIC_PARAM_KEY("myth:param:public:%s"),
;
private final String keyName;
RedisKeyNameEnum(String keyName) {
this.keyName = keyName;
}
public String getKeyName() {
return keyName;
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/RunRecording.java
......@@ -88,7 +88,7 @@ public class RunRecording implements Serializable {
/**
* 本次任务的执行时间
*/
@ApiModelProperty("本次任务的执行时间")
@ApiModelProperty("计划本次任务开始的时间")
private Long triggerTime;
/**
......@@ -106,13 +106,13 @@ public class RunRecording implements Serializable {
/**
* 开始时间
*/
@ApiModelProperty("开始时间")
@ApiModelProperty("实际开始时间")
private Date startTime;
/**
* 结束时间
*/
@ApiModelProperty("结束时间")
@ApiModelProperty("实际结束时间")
private Date endTime;
/**
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/RunNodeServiceImpl.java
package com.byit.service.impl;
import com.byit.dto.RunRecordingWrapped;
import com.byit.enums.NodeNameEnum;
import com.byit.enums.RunRecordingEnum;
import com.byit.enums.ScheduleTypeEnum;
......@@ -32,7 +31,7 @@ import java.util.*;
* @author huangfu
*/
@Service
@Transactional(propagation = Propagation.REQUIRED,rollbackFor = Exception.class)
@Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
@Slf4j
public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublisherAware {
private final RunRecordingService runRecordingService;
......@@ -56,16 +55,17 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
/**
* 保存到运行记录一份 将节点保存到job_task表
*
* @param flow 工作流
* @param nodes 节点
*/
@Override
public void saveRunRec(Flow flow, List<Node> nodes) {
log.info("---------saveRunRecAndTask start------【保存工作流：{}和节点：{}】-----------------------",flow,nodes);
log.info("---------saveRunRecAndTask start------【保存工作流：{}和节点：{}】-----------------------", flow, nodes);
String runId = IDGenerationStrategy.runIdGenerationStrategy(serverPort);
log.info("-------------【开始保存运行记录runId为：{}】------------------",runId);
log.info("-------------【开始保存运行记录runId为：{}】------------------", runId);
RunRecording build = new RunRecording();
BeanUtils.copyProperties(flow,build);
BeanUtils.copyProperties(flow, build);
build.setRunId(runId);
//TODO 这个不解释 不知道干嘛的 后续需要修改
build.setDispatchIp("0.0.0.0");
......@@ -73,19 +73,17 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
build.setTriggerTime(flow.getTriggerNextTime());
boolean isScheduleFollow = "1".equals(flow.getScheduleFollow());
log.info("-------------【开始保存节点信息，任务是否为跟随工作流,{}】---------------",isScheduleFollow);
log.info("-------------【开始保存节点信息，任务是否为跟随工作流,{}】---------------", isScheduleFollow);
List<JobTask> jobTasks = new ArrayList<>(32);
for (Node node : nodes) {
JobTask jobTask = new JobTask();
jobTask.setScheduleType(1);
jobTask.setTriggerStatus("1");
BeanUtils.copyProperties(node,jobTask);
if(isScheduleFollow){
BeanUtils.copyProperties(node, jobTask);
if (isScheduleFollow) {
jobTask.setTriggerTime(flow.getTriggerNextTime());
}else {
} else {
jobTask.setTriggerTime(node.getTriggerNextTime());
try {
node.setTriggerNextTime(new CronExpression(node.getNodeCron()).getNextValidTimeAfter(new Date(node.getTriggerNextTime())).getTime());
......@@ -98,7 +96,7 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
jobTask.setFlowName(flow.getFlowName());
jobTask.setScheduleType(ScheduleTypeEnum.NORMAL.getCode());
//如果不是开始节点
if(!NodeNameEnum.START_NODE.getNodeName().equals(jobTask.getNodeName())){
if (!NodeNameEnum.START_NODE.getNodeName().equals(jobTask.getNodeName())) {
List<Integer> dependIdByNodeId = nodeDependencyService.findDependIdByNodeId(jobTask.getNodeId());
String parentIds = StringUtils.join(dependIdByNodeId, ",");
jobTask.setNodeDepend(parentIds);
......@@ -112,12 +110,13 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
//回调生命周期
try {
for (Map.Entry<String, InstanceRunTheLifeCycleCallback> entry : entries) {
log.info("-------------【开始回调周期{}前置】---------------",entry.getKey());
RunRecordingWrapped runRecordingWrapped = entry.getValue().postProcessAfterInitialization(build, jobTasks);
build = runRecordingWrapped.getRunRecording();
jobTasks = runRecordingWrapped.getNodeList();
log.info("-------------【开始回调周期{}前置】---------------", entry.getKey());
RunRecording runRecording = entry.getValue().postProcessAfterInitialization(build);
if (runRecording != null) {
build = runRecording;
}
}catch (Exception e) {
}
} catch (Exception e) {
e.printStackTrace();
build.setFlowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
build.setFlowRunResult(RunResultEnum.RUN_ERROR.getCode());
......@@ -130,7 +129,7 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
jobTaskService.saveJobTasks(jobTasks);
log.info("-------------【开始修改工作流{}的下次运行时间，以及各种状态】---------------",flow);
log.info("-------------【开始修改工作流{}的下次运行时间，以及各种状态】---------------", flow);
try {
Long triggerNextTime = flow.getTriggerNextTime();
flow.setTriggerNextTime(new CronExpression(flow.getFlowCron()).getNextValidTimeAfter(new Date(triggerNextTime)).getTime());
......@@ -140,26 +139,27 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
e.printStackTrace();
}
updateFlow(flow);
applicationEventPublisher.publishEvent(new FlowScanEndEvent(this,build));
applicationEventPublisher.publishEvent(new FlowScanEndEvent(this, build));
log.info("-------saveRunRecAndTaskAndUpdate end-----------【运行结束】-----------------");
}
/**
* 修改工作流的信息
*
* @param flow 工作流
*/
private void updateFlow(Flow flow){
if (flow.getRemainingCount()>0) {
flow.setRemainingCount(flow.getRemainingCount()-1);
private void updateFlow(Flow flow) {
if (flow.getRemainingCount() > 0) {
flow.setRemainingCount(flow.getRemainingCount() - 1);
}
flowService.updateByIdSelective(flow);
}
/**
* 设置时间发布器
*
* @param applicationEventPublisher
*/
@Override
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/impl/RunRecordingAndJobTaskServiceImpl.java
......@@ -274,9 +274,11 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask
//回调生命周期
for (Map.Entry<String, InstanceRunTheLifeCycleCallback> entry : entries) {
log.info("-------------【开始回调周期{}前置】---------------",entry.getKey());
RunRecordingWrapped runRecordingWrapped = entry.getValue().postProcessAfterInitialization(runRecording, jobTasks);
runRecording = runRecordingWrapped.getRunRecording();
jobTasks = runRecordingWrapped.getNodeList();
RunRecording runRecordingLi = entry.getValue().postProcessAfterInitialization(runRecording);
if(runRecordingLi != null) {
runRecording = runRecordingLi;
}
}
}catch (Exception e) {
e.printStackTrace();
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/strategy/InstanceRunTheLifeCycleCallback.java
......@@ -17,14 +17,15 @@ public interface InstanceRunTheLifeCycleCallback {
/**
* 实例执行前
* @param runRecording 运行实例
* @param jobTaskList 节点
* @return 两者的包装对象
*/
RunRecordingWrapped postProcessAfterInitialization(RunRecording runRecording, List<JobTask> jobTaskList);
default RunRecording postProcessAfterInitialization(RunRecording runRecording){
return runRecording;
}
/**
* 后置处理器
* @param runRecording 实例
*/
void postProcessBeforeInitialization(RunRecording runRecording);
default void postProcessBeforeInitialization(RunRecording runRecording){}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/strategy/impl/RpcCallbackInstanceRunTheLifeCycleCallback.java
......@@ -3,14 +3,13 @@ package com.byit.strategy.impl;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.byit.call.TaskServer;
import com.byit.dto.RunRecordingWrapped;
import com.byit.dto.common.CallbackDto;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.executor.ScriptParamAndPlaceholderDto;
import com.byit.dto.plugin.FlowExtendedConfiguration;
import com.byit.dto.web.ReturnResult;
import com.byit.enums.CallMethodEnum;
import com.byit.model.JobTask;
import com.byit.enums.RedisKeyNameEnum;
import com.byit.enums.RunRecordingEnum;
import com.byit.model.RunRecording;
import com.byit.model.Workspace;
import com.byit.packet.request.PluginRpcRequestPacket;
......@@ -25,9 +24,7 @@ import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
/**
* rpc回调平台服务
......@@ -54,115 +51,41 @@ public class RpcCallbackInstanceRunTheLifeCycleCallback implements InstanceRunTh
* 实例执行前
*
* @param runRecording 运行实例
* @param jobTaskList 节点
* @return 两者的包装对象
*/
@Override
public RunRecordingWrapped postProcessAfterInitialization(RunRecording runRecording, List<JobTask> jobTaskList) {
public RunRecording postProcessAfterInitialization(RunRecording runRecording) {
String extendedConfiguration = runRecording.getExtendedConfiguration();
FlowExtendedConfiguration flowExtendedConfiguration = JSON.parseObject(extendedConfiguration, FlowExtendedConfiguration.class);
if(flowExtendedConfiguration == null){
return new RunRecordingWrapped(jobTaskList,runRecording);
if (flowExtendedConfiguration == null) {
return runRecording;
}
String rpcServerKey = flowExtendedConfiguration.getRpcStartServerKey();
if(StringUtils.isBlank(rpcServerKey)){
return new RunRecordingWrapped(jobTaskList,runRecording);
if (StringUtils.isBlank(rpcServerKey)) {
return runRecording;
}
CallbackDto callbackDto = new CallbackDto();
PluginRpcRequestPacket param = buildPluginRpcRequestPacket(runRecording,callbackDto,rpcServerKey);
PluginRpcRequestPacket param = buildPluginRpcRequestPacket(runRecording, callbackDto, rpcServerKey);
PluginRpcResponsePacket call = taskServer.call(param);
if (!call.isStatus()) {
throw new RuntimeException(call.getMsg());
}
Object result = call.getResult();
ReturnResult<String> returnResult = (ReturnResult<String>) result;
ScriptParamAndPlaceholderDto rpcScriptParamAndPlaceholderDto = JSON.parseObject(returnResult.getContent(), ScriptParamAndPlaceholderDto.class);
Map<String, String> lifeCyclePublicParam = JSON.parseObject(returnResult.getContent(), Map.class);
//放置到对应的redis里面
stringRedisTemplate.opsForValue().set(String.format(EXTENDED_PARAMS_S,runRecording.getRunId()),JSON.toJSONString(rpcScriptParamAndPlaceholderDto));
if (rpcScriptParamAndPlaceholderDto != null) {
List<JobTask> jobTasks = jobTaskList.stream().map(jobTask -> {
//新增参数
Map<String, String> placeholder = rpcScriptParamAndPlaceholderDto.getPlaceholder();
//新增参数
Map<String, String> scriptParam = rpcScriptParamAndPlaceholderDto.getParam();
//获取原本的数据
String runParam = jobTask.getRunParam();
//如果原本数据为空 直接重新设置然后返回
if (StringUtils.isBlank(runParam)) {
RunParamWrapped runParamWrapped = new RunParamWrapped();
runParamWrapped.setPrivateParam(JSON.toJSONString(rpcScriptParamAndPlaceholderDto));
jobTask.setRunParam(JSON.toJSONString(runParamWrapped));
return jobTask;
}
//转换节点参数为包装对象
RunParamWrapped runParamWrapped = JSON.parseObject(runParam, RunParamWrapped.class);
if("JAVA".equals(jobTask.getJobType())){
Map<String, String> publicParamMap = runParamWrapped.getPublicParamMap();
if(CollectionUtil.isEmpty(publicParamMap)){
runParamWrapped.setPublicParamMap(new HashMap<>(8));
}
Map<String, String> javaParamPublic = runParamWrapped.getPublicParamMap();
if(CollectionUtil.isNotEmpty(scriptParam)){
javaParamPublic.putAll(scriptParam);
}
if(CollectionUtil.isNotEmpty(placeholder)){
javaParamPublic.putAll(placeholder);
}
runParamWrapped.setPublicParamMap(javaParamPublic);
}else{
String privateParam = runParamWrapped.getPrivateParam();
//如果对应参数为空直接设置
if (StringUtils.isBlank(privateParam)) {
runParamWrapped.setPrivateParam(JSON.toJSONString(rpcScriptParamAndPlaceholderDto));
jobTask.setRunParam(JSON.toJSONString(runParamWrapped));
return jobTask;
}
//获取任务里面的数据
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = JSON.parseObject(privateParam, ScriptParamAndPlaceholderDto.class);
//替换参数
if (CollectionUtil.isNotEmpty(placeholder)) {
//获取任务里面的替换参数
Map<String, String> jobTaskPlaceholder = scriptParamAndPlaceholderDto.getPlaceholder();
//如果任务里面的替换参数为空 就直接设置
if (CollectionUtil.isEmpty(jobTaskPlaceholder)) {
scriptParamAndPlaceholderDto.setPlaceholder(placeholder);
} else {
placeholder.forEach(jobTaskPlaceholder::put);
//追加数据
scriptParamAndPlaceholderDto.setPlaceholder(jobTaskPlaceholder);
}
}
//脚本参数
if (CollectionUtil.isNotEmpty(scriptParam)) {
Map<String, String> jobTaskParam = scriptParamAndPlaceholderDto.getParam();
//如果任务里面的替换参数为空 就直接设置
if (!CollectionUtil.isEmpty(jobTaskParam)) {
scriptParam.forEach(jobTaskParam::put);
//追加数据
}
scriptParamAndPlaceholderDto.setParam(jobTaskParam);
}
runParamWrapped.setPrivateParam(JSON.toJSONString(scriptParamAndPlaceholderDto));
}
jobTask.setRunParam(JSON.toJSONString(runParamWrapped));
return jobTask;
}).collect(Collectors.toList());
stringRedisTemplate.opsForValue().set(String.format(EXTENDED_PARAMS_S, runRecording.getRunId()), JSON.toJSONString(lifeCyclePublicParam));
String publicParam = flowExtendedConfiguration.getPublicParam();
Map<String, String> publicParamMap = JSON.parseObject(publicParam, Map.class);
publicParamMap.putAll(lifeCyclePublicParam);
RunRecordingWrapped runRecordingWrapped = new RunRecordingWrapped();
runRecordingWrapped.setNodeList(jobTasks);
runRecordingWrapped.setRunRecording(runRecording);
return runRecordingWrapped;
}
RunRecordingWrapped runRecordingWrapped = new RunRecordingWrapped();
runRecordingWrapped.setNodeList(jobTaskList);
runRecordingWrapped.setRunRecording(runRecording);
return runRecordingWrapped;
String publicParamMapValue = JSON.toJSONString(publicParamMap);
flowExtendedConfiguration.setPublicParam(publicParamMapValue);
String format = String.format(RedisKeyNameEnum.REDIS_PUBLIC_PARAM_KEY.getKeyName(), runRecording.getRunId());
//放置到对应的redis里面
stringRedisTemplate.opsForValue().set(format, publicParamMapValue);
runRecording.setExtendedConfiguration(JSON.toJSONString(flowExtendedConfiguration));
return runRecording;
}
/**
......@@ -171,7 +94,7 @@ public class RpcCallbackInstanceRunTheLifeCycleCallback implements InstanceRunTh
* @param runRecording 执行实例
* @return {@link PluginRpcRequestPacket}
*/
private PluginRpcRequestPacket buildPluginRpcRequestPacket(RunRecording runRecording, CallbackDto callbackDto,String serverKey) {
private PluginRpcRequestPacket buildPluginRpcRequestPacket(RunRecording runRecording, CallbackDto callbackDto, String serverKey) {
PluginRpcRequestPacket rpcRequestPacket = new PluginRpcRequestPacket();
rpcRequestPacket.setCallMethodEnum(CallMethodEnum.SYNCHRONIZE);
rpcRequestPacket.setJobName(serverKey);
......@@ -201,46 +124,51 @@ public class RpcCallbackInstanceRunTheLifeCycleCallback implements InstanceRunTh
String extendedConfiguration = runRecording.getExtendedConfiguration();
FlowExtendedConfiguration flowExtendedConfiguration = JSON.parseObject(extendedConfiguration, FlowExtendedConfiguration.class);
if(flowExtendedConfiguration == null){
if (flowExtendedConfiguration == null) {
return;
}
String rpcEndServerKey = flowExtendedConfiguration.getRpcEndServerKey();
if(StringUtils.isBlank(rpcEndServerKey)){
if (StringUtils.isBlank(rpcEndServerKey)) {
return;
}
String extendedParam = stringRedisTemplate.opsForValue().get(String.format(EXTENDED_PARAMS_S, runRecording.getRunId()));
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = JSON.parseObject(extendedParam, ScriptParamAndPlaceholderDto.class);
String publicParam = flowExtendedConfiguration.getPublicParam();
Map<String,String> callData = new HashMap<>(2);
Map<String, String> callData = new HashMap<>(2);
if(StringUtils.isNoneBlank(publicParam)){
if (StringUtils.isNoneBlank(publicParam)) {
Map<String, String> stringStringMap = (Map<String, String>) JSON.parse(publicParam);
if(CollectionUtil.isNotEmpty(stringStringMap)) {
if (CollectionUtil.isNotEmpty(stringStringMap)) {
callData.putAll(stringStringMap);
}
}
if(scriptParamAndPlaceholderDto != null){
if (scriptParamAndPlaceholderDto != null) {
Map<String, String> param = scriptParamAndPlaceholderDto.getParam();
Map<String, String> placeholder = scriptParamAndPlaceholderDto.getPlaceholder();
if(CollectionUtil.isNotEmpty(param)) {
if (CollectionUtil.isNotEmpty(param)) {
callData.putAll(param);
}
if(CollectionUtil.isNotEmpty(placeholder)) {
if (CollectionUtil.isNotEmpty(placeholder)) {
callData.putAll(placeholder);
}
}
CallbackDto callbackDto = new CallbackDto();
if (RunRecordingEnum.RUN_FLOW_SUCCESS.getCode().equals(runRecording.getFlowRunResult()) || RunRecordingEnum.RUN_FLOW_RE_SUCCESS.getCode().equals(runRecording.getFlowRunResult())) {
callbackDto.setRunRecordingStatus(true);
}
callbackDto.setData(JSON.toJSONString(callData));
PluginRpcRequestPacket rpcRequestPacket = buildPluginRpcRequestPacket(runRecording, callbackDto, rpcEndServerKey);
try {
PluginRpcResponsePacket call = taskServer.call(rpcRequestPacket);
log.info("回调{}通讯成功！",call);
}catch (Exception e) {
log.info("回调{}通讯成功！", call);
} catch (Exception e) {
log.info("回调通讯失败！");
}finally {
} finally {
stringRedisTemplate.delete(String.format(EXTENDED_PARAMS_S, runRecording.getRunId()));
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/strategy/impl/RunRecordingRunRedisRemoveTheLifeCycleCallback.java 0 → 100644
package com.byit.strategy.impl;
import com.byit.enums.RedisKeyNameEnum;
import com.byit.model.RunRecording;
import com.byit.strategy.InstanceRunTheLifeCycleCallback;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
/**
* 关于公共参数的redis数据删除
*
* @author huangfu
* @date 2020年12月14日11:55:37
*/
@Component
@Slf4j
public class RunRecordingRunRedisRemoveTheLifeCycleCallback implements InstanceRunTheLifeCycleCallback {
private final StringRedisTemplate stringRedisTemplate;
public RunRecordingRunRedisRemoveTheLifeCycleCallback(StringRedisTemplate stringRedisTemplate) {
this.stringRedisTemplate = stringRedisTemplate;
}
/**
* 后置处理器 清除实例执行中产生的各种redis信息
*
* @param runRecording 实例
*/
@Override
public void postProcessBeforeInitialization(RunRecording runRecording) {
//删除本次实例的公共参数
String runId = runRecording.getRunId();
String publicRedisName = String.format(RedisKeyNameEnum.REDIS_PUBLIC_PARAM_KEY.getKeyName(), runId);
stringRedisTemplate.delete(publicRedisName);
}
}
byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml
......@@ -229,8 +229,8 @@
select
<include refid="Base_Column_List" />
from run_recording
where UNIX_TIMESTAMP(start_time) * 1000 &gt;= #{loadScheduleCondition.startTime}
and UNIX_TIMESTAMP(start_time) * 1000 &lt;= #{loadScheduleCondition.endTime}
where trigger_time &gt;= #{loadScheduleCondition.startTime}
and trigger_time &lt;= #{loadScheduleCondition.endTime}
and schedule_type != 4
<if test="flowIds != null">
and flow_id in (
......
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/common/CallbackDto.java
......@@ -19,4 +19,5 @@ public class CallbackDto implements Serializable {
private String versionName;
private String runId;
private String data;
private Boolean runRecordingStatus = false;
}
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/plugin/FlowExtendedConfiguration.java
......@@ -23,4 +23,9 @@ public class FlowExtendedConfiguration {
* 公共参数
*/
public String publicParam;
/**
* 节点嵌入标记 a:b:c
*/
private String flowEmbedLogo;
}
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/plugin/PluginFlowConfig.java
......@@ -20,11 +20,6 @@ import java.util.Map;
public class PluginFlowConfig {
/**
* 公共参数
*/
private Map<String,String> publicParam;
/**
* 当前工作流版本的告警的时机(0 不告警， 1 完成时告警， 2 失败时告警， 3 成功时告警)
*/
private String alarmlAction;
......
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