Skip to content

B
byit-myth-job
Commit 961480e6 by guominglei
Options

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

parents 8edbcee1 8c195b02
Showing with 166 additions and 52 deletions
.gitignore 0 → 100644
*.class
*.iml
compiler.xml
encodings.xml
Maven_*.xml
misc.xml
modules.xml
Project_Default.xml
vcs.xml
workspace.xml
target
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/conf/MythJobScheduler.java
......@@ -32,7 +32,8 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
*/
@Override
public void destroy() throws Exception {
this.jobScheduleHelper.doStop();
this.logScanHelper.doStop();
}
/**
......@@ -42,8 +43,8 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
@Override
public void afterPropertiesSet() throws Exception {
//启用扫描线程
jobScheduleHelper.start();
logScanHelper.start();
this.jobScheduleHelper.start();
this.logScanHelper.start();
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/enums/RunRecordingEnum.java 0 → 100644
package com.byit.enums;
/**
* @description: 运行记录的枚举类
* @author huangfu
*/
public enum RunRecordingEnum {
RUN_FLOW_SUCCESS("1","工作流运行成功")
,RUN_FLOW_FAILURE("2","工作流运行失败")
,RUN_FLOW_RE_SUCCESS("3","补批成功")
,RUN_FLOW_RE_FAILURE("4","补批失败")
,RUN_FLOW_KILL("5","工作流进程被杀死")
,FLOW_STATUS_NOT_RUN("1","工作流未开始")
,FLOW_STATUS_RUN_ING("2","工作流运行中")
,FLOW_STATUS_IS_STOP("3","工作流被暂停")
,FLOW_STATUS_IS_END("4","工作流已经完结")
;
private String code;
private String message;
RunRecordingEnum(String code, String message) {
this.code = code;
this.message = message;
}
public String getCode() {
return code;
}
public String getMessage() {
return message;
}}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/JobScheduleHelper.java
......@@ -320,6 +320,43 @@ public class JobScheduleHelper{
scheduleThread.setDaemon(true);
scheduleThread.start();
}
public void doStop(){
this.jobInfoThreadToStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (jobInfoThread.getState() != Thread.State.TERMINATED) {
jobInfoThread.interrupt();
try {
jobInfoThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【任务节点节点扫描线程被注销】-----------------------");
this.scheduleThreadToStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (scheduleThread.getState() != Thread.State.TERMINATED) {
scheduleThread.interrupt();
try {
scheduleThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【排期表扫描线程被注销】-----------------------");
}
@Autowired
public void setDataSource(DataSource dataSource) {
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/LogScanHelper.java
package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.enums.RunRecordingEnum;
import com.byit.model.JobTaskRunLog;
import com.byit.model.RunRecording;
import com.byit.service.JobTaskRunLogService;
......@@ -109,7 +110,43 @@ public class LogScanHelper {
e.printStackTrace();
}finally {
//释放资源
freedResource(conn,preparedStatement,connAutoCommit,virtualNodeScanIsStop);
if(conn != null){
try {
conn.commit();
} catch (SQLException e) {
if(!virtualNodeScanIsStop){
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if(conn != null){
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if(!virtualNodeScanIsStop){
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if(preparedStatement != null){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!notAlarmedNodeScanIsStop){
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
conn.close();
} catch (SQLException e) {
if(!virtualNodeScanIsStop){
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
}
......@@ -135,7 +172,6 @@ public class LogScanHelper {
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'log_node_callback_lock' FOR UPDATE ");
preparedStatement.execute();
System.out.println("--------------火球锁成功------------------" );
//查询的是失败的或者是已经结束的节点（完成的）
List<JobTaskRunLog> jobTaskRunLogEndOrFailureNode = jobTaskRunLogService.findJobTaskRunLogEndOrFailureNode( );
if(CollectionUtil.isNotEmpty(jobTaskRunLogEndOrFailureNode)){
......@@ -162,7 +198,7 @@ public class LogScanHelper {
successNodes.forEach(successNode ->{
String runId = successNode.getRunId( );
Integer flowId = successNode.getFlowId( );
runRecordingService.updateRunRecordingByFlowIdAndRunId(RunRecording.builder().runId(runId).flowId(flowId).flowRunResult("1").flowStatus("4").build());
runRecordingService.updateRunRecordingByFlowIdAndRunId(RunRecording.builder().runId(runId).flowId(flowId).flowRunResult(RunRecordingEnum.RUN_FLOW_SUCCESS.getCode()).flowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode()).build());
successNode.setAlertEnd("1");
jobTaskRunLogService.updateJobTaskRunLog(successNode);
});
......@@ -183,7 +219,43 @@ public class LogScanHelper {
}catch (Exception e){
e.printStackTrace();
}finally {
freedResource(conn,preparedStatement,connAutoCommit,notAlarmedNodeScanIsStop);
if(conn != null){
try {
conn.commit();
} catch (SQLException e) {
if(!notAlarmedNodeScanIsStop){
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if(conn != null){
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if(!notAlarmedNodeScanIsStop){
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if(preparedStatement != null){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!notAlarmedNodeScanIsStop){
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
conn.close();
} catch (SQLException e) {
if(!notAlarmedNodeScanIsStop){
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
}
});
......@@ -192,51 +264,6 @@ public class LogScanHelper {
notAlarmedNodeScanThread.start();
}
/**
* 释放资源
* @param conn 数据库链接
* @param preparedStatement 执行器
* @param connAutoCommit 原来的提交状态
*/
private void freedResource(Connection conn,PreparedStatement preparedStatement,boolean connAutoCommit,boolean isStop){
if(conn != null){
try {
conn.commit();
} catch (SQLException e) {
if(!isStop){
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
assert conn != null;
conn.setAutoCommit(connAutoCommit);
} catch (SQLException e) {
if(!isStop){
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if(preparedStatement != null){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!isStop){
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
conn.close();
} catch (SQLException e) {
if(!isStop){
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
public void doStop(){
this.virtualNodeScanIsStop = true;
......@@ -254,6 +281,7 @@ public class LogScanHelper {
e.printStackTrace( );
}
}
log.warn("---------------【虚节点扫描线程被注销】-----------------------");
this.notAlarmedNodeScanIsStop = true;
try {
......@@ -269,6 +297,8 @@ public class LogScanHelper {
e.printStackTrace( );
}
}
log.warn("---------------【日志扫描线程被注销】-----------------------");
}
......
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