Skip to content

B
byit-myth-job
Commit cdf79dd0 by huangfusuper
Options

Merge remote-tracking branch 'origin/developer' into developer

parents 2a6faba5 8d767aee
Showing with 76 additions and 38 deletions
byit-myth-admin/src/main/java/com/byit/service/impl/ApiFlowServiceImpl.java
......@@ -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);
......@@ -592,6 +591,12 @@ 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);
//将工作流下的节点改为运行中
jobTaskRunLogMapper.updateByFlowIdAndRunId(flow.getFlowId(), runInfo.getRunId());
//生成新的工作流实例
runRecordingMapper.saveRunRecording(newRunRecording);
jobTaskMapper.saveJobTasks(jobTaskList);
......@@ -617,6 +622,9 @@ public class ApiFlowServiceImpl implements ApiFlowService {
if (dependNodeIdList != null && dependNodeIdList.size() > 0){
jobTask.setNodeDepend(Joiner.on(",").join(dependNodeIdList));
}
//将节点对应的日志改为运行中
jobTaskRunLog.setRunCode(NodeRunStatusPropertyEnum.RUN_ING.getCode());
jobTaskRunLogMapper.updateJobTaskRunLog(jobTaskRunLog);
jobTaskList.add(jobTask);
//查询依赖于当前节点的下级节点
List<Integer> childNodeIdList = nodeDependencyMapper.findSubNodeList(childNodeId);
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskRunLogMapper.java
......@@ -152,4 +152,6 @@ public interface JobTaskRunLogMapper {
List<JobTaskRunLog> findJavaTaskByTaskName(@Param("nodeIdList")List<Integer> nodeIdList, @Param("startTime")Date startTime, @Param("endTime")Date endTime);
JobTaskRunLog findNewJavaTaskByJobName(Integer javaTaskId);
int updateByFlowIdAndRunId(@Param("flowId")Integer flowId, @Param("runId")String runId);
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/RunRecordingMapper.java
......@@ -205,4 +205,6 @@ public interface RunRecordingMapper {
* @return
*/
RunRecording findNewStatus(Integer flowId);
int updateState(RunRecording runRecording);
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/StatusSnapshootThreadRunHelper.java
......@@ -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;
}
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml
......@@ -636,4 +636,8 @@
</set>
where log_id = #{logId,jdbcType=INTEGER}
</update>
<update id="updateByFlowIdAndRunId">
update job_task_run_log set run_code = '0'
where flow_id = #{flowId} and run_id = #{runId}
</update>
</mapper>
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml
......@@ -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