Commit df05f326 by huangfusuper

合并扫描工作流操作为一个事务

parent c5fdb7bf
...@@ -18,6 +18,6 @@ public interface RunNodeServer { ...@@ -18,6 +18,6 @@ public interface RunNodeServer {
* @param flow * @param flow
* @param nodes * @param nodes
*/ */
void saveRunRecAndTask(Flow flow, List<Node> nodes); void saveRunRec(Flow flow, List<Node> nodes);
} }
package com.byit.service.impl; package com.byit.service.impl;
import com.byit.enums.FlowPropertyEnum;
import com.byit.job.utils.CronExpression;
import com.byit.model.Flow; import com.byit.model.Flow;
import com.byit.model.JobTask; import com.byit.model.JobTask;
import com.byit.model.Node; import com.byit.model.Node;
import com.byit.model.RunRecording; import com.byit.model.RunRecording;
import com.byit.service.FlowService;
import com.byit.service.JobTaskService; import com.byit.service.JobTaskService;
import com.byit.service.RunNodeServer; import com.byit.service.RunNodeServer;
import com.byit.service.RunRecordingService; import com.byit.service.RunRecordingService;
...@@ -13,7 +16,9 @@ import org.springframework.stereotype.Service; ...@@ -13,7 +16,9 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import java.text.ParseException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Date;
import java.util.List; import java.util.List;
import java.util.UUID; import java.util.UUID;
...@@ -26,19 +31,22 @@ import java.util.UUID; ...@@ -26,19 +31,22 @@ import java.util.UUID;
public class RunNodeServiceImpl implements RunNodeServer { public class RunNodeServiceImpl implements RunNodeServer {
private final RunRecordingService runRecordingService; private final RunRecordingService runRecordingService;
private final JobTaskService jobTaskService; private final JobTaskService jobTaskService;
private final FlowService flowService;
public RunNodeServiceImpl(RunRecordingService runRecordingService, JobTaskService jobTaskService) { public RunNodeServiceImpl(RunRecordingService runRecordingService, JobTaskService jobTaskService,
FlowService flowService) {
this.runRecordingService = runRecordingService; this.runRecordingService = runRecordingService;
this.jobTaskService = jobTaskService; this.jobTaskService = jobTaskService;
this.flowService = flowService;
} }
/** /**
* 保存到运行记录一份 将节点保存到job_task表 * 保存到运行记录一份 将节点保存到job_task表
* @param flow * @param flow 工作流
* @param nodes * @param nodes 节点
*/ */
@Override @Override
public void saveRunRecAndTask(Flow flow, List<Node> nodes) { public void saveRunRec(Flow flow, List<Node> nodes) {
log.info("---------saveRunRecAndTask start------【保存工作流:{}和节点:{}】-----------------------",flow,nodes); log.info("---------saveRunRecAndTask start------【保存工作流:{}和节点:{}】-----------------------",flow,nodes);
String runId = UUID.randomUUID().toString().replace("-",""); String runId = UUID.randomUUID().toString().replace("-","");
log.info("-------------【开始保存运行记录runId为:{}】------------------",runId); log.info("-------------【开始保存运行记录runId为:{}】------------------",runId);
...@@ -65,6 +73,34 @@ public class RunNodeServiceImpl implements RunNodeServer { ...@@ -65,6 +73,34 @@ public class RunNodeServiceImpl implements RunNodeServer {
jobTasks.add(jobTask); jobTasks.add(jobTask);
}); });
jobTaskService.saveJobTasks(jobTasks); jobTaskService.saveJobTasks(jobTasks);
log.info("-------saveRunRecAndTask end-----------【运行结束】-----------------"); log.info("-------------【开始修改工作流{}的下次运行时间,以及各种状态】---------------",flow);
updateFlow(flow);
log.info("-------saveRunRecAndTaskAndUpdate end-----------【运行结束】-----------------");
}
/**
* 修改工作流的信息
* @param flow 工作流
*/
private void updateFlow(Flow flow){
if (flow.getRemainingCount()>0) {
flow.setRemainingCount(flow.getRemainingCount()-1);
}
//获取cron表达式
String flowCron = flow.getFlowCron();
//设置下一周期的时间
Date nextValidTime = null;
try {
nextValidTime = new CronExpression(flowCron).getNextValidTimeAfter(new Date(flow.getTriggerNextTime()));
} catch (ParseException e) {
e.printStackTrace();
}
flow.setTriggerNextTime(nextValidTime!=null?nextValidTime.getTime():999999999999L);
flow.setScanMark(FlowPropertyEnum.NOT_SCAN.getCode());
flowService.updateByIdSelective(flow);
} }
} }
...@@ -50,32 +50,17 @@ public class FlowThreadRunHelper extends BaseThreadRunHelper { ...@@ -50,32 +50,17 @@ public class FlowThreadRunHelper extends BaseThreadRunHelper {
if (CollectionUtil.isNotEmpty(halfAnHourFlow)) { if (CollectionUtil.isNotEmpty(halfAnHourFlow)) {
for(Flow flow : halfAnHourFlow ){ for(Flow flow : halfAnHourFlow ){
log.debug("-----------------【工作流{}的执行次数不等于0,放行】-------------------------",flow.getFlowName()); log.debug("-----------------【工作流{}的执行次数不等于0,放行】-------------------------",flow.getFlowName());
if (runRecordingService.findRunRecordingIsRunning(flow.getFlowId())) { /*if (runRecordingService.findRunRecordingIsRunning(flow.getFlowId())) {
//TODO 是否可以使用事件通知机制,由完结的工作流实例通知通过流可以运行了 //TODO 是否可以使用事件通知机制,由完结的工作流实例通知通过流可以运行了
log.info("-----工作流{},有正在运行中的实例,跳过等待------",flow); log.info("-----工作流{},有正在运行中的实例,跳过等待------",flow)
continue; continue;
} }*/
//String versionName = flow.getVersionName()
Integer flowId = flow.getFlowId(); Integer flowId = flow.getFlowId();
//根据工作流查询工作流下所有的节点 //根据工作流查询工作流下所有的节点
List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(flowId); List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(flowId);
if(CollectionUtil.isNotEmpty(nodeByFlowIdAndVersionName)){ if(CollectionUtil.isNotEmpty(nodeByFlowIdAndVersionName)){
//保存到运行记录表和任务表 //保存到运行记录表和任务表
runNodeServer.saveRunRecAndTask(flow,nodeByFlowIdAndVersionName); runNodeServer.saveRunRec(flow,nodeByFlowIdAndVersionName);
if (flow.getRemainingCount()>0) {
flow.setRemainingCount(flow.getRemainingCount()-1);
}
//获取cron表达式
String flowCron = flow.getFlowCron();
//设置下一周期的时间
Date nextValidTime = null;
try {
nextValidTime = new CronExpression(flowCron).getNextValidTimeAfter(new Date(flow.getTriggerNextTime()));
} catch (ParseException e) {
e.printStackTrace();
}
flow.setTriggerNextTime(nextValidTime!=null?nextValidTime.getTime():999999999999L);
flowService.updateByIdSelective(flow);
} }
} }
}else{ }else{
......
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