Commit 8db192ba by guominglei

修改重跑时运行实例状态

parent 80d1a0f5
......@@ -4,10 +4,7 @@ import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.byit.dto.api.DeleteDto;
import com.byit.dto.plugin.*;
import com.byit.enums.DagCheckEnum;
import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.NodePropertyEnum;
import com.byit.enums.ScheduleTypeEnum;
import com.byit.enums.*;
import com.byit.enums.plugin.PluginNodeTypeEnum;
import com.byit.job.utils.CronExpression;
import com.byit.job.utils.CurrentUserUtils;
......@@ -513,7 +510,9 @@ public class ApiFlowServiceImpl implements ApiFlowService {
}
RunRecording newRunRecording = buildRunRecording(runRecording, triggerTime, reRunId, flow, userName);
runRecording.setFlowStatus(RunRecordingEnum.FLOW_STATUS_RUN_ING.getCode());
runRecording.setFlowRunResult(null);
runRecordingMapper.updateState(runRecording);
//生成新的工作流实例
runRecordingMapper.saveRunRecording(newRunRecording);
jobTaskMapper.saveJobTasks(jobTaskList);
......
......@@ -205,4 +205,6 @@ public interface RunRecordingMapper {
* @return
*/
RunRecording findNewStatus(Integer flowId);
int updateState(RunRecording runRecording);
}
\ No newline at end of file
......@@ -90,6 +90,53 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
List<RunRecording> runRecordingList = runRecordingMapper.findByPreTime(new Date(preHourTime), flow.getFlowId());
//判断是否存在过去一个小时的和未完成的工作流实例
if(null != runRecordingList && runRecordingList.size() > 0){
//处理未完成和过去一个小时的工作流实例
buildUnFinish(runRecordingList, flowStatusSnapshootList, flowStatusSnapshoot, flow);
}else {
//如果不存在实例,获取上一个小时的运行快照保存
LocalDateTime preHourLocalDateTime = localDateTime.minusHours(1);
String preHourDate = preHourLocalDateTime.format(DATE_FORMATTER);
String preHourHour = String.valueOf(preHourLocalDateTime.getHour());
List<FlowStatusSnapshoot> preHourFlowStatusSnapshootList = flowStatusSnapshootMapper.findByDateAndHourAndFlowId(preHourDate, preHourHour, flow.getFlowId());
if (null != preHourFlowStatusSnapshootList || preHourFlowStatusSnapshootList.size() > 0){
flowStatusSnapshootList.addAll(preHourFlowStatusSnapshootList);
}else {
//如果上个小时也没有,就设置为未运行
flowStatusSnapshoot.setFlowStatus(1);
flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount());
flowStatusSnapshootList.add(flowStatusSnapshoot);
}
}
}
}else {
//如果不存在运行实例
flowStatusSnapshoot.setFlowStatus(1);
flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount());
flowStatusSnapshootList.add(flowStatusSnapshoot);
}
});
if (null != flowStatusSnapshootList && flowStatusSnapshootList.size() > 0){
flowStatusSnapshootMapper.saveList(flowStatusSnapshootList);
}
//获取过期的时间戳
Long outTime = hourDateTime.minusDays(snapshootDate).toInstant(ZoneOffset.of("+8")).toEpochMilli(); //转换时间戳;
//删除超过期限的快照
flowStatusSnapshootMapper.deleteOutSnapShoot(outTime);
}
}
return (60 - LocalTime.now().getMinute()) * 60 * 1000L;
}
/**
* 功能描述 处理过去一小时和未完成的工作流的快照
* @author gml
* @date 2020-05-08 10:20
* @param runRecordingList
* @param flowStatusSnapshootList
* @param flowStatusSnapshoot
* @return void
*/
private void buildUnFinish(List<RunRecording> runRecordingList, List<FlowStatusSnapshoot> flowStatusSnapshootList, FlowStatusSnapshoot flowStatusSnapshoot, Flow flow){
//如果存在,记录快照
runRecordingList.forEach(runRecording -> {
FlowStatusSnapshoot hourFlowStatusSnapshoot = new FlowStatusSnapshoot();
......@@ -134,39 +181,6 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
}
flowStatusSnapshootList.add(hourFlowStatusSnapshoot);
});
}else {
//如果不存在实例,获取上一个小时的运行快照保存
LocalDateTime preHourLocalDateTime = localDateTime.minusHours(1);
String preHourDate = preHourLocalDateTime.format(DATE_FORMATTER);
String preHourHour = String.valueOf(preHourLocalDateTime.getHour());
List<FlowStatusSnapshoot> preHourFlowStatusSnapshootList = flowStatusSnapshootMapper.findByDateAndHourAndFlowId(preHourDate, preHourHour, flow.getFlowId());
if (null != preHourFlowStatusSnapshootList || preHourFlowStatusSnapshootList.size() > 0){
flowStatusSnapshootList.addAll(preHourFlowStatusSnapshootList);
}else {
//如果上个小时也没有,就设置为未运行
flowStatusSnapshoot.setFlowStatus(1);
flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount());
flowStatusSnapshootList.add(flowStatusSnapshoot);
}
}
}
}else {
//如果不存在运行实例
flowStatusSnapshoot.setFlowStatus(1);
flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount());
flowStatusSnapshootList.add(flowStatusSnapshoot);
}
});
if (null != flowStatusSnapshootList && flowStatusSnapshootList.size() > 0){
flowStatusSnapshootMapper.saveList(flowStatusSnapshootList);
}
//获取过期的时间戳
Long outTime = hourDateTime.minusDays(snapshootDate).toInstant(ZoneOffset.of("+8")).toEpochMilli(); //转换时间戳;
//删除超过期限的快照
flowStatusSnapshootMapper.deleteOutSnapShoot(outTime);
}
}
return (60 - LocalTime.now().getMinute()) * 60 * 1000L;
}
......
......@@ -588,5 +588,9 @@
where run_id = #{runId,jdbcType=VARCHAR}
and flow_name = #{flowName,jdbcType=VARCHAR}
</update>
<update id="updateState" parameterType="com.byit.model.RunRecording">
update run_recording set flow_status = #{flowStatus},flow_run_result = #{flowRunResult}
where recording_id = #{recordingId,jdbcType=INTEGER}
</update>
</mapper>
\ No newline at end of file
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