Skip to content

B
byit-myth-job
Commit 9731d298 by huangfusuper
Options

运行实例验证是否完成流程修改

parent 6d59c6e9
Showing with 162 additions and 4 deletions
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/RunRecordingMapper.java
......@@ -18,6 +18,11 @@ public interface RunRecordingMapper {
*/
List<RunRecording> findRunRecordingByEndAndNotIsAlarm();
/**
* 查询运行中的运行实例
* @return
*/
List<RunRecording> findRunningRunRecording();
/**
* 这个方法是会根据任务流的id和运行标识找到唯一对应的一个任务流，这个任务流就是一个虚拟节点
* @param flowId
* @param runId
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/RunRecording.java
......@@ -132,6 +132,11 @@ public class RunRecording implements Serializable {
*/
@ApiModelProperty("快速失败标识 0 不快速失败 1 快速失败")
private String failFast;
/**
* 工作流节点数量
*/
@ApiModelProperty("工作流节点数量")
private Integer flowNodeCount;
/**
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/RunRecordingService.java
......@@ -14,6 +14,13 @@ public interface RunRecordingService {
* @return
*/
List<RunRecording> findRunRecordingByEndAndNotIsAlarm();
/**
* 查询运行中的运行实例
* @return
*/
List<RunRecording> findRunningRunRecording();
/**
* 这个方法是会根据任务流的id和运行标识找到唯一对应的一个任务流，这个任务流就是一个虚拟节点
* @param flowId
......@@ -54,4 +61,6 @@ public interface RunRecordingService {
* @return
*/
int deleteById(Integer recordingId);
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/RunRecordingServiceImpl.java
......@@ -29,6 +29,11 @@ public class RunRecordingServiceImpl implements RunRecordingService {
}
@Override
public List<RunRecording> findRunningRunRecording() {
return null;
}
@Override
public RunRecording findRunRecordingByFlowIdAndRunId(Integer flowId, String runId) {
return runRecordingMapper.findRunRecordingByFlowIdAndRunId(flowId,runId);
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/LogScanHelper.java
......@@ -73,6 +73,7 @@ public class LogScanHelper {
public void start(){
virtualNodeScanMethod();
//TODO 不推荐使用，详情见方法介绍
notAlarmedNodeScanMethod();
errorNodeScan();
}
......@@ -173,6 +174,9 @@ public class LogScanHelper {
}
/**
* @deprecated 废除原因：不需要这条线程去扫描结束节点，只需要扫描对应工作流的节点数目是否全部匹配即可
* 这一操作需要放置到扫描运行实例的线程里面
* @deprecatedDate 2020年3月6日10:36:19
* 未告警的节点扫描
*/
private void notAlarmedNodeScanMethod(){
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/RunRecordingScanHelper.java
......@@ -69,9 +69,22 @@ public class RunRecordingScanHelper {
private final JobTaskRunLogService jobTaskRunLogService;
private final RunRecordingService runRecordingService;
private final FileSystem fileSystem;
/**
* 查询完结且没有告警的节点线程是否停止
*/
private volatile boolean runRecordingThreadStop = false;
/**
* 查询完结且没有告警的节点线程
*/
private Thread runRecordingThread;
/**
* 判断工作流是否完结是否停止
*/
private volatile boolean judgeFlowIsEndStop = false;
/**
* 判断工作流是否完结的线程
*/
private Thread judgeFlowIsEndThread;
@Autowired
public RunRecordingScanHelper(EmailAlarmService emailAlarmService, JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService, FileSystem fileSystem) {
......@@ -82,9 +95,99 @@ public class RunRecordingScanHelper {
}
public void start() {
scanRunRecThread();
judgeFlowIsEndThread();
}
/**
* 判断工作流是否完结
*/
public void judgeFlowIsEndThread(){
judgeFlowIsEndThread = new Thread(() ->{
dateAligned(5000);
log.info("----------------------【com.byit.thread.RunRecordingScanThread#judgeFlowIsEnd】-------------------");
while (judgeFlowIsEndStop){
boolean isSleep = false;
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
try {
conn = dataSource.getConnection();
connAutoCommit = conn.getAutoCommit();
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'judge_flow_end_lock' FOR UPDATE ");
preparedStatement.execute();
//进行操作
}catch (Exception e){
if (!judgeFlowIsEndStop) {
e.printStackTrace();
}
}finally {
if (conn != null) {
try {
conn.commit();
} catch (SQLException e) {
if (!judgeFlowIsEndStop) {
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if (conn != null) {
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if (!judgeFlowIsEndStop) {
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if (preparedStatement != null) {
try {
preparedStatement.close();
} catch (SQLException e) {
if (!judgeFlowIsEndStop) {
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
if(conn!=null){
conn.close();
}
} catch (SQLException e) {
if (!judgeFlowIsEndStop) {
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
if(isSleep){
dateAligned(10000);
}
}
});
}
/**
* 扫描未完成的工作流
*/
private void judgeFlowIsEnd(){
//查询未完结的工作流实例
}
/**
* 查询完结且没有告警的节点线程构建
*/
private void scanRunRecThread(){
runRecordingThread = new Thread(() -> {
dateAligned(5000);
log.info("--------------------【com.byit.thread.RunRecordingScanThread#start】init success----------------------");
log.info("--------------------【com.byit.thread.RunRecordingScanThread#scanRunRecThread】init success----------------------");
while (!runRecordingThreadStop) {
boolean isSleep = false;
Connection conn = null;
......@@ -135,7 +238,9 @@ public class RunRecordingScanHelper {
}
try {
if(conn!=null){
conn.close();
}
} catch (SQLException e) {
if (!runRecordingThreadStop) {
log.error("--------------------【关闭链接出错】---------------------");
......@@ -153,6 +258,9 @@ public class RunRecordingScanHelper {
runRecordingThread.start();
}
/**
* 查询完结 但是没有告警的节点
*/
private void scanRunRec() {
//查询完结且未告警的工作流信息
List<RunRecording> runRecordingByEndAndNotIsAlarm = runRecordingService.findRunRecordingByEndAndNotIsAlarm();
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/FlowMapper.xml
......@@ -27,7 +27,6 @@
<result column="is_update" jdbcType="CHAR" property="isUpdate" />
</resultMap>
<sql id="Base_Column_List">
<!-- generated @mbg.generated date: 2019-12-31 -->
flow_id, alarm_email, exec_type, flow_cron, flow_desc, flow_name, flow_node_count,
flow_timeout, is_inner, alarml_action, priority, trigger_next_time, workspace_id,
author, add_time, start_up, principal, version_name, repeat_count, remaining_count,
......@@ -150,6 +149,7 @@
<if test="isUpdate != null">
is_update,
</if>
</trim>
<trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="flowId != null">
......@@ -218,6 +218,7 @@
<if test="isUpdate != null">
#{isUpdate,jdbcType=CHAR},
</if>
</trim>
</insert>
<update id="updateByIdSelective" parameterType="com.byit.model.Flow">
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml
......@@ -21,11 +21,12 @@
<result column="is_alarm" jdbcType="CHAR" property="isAlarm" />
<result column="is_inner" jdbcType="CHAR" property="isInner" />
<result column="fail_fast" jdbcType="CHAR" property="failFast" />
<result column="flow_node_count" jdbcType="VARCHAR" property="flowNodeCount" />
</resultMap>
<sql id="Base_Column_List">
recording_id, run_id, alarm_email, dispatch_ip, flow_name, flow_run_result, flow_status,
flow_timeout, flow_version_name, alarml_action, priority, trigger_time, principal,
flow_id, start_time, end_time, is_alarm,is_inner,fail_fast
flow_id, start_time, end_time, is_alarm,is_inner,fail_fast, flow_node_count
</sql>
<!--查询已完结 没有告警的-->
......@@ -36,6 +37,13 @@
where flow_status ='4' and is_alarm = '1'
</select>
<select id="findRunningRunRecording" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
where flow_status = '2'
</select>
<select id="findRunRecordingByFlowIdAndRunId" resultMap="BaseResultMap" >
select
<include refid="Base_Column_List" />
......@@ -169,6 +177,9 @@
<if test="failFast != null">
fail_fast,
</if>
<if test="flowNodeCount != null">
flow_node_count,
</if>
</trim>
<trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="recordingId != null">
......@@ -228,6 +239,9 @@
<if test="failFast != null">
#{failFast,jdbcType=CHAR},
</if>
<if test="flowNodeCount != null">
#{flowNodeCount,jdbcType=INTEGER},
</if>
</trim>
</insert>
<update id="updateRunRecordingById" parameterType="com.byit.model.RunRecording">
......@@ -287,6 +301,9 @@
<if test="failFast != null">
fail_fast = #{failFast,jdbcType=CHAR},
</if>
<if test="flowNodeCount != null">
flow_node_count = #{flowNodeCount,jdbcType=INTEGER},
</if>
</set>
where recording_id = #{recordingId,jdbcType=INTEGER}
</update>
......@@ -348,6 +365,9 @@
<if test="failFast != null">
fail_fast = #{failFast,jdbcType=CHAR},
</if>
<if test="flowNodeCount != null">
flow_node_count = #{flowNodeCount,jdbcType=INTEGER},
</if>
</set>
where flow_id = #{flowId,jdbcType=INTEGER} and run_id = #{runId,jdbcType=VARCHAR}
</update>
......
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