Skip to content

B
byit-myth-job
Commit a1aadf5f by huangfusuper
Options

删除废弃代码

parent 6dbfca7c
Showing with 0 additions and 1763 deletions
byit-myth-admin/src/main/java/com/byit/controller/JobController.java
...@@ -8,7 +8,6 @@ import com.byit.model.JobTask; ...@@ -8,7 +8,6 @@ import com.byit.model.JobTask;
import com.byit.service.JobTaskService; import com.byit.service.JobTaskService;
import com.byit.service.RunScriptService; import com.byit.service.RunScriptService;
import com.byit.thread.LogCallbackThread; import com.byit.thread.LogCallbackThread;
import com.byit.thread.RunRecordingScanHelper;
import com.byit.util.SourceObj2TargetObjUtil; import com.byit.util.SourceObj2TargetObjUtil;
import com.byit.utils.ValidationUtil; import com.byit.utils.ValidationUtil;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/conf/MythJobScheduler.java
...@@ -21,15 +21,6 @@ import java.util.Map; ...@@ -21,15 +21,6 @@ import java.util.Map;
@Slf4j @Slf4j
public class MythJobScheduler implements InitializingBean, DisposableBean { public class MythJobScheduler implements InitializingBean, DisposableBean {
/**
* private final JobScheduleHelper jobScheduleHelper
* private final EmailScanHelper emailScanHelper
* private final FlowScanHelper flowScanHelper
* private final LogScanHelper logScanHelper
* private final RunRecordingScanHelper runRecordingScanHelper
*/
private final ApplicationContext applicationContext; private final ApplicationContext applicationContext;
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/EmailScanHelper.java deleted 100644 → 0
package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.model.EmailAlarm;
import com.byit.model.vo.EmailAlarmVo;
import com.byit.service.EmailAlarmService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* @program: com.byit.thread.EmailScanHelper
* @description: 对于邮箱的扫描
* @author: huangfu
* @date: 2020年1月7日10:44:01
**/
@Component
@Slf4j
@Deprecated
public class EmailScanHelper {
private final DataSource dataSource;
private final EmailAlarmService emailAlarmService;
private volatile boolean emailThreadIsStop = false;
private Thread emailThread;
public EmailScanHelper(DataSource dataSource, EmailAlarmService emailAlarmService) {
this.dataSource = dataSource;
this.emailAlarmService = emailAlarmService;
}
public void start(){
startScanNotSentEmailFlow();
}
public void startScanNotSentEmailFlow(){
emailThread = new Thread(() ->{
dateAligned(5000);
log.info("--------------【com.byit.thread.EmailScanHelper.startScanNotSentEmailFlow】init success---------------");
while (!emailThreadIsStop){
//是否需要睡眠
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 = 'emali_alarm_lock' FOR UPDATE ");
preparedStatement.execute();
//查询未告警的邮箱
List<EmailAlarm> emailAlarmByAlarmResults = emailAlarmService.findEmailAlarmByAlarmResult();
if(CollectionUtil.isNotEmpty(emailAlarmByAlarmResults)){
emailAlarmByAlarmResults.forEach(emailAlarm ->{
EmailAlarmVo emailAlarmVo = new EmailAlarmVo();
BeanUtils.copyProperties(emailAlarm,emailAlarmVo);
emailAlarmService.sendEmail(emailAlarmVo);
});
}else{
isSleep = true;
}
}catch (Exception e){
e.printStackTrace();
}finally {
//释放资源
if(conn != null){
try {
conn.commit();
} catch (SQLException e) {
if(!emailThreadIsStop){
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if(conn != null){
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if(!emailThreadIsStop){
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if(preparedStatement != null){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!emailThreadIsStop){
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
conn.close();
} catch (SQLException e) {
if(!emailThreadIsStop){
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
if(isSleep){
dateAligned(20000);
}
}
});
emailThread.setDaemon(true);
emailThread.setName("myth-job#【EmailScanHelper】#startScanNotSentEmailFlow");
emailThread.start();
}
public void doStop(){
this.emailThreadIsStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (emailThread.getState() != Thread.State.TERMINATED) {
emailThread.interrupt();
try {
emailThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【日志扫描线程被注销】-----------------------");
}
/**
* 对齐时钟。整秒运行
*/
private void dateAligned(long waitTime){
try {
TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis()%1000);
} catch (InterruptedException e) {
log.warn("----------------【线程被中断】-----------------------");
}
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/FlowScanHelper.java deleted 100644 → 0
package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.job.utils.CronExpression;
import com.byit.model.Flow;
import com.byit.model.Node;
import com.byit.service.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.ParseException;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* 扫描任务流表的线程
* 目的：扫描即将要执行的任务流表 半个小时
* @author huangfu
*/
@Component
@Slf4j
@Deprecated
public class FlowScanHelper {
private final Long PRE_TEST_TIME = System.currentTimeMillis()+TimeUnit.HOURS.toMillis(30);
private final FlowService flowService;
private final DataSource dataSource;
private final NodeService nodeService;
private final RunNodeServer runNodeServer;
private Thread flowThread;
private volatile boolean flowThreadIsStop = false;
public FlowScanHelper(FlowService flowService, DataSource dataSource, NodeService nodeService, RunNodeServer runNodeServer) {
this.flowService = flowService;
this.dataSource = dataSource;
this.nodeService = nodeService;
this.runNodeServer = runNodeServer;
}
public void start(){
flowThreadStart();
}
/**
* 运行扫描工作流的线程
*/
private void flowThreadStart(){
flowThread = new Thread(() ->{
dateAligned(5000);
log.info("--------------【com.byit.thread.FlowScanHelper#flowThreadStart】init success---------------");
while (!flowThreadIsStop){
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
//是否需要睡眠
boolean isSleep = false;
try {
/**
* 添加行锁
*/
conn = dataSource.getConnection();
connAutoCommit = conn.getAutoCommit();
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'flow_lock' FOR UPDATE ");
preparedStatement.execute();
//这个查询时有一个条件是 剩余次数不等于0也就是说 等于0的就查询不出来
List<Flow> halfAnHourFlow = flowService.findHalfAnHourFlow(PRE_TEST_TIME);
if (CollectionUtil.isNotEmpty(halfAnHourFlow)) {
for(Flow flow : halfAnHourFlow ){
log.debug("-----------------【工作流{}的执行次数大于0，放行】-------------------------",flow.getFlowName());
//String versionName = flow.getVersionName()
Integer flowId = flow.getFlowId();
//根据工作流查询工作流下所有的节点
List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(flowId);
if(CollectionUtil.isNotEmpty(nodeByFlowIdAndVersionName)){
//保存到运行记录表和任务表
runNodeServer.saveRunRec(flow,nodeByFlowIdAndVersionName);
if (flow.getRemainingCount()>0) {
flow.setRemainingCount(flow.getRemainingCount()-1);
}
//获取cron表达式
String flowCron = flow.getFlowCron();
//设置下一周期的时间
Date nextValidTime = new CronExpression(flowCron).getNextValidTimeAfter(new Date(flow.getTriggerNextTime()));
flow.setTriggerNextTime(nextValidTime.getTime());
flowService.updateByIdSelective(flow);
}
}
}else{
isSleep = true;
}
}catch (Exception e){
e.printStackTrace();
}finally {
//释放资源
if(conn != null){
try {
conn.commit();
} catch (SQLException e) {
if(!flowThreadIsStop){
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if(conn != null){
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if(!flowThreadIsStop){
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if(preparedStatement != null){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!flowThreadIsStop){
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
conn.close();
} catch (SQLException e) {
if(!flowThreadIsStop){
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
if(isSleep){
try {
log.info("------------------【未扫描到要执行的工作流】-------------------");
TimeUnit.MINUTES.sleep(1);
} catch (InterruptedException e) {
log.warn("----------------------【扫描工作流的线程被关闭了】----------------------------");
}
}
}
});
flowThread.setDaemon(true);
flowThread.setName("myth-job#【FlowScanHelper】#flowThreadStart");
flowThread.start();
}
public void doStop(){
this.flowThreadIsStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (flowThread.getState() != Thread.State.TERMINATED) {
flowThread.interrupt();
try {
flowThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【工作扫描线程被注销】-----------------------");
}
/**
* 对齐时钟。整秒运行
*/
private void dateAligned(long waitTime){
try {
TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis()%1000);
} catch (InterruptedException e) {
log.warn("----------------【线程被中断】-----------------------");
}
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/JobScheduleHelper.java deleted 100644 → 0
package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.JobTriggerStatusEnums;
import com.byit.job.WorkRoulette;
import com.byit.model.*;
import com.byit.service.*;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.service.mapservice.RunRecordingAndJobTaskService;
import com.byit.service.mapservice.TaskAndLogServer;
import com.byit.task.JavaBeanJobTask;
import com.byit.task.ScriptExecutorJobTask;
import com.byit.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/**
* @program: byit-myth-job->JobScheduleHeloer
* @description: 工作时间排期表，这里开启了两条线程：
* 一条线程去任务节点读取七秒内将要执行的任务添加进任务排期表
* 一条线程去任务排期表中预读五秒将要执行的任务添加进任务调度轮盘
* @author: huangfu
* @date: 2019/12/9 11:16
**/
@Slf4j
@Component
@Deprecated
public class JobScheduleHelper{
private DataSource dataSource;
private final RunRecordingAndJobTaskService runRecordingAndJobTaskService;
private final JobTaskService jobTaskService;
private final JobTaskScheduleService jobTaskScheduleService;
private final NodeDependencyService nodeDependencyService;
private final JobTaskRunLogService jobTaskRunLogService;
private final TaskAndLogServer taskAndLogServer;
private final RunRecordingService runRecordingService;
/**
* 读取任务节点的预读
*/
private static final long PRE_READ_MS = 7000;
/**
* 读取任务排期表的预读
*/
private static final long SCHEDULE_READ_MS=5000;
/**
* 任务节点线程
*/
private Thread jobInfoThread;
/**
* 任务排期表线程
*/
private Thread scheduleThread;
/**
* 是否停止扫描任务节点表
*/
private volatile boolean jobInfoThreadToStop = false;
/**
* 是否停止排期表的扫描，停止向任务调度轮盘添加任务
*/
private volatile boolean scheduleThreadToStop = false;
@Autowired
public JobScheduleHelper(RunRecordingAndJobTaskService runRecordingAndJobTaskService, JobTaskScheduleService jobTaskScheduleService, JobTaskService jobTaskService, NodeDependencyService nodeDependencyService, JobTaskRunLogService jobTaskRunLogService, TaskAndLogServer taskAndLogServer, RunRecordingService runRecordingService) {
this.runRecordingAndJobTaskService = runRecordingAndJobTaskService;
this.jobTaskScheduleService = jobTaskScheduleService;
this.jobTaskService = jobTaskService;
this.nodeDependencyService = nodeDependencyService;
this.jobTaskRunLogService = jobTaskRunLogService;
this.taskAndLogServer = taskAndLogServer;
this.runRecordingService = runRecordingService;
}
/**
* 启动两条线程
*/
public void start(){
jobInfoThreadStart();
scheduleThreadStart();
}
/**
* 任务节点扫描
*/
public void jobInfoThreadStart(){
jobInfoThread = new Thread(()->{
try {
TimeUnit.MILLISECONDS.sleep(PRE_READ_MS - System.currentTimeMillis()%1000 );
} catch (InterruptedException e) {
if (!jobInfoThreadToStop) {
log.error(e.getMessage(), e);
}
}
log.info("---------------------【init myth-job admin jobInfoThread success】------------------------");
boolean preReadSuc;
while (!jobInfoThreadToStop) {
//开始去扫描任务节点
long start = System.currentTimeMillis();
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
preReadSuc = true;
try{
conn = dataSource.getConnection();
//记录当前的自动提交状态
connAutoCommit = conn.getAutoCommit();
//修改为不自动提交
conn.setAutoCommit(false);
//先添加扫描任务节点的行锁
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'job_task_lock' FOR UPDATE ");
preparedStatement.execute();
//行锁已经加上 后续处理
long nowTime = System.currentTimeMillis();
//开始寻找此时 不是暂停状态，而且七秒内即将运行的任务 而且还不是暂停的节点
List<JobTask> jobTasks = jobTaskService.findJobTaskByTriggerNextTimeLessThanEqual(nowTime + PRE_READ_MS);
if(CollectionUtil.isNotEmpty(jobTasks)){
List<JobTaskSchedule> jobTaskSchedules = new ArrayList<JobTaskSchedule>(15);
//遍历七秒内将要运行的节点数据
for(JobTask jobTask : jobTasks){
/**
* 判断节点状态
* 1.虚节点状态，虚节点状态是映射了一个工作流，需要将该节点映射的工作流下所由的几点拉取到任务表
* 2.普通节点也有两种状态：
* I.开始节点：开始节点不需要验证上级工作流，直接放行执行
* II.正常节点：正常节点需要验证上级节点，首先判断自己是否收弱引用，如果是弱引用那么需要判断
* 上级节点是否已经全部都执行完了，执行完后不论成功与否都执行，同时工作流的运行结果
* 只与end节点关联
*/
//虚节点的状态
if(FlowPropertyEnum.IS_INNER.getCode().equals(jobTask.getIsVirtual())){
try {
runRecordingAndJobTaskService.saveRunRecordingAndTask(jobTask);
log.info("----------------【虚节点保存成功，删除虚节点】--------------------");
jobTaskService.removeMythJobTaskById(jobTask.getId());
} catch (Exception e) {
e.printStackTrace();
}
}else{
if("start".equals(jobTask.getNodeName()) ){
log.debug("任务：{}", jobTask);
JobTaskSchedule jobTaskSchedule = new JobTaskSchedule();
BeanUtils.copyProperties(jobTask,jobTaskSchedule);
jobTaskSchedules.add(jobTaskSchedule);
//更改运行记录为运行中
String runId = jobTask.getRunId();
Integer flowId = jobTask.getFlowId();
RunRecording runRecordingByFlowIdAndRunId = runRecordingService.findRunRecordingByFlowIdAndRunId(flowId, runId);
runRecordingByFlowIdAndRunId.setFlowStatus(FlowPropertyEnum.FLOW_RUN_ING.getCode());
runRecordingService.updateRunRecordingById(runRecordingByFlowIdAndRunId);
}else{
//查询该节点的依赖节点
List<Integer> dependIdByNodeId = nodeDependencyService.findDependIdByNodeId(jobTask.getNodeId());
//这里返回的是上级节点的日志执行情况 把运行中的数据给过滤掉了
List<JobTaskRunLog> jobTaskRunLogList = jobTaskRunLogService.findJobTaskRunLogNotEndNodeByRunCodeCount(dependIdByNodeId, jobTask.getRunId());
if (CollectionUtil.isNotEmpty(jobTaskRunLogList)) {
//判断父类节点是否已经全部完成，只需要判断依赖节点的数目和查询出来的日志数据是否相同
if(dependIdByNodeId.size() == jobTaskRunLogList.size()){
//过滤失败的节点
List<JobTaskRunLog> errorJobLog = jobTaskRunLogList.stream().
filter(jobTaskRunLog -> ("2".equals(jobTaskRunLog.getRunCode()) || "4".equals(jobTaskRunLog.getRunCode()) || "6".equals(jobTaskRunLog.getRunCode())))
.collect(Collectors.toList());
//判断剩余执行次数是否为0
if (parentNodeErrorCount(errorJobLog)) {
log.debug("--------------【{}的上级节点的失败节点已经全部重试完毕】------------------",jobTask);
//该节点如果为弱引用
if ("1".equals(jobTask.getSuperSuccessRun())) {
log.info("-------------【查询到有弱引用节点】-----------------");
//执行代码
runJobTask(jobTask,jobTaskSchedules);
}else{
//如果有失败的节点 就把该节点置为失败
if(CollectionUtil.isNotEmpty(errorJobLog)){
//删除这个数据 并且添加到日志
taskAndLogServer.addRunLogAndRemoveTask(jobTask);
}else{
//执行代码
runJobTask(jobTask,jobTaskSchedules);
}
}
}
}
}
}
}
}
if(CollectionUtil.isNotEmpty(jobTaskSchedules)) {
jobTaskScheduleService.saveAllData(jobTaskSchedules);
jobTaskService.removeMythJobTaskInIds(jobTaskSchedules);
}
}else{
//log.info("-------------------空轮转-------------------");
preReadSuc = false;
}
}catch (Exception e){
if(!jobInfoThreadToStop){
e.printStackTrace();
log.error("------------------扫描任务表出现异常：{}-----------------",e.getMessage());
}
}finally {
//commit
if (conn!=null) {
try {
conn.commit();
}catch (SQLException e){
if(!jobInfoThreadToStop){
log.error("----------------提交行锁错误：{}-------------------",e.getMessage());
}
}
//设置提交状态恢复原来的值
try {
conn.setAutoCommit(connAutoCommit);
}catch (SQLException e){
if(!jobInfoThreadToStop){
log.error("------------------设置为自动提交出错：{}------------------",e.getMessage());
}
}
//关闭连接
try {
conn.close();
}catch (SQLException e){
if(!jobInfoThreadToStop){
log.error("----------------关闭数据库连接：{}-------------------",e.getMessage());
}
}
}
//关闭执行器
if(null !=preparedStatement){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!jobInfoThreadToStop){
log.error("---------------关闭执行器出错：{}----------------------",e.getMessage());
}
}
}
}
//计算总的处理时间 将时间保持在一秒处理一次
long cost = System.currentTimeMillis()-start;
//不足一秒的等待 等够1秒为止
if(cost < 1000){
// 预读期：成功-每秒扫描一次；失败-跳过这段时间
try {
//这个睡眠是空轮转时，延长睡眠时间，奖励CUP的使用频率
TimeUnit.MILLISECONDS.sleep((preReadSuc?1000:PRE_READ_MS) - System.currentTimeMillis()%1000);
} catch (InterruptedException e) {
if (!jobInfoThreadToStop) {
log.error("----------------------{}-------------------------",e.getMessage());
}
}
}
}
});
//设置为守护线程
jobInfoThread.setDaemon(true);
//设置名字
jobInfoThread.setName("myth-job,admin JobScheduleHelper#jobInfoThread");
jobInfoThread.start();
}
/**
* 开始操作任务排期表：
* 1. 扫描五秒要执行的数据
* 2.加载时，创建任务日志，传入调度时间
* 3.加载到任务调度轮盘
* 4.删除数据
*/
private void scheduleThreadStart(){
scheduleThread = new Thread(() ->{
try {
TimeUnit.MILLISECONDS.sleep(4000 - System.currentTimeMillis()%1000 );
} catch (InterruptedException e) {
if (!scheduleThreadToStop) {
log.error(e.getMessage(), e);
}
}
log.info("---------------------init myth-job admin scheduleThread success------------------------");
/**
* 判断是否是空轮转
* 空轮转的话是需要休眠的
*/
boolean preReadSuc;
while (!scheduleThreadToStop){
//开始去扫描任务节点
long start = System.currentTimeMillis();
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
preReadSuc = true;
try{
conn = dataSource.getConnection();
//记录当前的自动提交状态
connAutoCommit = conn.getAutoCommit();
//修改为不自动提交
conn.setAutoCommit(false);
//先添加扫描任务节点的行锁
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'job_task_schedule_lock' FOR UPDATE ");
preparedStatement.execute();
//行锁已经加上 后续处理
long nowTime = System.currentTimeMillis();
//查询所有符合条件的任务节点
List<JobTaskSchedule> jobTaskSchedules = jobTaskScheduleService.findJobTaskScheduleByTriggerNextTimeLessThanEqual(nowTime + SCHEDULE_READ_MS);
if(CollectionUtil.isNotEmpty(jobTaskSchedules)){
//循环遍历添加任务
jobTaskSchedules.forEach(mythJobTaskSchedule ->{
//如果是重跑就有logId
Integer logId = mythJobTaskSchedule.getLogId();
if(logId == null){
logId = saveLog(mythJobTaskSchedule);
}
mythJobTaskSchedule.setLogId(logId);
Long triggerTime = mythJobTaskSchedule.getTriggerTime();
if ("JAVA".equals(mythJobTaskSchedule.getJobType())) {
//构建调度执行器
JavaBeanJobTask javaBeanJobTask = new JavaBeanJobTask(mythJobTaskSchedule);
WorkRoulette.addJob(javaBeanJobTask,triggerTime);
}else if("SCRIPT".equals(mythJobTaskSchedule.getJobType())){
ScriptExecutorJobTask scriptExecutorJobTask = new ScriptExecutorJobTask(mythJobTaskSchedule);
WorkRoulette.addJob(scriptExecutorJobTask,triggerTime);
}
jobTaskScheduleService.delete(mythJobTaskSchedule.getId());
});
}else{
preReadSuc = false;
}
}catch (Exception e){
if(!scheduleThreadToStop){
log.error("------------------扫描排期表出现异常：{}-----------------",e.getMessage());
}
}finally {
//commit
if (conn!=null) {
try {
conn.commit();
}catch (SQLException e){
if(!scheduleThreadToStop){
log.error("----------------提交行锁错误：{}-------------------",e.getMessage());
}
}
//设置提交状态恢复原来的值
try {
conn.setAutoCommit(connAutoCommit);
}catch (SQLException e){
if(!scheduleThreadToStop){
log.error("------------------设置为自动提交出错：{}------------------",e.getMessage());
}
}
//关闭连接
try {
conn.close();
}catch (SQLException e){
if(!scheduleThreadToStop){
log.error("----------------关闭数据库连接：{}-------------------",e.getMessage());
}
}
}
//关闭执行器
if(null !=preparedStatement){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!scheduleThreadToStop){
log.error("---------------关闭执行器出错：{}----------------------",e.getMessage());
}
}
}
//计算总的处理时间 将时间保持在一秒处理一次
long cost = System.currentTimeMillis()-start;
//不足一秒的等待 等够1秒为止
if(cost < 1000){
// 预读期：成功-每秒扫描一次；失败-跳过这段时间
try {
//这个睡眠是空轮转时，延长睡眠时间，奖励CUP的使用频率
TimeUnit.MILLISECONDS.sleep((preReadSuc?1000:SCHEDULE_READ_MS) - System.currentTimeMillis()%1000);
} catch (InterruptedException e) {
if (!scheduleThreadToStop) {
log.error("----------------------{}-------------------------",e.getMessage());
}
}
}
}
}
});
scheduleThread.setName("myth-job,admin JobScheduleHelper#scheduleThread");
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) {
this.dataSource = dataSource;
}
private Integer saveLog(JobTaskSchedule mythJobTaskSchedule){
JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
jobTaskRunLog.setRunId(mythJobTaskSchedule.getRunId());
jobTaskRunLog.setIsVirtual(mythJobTaskSchedule.getIsVirtual());
jobTaskRunLog.setFlowId(mythJobTaskSchedule.getFlowId());
jobTaskRunLog.setFlowName(mythJobTaskSchedule.getFlowName());
jobTaskRunLog.setNodeId(mythJobTaskSchedule.getNodeId());
jobTaskRunLog.setNodeName(mythJobTaskSchedule.getNodeName());
jobTaskRunLog.setRunParams(mythJobTaskSchedule.getRunParam());
jobTaskRunLog.setFailedRemainingCount(mythJobTaskSchedule.getFailedRetryCount());
jobTaskRunLog.setJobType(mythJobTaskSchedule.getJobType());
JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
mythJobTaskRunLogService.saveJobTaskRunLog(jobTaskRunLog);
return jobTaskRunLog.getLogId();
}
/**
* 判断失败节点的重试次数是不是为0
* @param errorJobLog 上级节点的全部失败节点
* @return
*/
private boolean parentNodeErrorCount(List<JobTaskRunLog> errorJobLog){
if(CollectionUtil.isEmpty(errorJobLog)){
return true;
}
for (JobTaskRunLog jobTaskRunLog : errorJobLog) {
//失败重试次数大于0 而且错误原因不是上级节点执行失败
if(jobTaskRunLog.getFailedRemainingCount()>0 && !("6".equals(jobTaskRunLog.getRunCode()))){
log.debug("----------------【{}节点没有重试完毕】----------------",jobTaskRunLog);
return false;
}
}
return true;
}
private void runJobTask(JobTask jobTask,List<JobTaskSchedule> jobTaskSchedules){
//到这里 父类节点一定是全部都执行成功了！
JobTaskSchedule jobTaskSchedule = new JobTaskSchedule();
BeanUtils.copyProperties(jobTask,jobTaskSchedule);
jobTaskSchedules.add(jobTaskSchedule);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/LogScanHelper.java deleted 100644 → 0
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.mapservice.JobTaskRunLogAndJobTaskService;
import com.byit.service.JobTaskRunLogService;
import com.byit.service.RunRecordingService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* @program: byit-myth-job->LogScanHelper
* @description: 对日志表的扫描
* @author: huangfu
* @date: 2019/12/26 10:34
**/
@Component
@Slf4j
@Deprecated
public class LogScanHelper {
private DataSource dataSource;
private final JobTaskRunLogService jobTaskRunLogService;
private final RunRecordingService runRecordingService;
private final JobTaskRunLogAndJobTaskService jobTaskRunLogAndJobTaskService;
@Autowired
public LogScanHelper(JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService, JobTaskRunLogAndJobTaskService jobTaskRunLogAndJobTaskService) {
this.jobTaskRunLogService = jobTaskRunLogService;
this.runRecordingService = runRecordingService;
this.jobTaskRunLogAndJobTaskService = jobTaskRunLogAndJobTaskService;
}
@Autowired
public void setDataSource(DataSource dataSource) {
this.dataSource = dataSource;
}
/**
* 扫描虚节点的线程是否停止
*/
private volatile boolean virtualNodeScanIsStop = false;
/**
* 扫描未完成告警的节点的线程是否停止
*/
private volatile boolean notAlarmedNodeScanIsStop = false;
/**
* 扫描失败的节点的线程是否停止
*/
private volatile boolean errorNodeScanIsStop = false;
/**
* 扫描失败的节点的线程是否停止
*/
private Thread errorNodeScanThread = null;
/**
* 扫描虚节点的线程定义
*/
private Thread virtualNodeScanThread = null;
/**
* 扫描未完成告警的节点的线程
*/
private Thread notAlarmedNodeScanThread = null;
public void start(){
virtualNodeScanMethod();
//notAlarmedNodeScanMethod();
errorNodeScan();
}
/**
* 虚节点扫描
*/
private void virtualNodeScanMethod(){
//扫描虚节点线程
virtualNodeScanThread = new Thread(() ->{
dateAligned(5000);
log.info("---------------------【com.byit.thread.LogScanHelper#virtualNodeScanMethod】init success------------------------");
while (!virtualNodeScanIsStop){
//是否需要睡眠
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 = 'log_virtual_node_lock' FOR UPDATE ");
preparedStatement.execute();
//查询日志表没有完结的虚拟节点
List<JobTaskRunLog> notEndVirtualNodes = jobTaskRunLogService.findNotEndVirtualNode( );
if(CollectionUtil.isNotEmpty(notEndVirtualNodes)){
notEndVirtualNodes.forEach(notEndVirtualNode ->{
log.debug("------------【开始查询虚拟节点的执行情况】----------------");
//根据运行标识和工作流id查询运行日志
RunRecording runRecordingByFlowIdAndRunId = runRecordingService.findRunRecordingByFlowIdAndRunId(notEndVirtualNode.getMapFlowId( ), notEndVirtualNode.getRunId());
//判断当前的工作流是否已经完结
if(runRecordingByFlowIdAndRunId!=null && "4".equals(runRecordingByFlowIdAndRunId.getFlowStatus())){
log.debug("------------【查询到有已经完成的虚拟节点修改日志】----------------");
notEndVirtualNode.setRunCode(runRecordingByFlowIdAndRunId.getFlowRunResult());
notEndVirtualNode.setEndTime(new Date());
//修改日志信息 改为成功或者失败
jobTaskRunLogService.updateJobTaskRunLog(notEndVirtualNode);
}else{
dateAligned(1000);
}
});
}else{
isSleep = true;
}
}catch (Exception e){
e.printStackTrace();
}finally {
//释放资源
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("--------------------【关闭链接出错】---------------------");
}
}
}
if(isSleep){
dateAligned(20000);
}
}
});
virtualNodeScanThread.setName("myth-job#【LogScanHelper】#virtualNodeScanThread");
virtualNodeScanThread.setDaemon(true);
virtualNodeScanThread.start();
}
/**
* @deprecated 废除原因：不需要这条线程去扫描结束节点，只需要扫描对应工作流的节点数目是否全部匹配即可
* 这一操作需要放置到扫描运行实例的线程里面
* @deprecatedDate 2020年3月6日10:36:19
* 未告警的节点扫描
*/
private void notAlarmedNodeScanMethod(){
notAlarmedNodeScanThread = new Thread(() ->{
dateAligned(5000);
log.info("---------------------【com.byit.thread.LogScanHelper#notAlarmedNodeScanMethod】init success------------------------");
while (!notAlarmedNodeScanIsStop){
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 = 'log_node_callback_lock' FOR UPDATE ");
preparedStatement.execute();
//查询的是失败的或者是已经结束的节点（完成的）
List<JobTaskRunLog> jobTaskRunLogEndOrFailureNode = jobTaskRunLogService.findJobTaskRunLogEndOrFailureNode( );
if(CollectionUtil.isNotEmpty(jobTaskRunLogEndOrFailureNode)){
log.debug("-------------【查询到有完结而且未告警的节点】-------------");
//遍历结束的节点 修改执行记录表
jobTaskRunLogEndOrFailureNode.forEach(endNode ->{
String runId = endNode.getRunId( );
Integer flowId = endNode.getFlowId( );
//如果运行结果为null 那么就是调度都没成功 那么就取调度的值
String code = endNode.getRunCode();
runRecordingService.updateRunRecordingByFlowIdAndRunId(RunRecording.builder().runId(runId).flowId(flowId).flowRunResult(code).flowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode()).build());
endNode.setAlertEnd("1");
jobTaskRunLogService.updateJobTaskRunLog(endNode);
});
}else{
isSleep = true;
}
}catch (Exception e){
if(!notAlarmedNodeScanIsStop){
e.printStackTrace();
}
}finally {
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("--------------------【关闭链接出错】---------------------");
}
}
}
if(isSleep){
dateAligned(20000);
}
}
});
notAlarmedNodeScanThread.setName("myth-job#【LogScanHelper】#notAlarmedNodeScanThread");
notAlarmedNodeScanThread.setDaemon(true);
notAlarmedNodeScanThread.start();
}
/**
* 失败的节点扫描 执行失败重试
* 扫描到失败节点，过滤引上级节点失败的情况
*/
public void errorNodeScan(){
errorNodeScanThread = new Thread(()->{
dateAligned(5000);
log.info("---------------------【com.byit.thread.LogScanHelper#errorNodeScan】init success------------------------");
while (!errorNodeScanIsStop) {
//是否需要睡眠
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 = 'log_error_lock' FOR UPDATE ");
preparedStatement.execute();
//查询又重试次数的失败节点
List<JobTaskRunLog> errorNodes = jobTaskRunLogService.findErrorNode();
if(CollectionUtil.isNotEmpty(errorNodes)){
errorNodes.forEach(errorNode ->{
log.debug("-----------------【操作失败节点{}】-----------------",errorNode);
jobTaskRunLogAndJobTaskService.updateLogAndSaveJobTask(errorNode);
});
}else{
isSleep = true;
}
} catch (Exception e) {
e.printStackTrace();
} finally {
//释放资源
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("--------------------【关闭链接出错】---------------------");
}
}
}
if (isSleep) {
dateAligned(20000);
}
}
});
errorNodeScanThread.setName("myth-job#【LogScanHelper】#errorNodeScan");
errorNodeScanThread.setDaemon(true);
errorNodeScanThread.start();
}
public void doStop(){
this.virtualNodeScanIsStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (virtualNodeScanThread.getState() != Thread.State.TERMINATED) {
virtualNodeScanThread.interrupt();
try {
virtualNodeScanThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【虚节点扫描线程被注销】-----------------------");
this.notAlarmedNodeScanIsStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (notAlarmedNodeScanThread.getState() != Thread.State.TERMINATED) {
notAlarmedNodeScanThread.interrupt();
try {
notAlarmedNodeScanThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【日志扫描线程被注销】-----------------------");
this.errorNodeScanIsStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (errorNodeScanThread.getState() != Thread.State.TERMINATED) {
errorNodeScanThread.interrupt();
try {
errorNodeScanThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【失败重试节点扫描线程被注销】-----------------------");
}
/**
* 对齐时钟。整秒运行
*/
private void dateAligned(long waitTime){
try {
TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis()%1000);
} catch (InterruptedException e) {
log.warn("----------------【线程被中断】-----------------------");
}
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/RunRecordingScanHelper.java deleted 100644 → 0
package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import com.byit.enums.RunRecordingEnum;
import com.byit.filesystem.FileSystem;
import com.byit.enums.JobResultEnum;
import com.byit.job.exceptions.BusinessException;
import com.byit.model.EmailAlarm;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.RunRecording;
import com.byit.service.EmailAlarmService;
import com.byit.service.JobTaskRunLogService;
import com.byit.service.RunRecordingService;
import com.byit.util.TimeFormatUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.csource.common.MyException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* 运行记录扫描线程
*
* @author huangfu
*/
@Deprecated
@Component
@Slf4j
public class RunRecordingScanHelper {
public static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
/**
* 完成时告警
*/
public static final String WHEN_DONE = "1";
/**
* 失败时告警
*/
public static final String FAILURE_DONE = "2";
/**
* 成功时告警
*/
public static final String SUCCESS_DONE = "3";
/**
* 文件服务器ip
*/
@Value("${file.system.ip}")
private String fileSystemIp;
/**
* 文件服务器端口号
*/
@Value("${file.system.port}")
private String fileSystemPort;
private static final String FILE_SYSTEM_PRE="http://";
private DataSource dataSource;
private final EmailAlarmService emailAlarmService;
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) {
this.emailAlarmService = emailAlarmService;
this.jobTaskRunLogService = jobTaskRunLogService;
this.runRecordingService = runRecordingService;
this.fileSystem = fileSystem;
}
public void start() {
scanRunRecThread();
judgeFlowIsEndThread();
}
/**
* 判断工作流是否完结
*/
public void judgeFlowIsEndThread(){
judgeFlowIsEndThread = new Thread(() ->{
dateAligned(5000);
log.info("----------------------【com.byit.thread.RunRecordingScanThread#judgeFlowIsEnd】start-------------------");
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();
//进行操作
judgeFlowIsEnd();
isSleep = true;
}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(20000);
}
}
});
judgeFlowIsEndThread.setDaemon(true);
judgeFlowIsEndThread.setName("myth-job#【judgeFlowIsEndThread】# judgeFlowIsEndThread");
judgeFlowIsEndThread.start();
}
/**
* 扫描未完成的工作流
*/
private void judgeFlowIsEnd(){
log.debug("---------------开始扫描运行中的运行实例----------------");
//查询未完结的工作流实例
List<RunRecording> runningRunRecordings = runRecordingService.findRunningRunRecording();
log.debug("---------扫描到运行实例{}个----------",runningRunRecordings.size());
runningRunRecordings.forEach(runRecording -> {
String runId = runRecording.getRunId();
Integer flowId = runRecording.getFlowId();
List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBsByFlowIdAndRunId = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
//判断节点数量与设定数量是否一致
if(jobTaskRunLogWithBLOBsByFlowIdAndRunId==null || jobTaskRunLogWithBLOBsByFlowIdAndRunId.size() != runRecording.getFlowNodeCount()){
log.debug("------------发现工作流{}不符合条件----------------",jobTaskRunLogWithBLOBsByFlowIdAndRunId);
return;
}
//判断是否全部完结 如果全部完结则判断工作流是否成功
if(logIsAllEnd(jobTaskRunLogWithBLOBsByFlowIdAndRunId)){
String runCode = allNodeIsSuccess(jobTaskRunLogWithBLOBsByFlowIdAndRunId);
//设置运行结果
runRecording.setFlowRunResult(runCode);
log.debug("----------------扫描到有完结的运行实例{},{}-------------",runCode,RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
runRecording.setFlowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
runRecordingService.updateRunRecordingById(runRecording);
}
});
}
private String allNodeIsSuccess(List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBs){
for (JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOB : jobTaskRunLogWithBLOBs) {
if (StringUtils.isNotEmpty(jobTaskRunLogWithBLOB.getRunCode())) {
if(!"1".equals(jobTaskRunLogWithBLOB.getRunCode()) || !"3".equals(jobTaskRunLogWithBLOB.getRunCode())){
return jobTaskRunLogWithBLOB.getRunCode();
}
}else{
throw new BusinessException(JobResultEnum.RUN_MSG_FAIL);
}
}
return "1";
}
/**
* 判断节点是否全部完结
* @param jobTaskRunLogWithBLOBs
* @return
*/
private boolean logIsAllEnd(List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBs){
for (JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOB : jobTaskRunLogWithBLOBs) {
if ("0".equals(jobTaskRunLogWithBLOB.getRunCount())) {
//存在运行中的直接返回false
return false;
}
}
return true;
}
/**
* 查询完结且没有告警的节点线程构建
*/
private void scanRunRecThread(){
runRecordingThread = new Thread(() -> {
dateAligned(5000);
log.info("--------------------【com.byit.thread.RunRecordingScanThread#scanRunRecThread】init success----------------------");
while (!runRecordingThreadStop) {
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 = 'run_recording_lock' FOR UPDATE ");
preparedStatement.execute();
//进行操作
scanRunRec();
isSleep = true;
} catch (Exception e) {
if (!runRecordingThreadStop) {
e.printStackTrace();
}
} finally {
if (conn != null) {
try {
conn.commit();
} catch (SQLException e) {
if (!runRecordingThreadStop) {
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if (conn != null) {
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if (!runRecordingThreadStop) {
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if (preparedStatement != null) {
try {
preparedStatement.close();
} catch (SQLException e) {
if (!runRecordingThreadStop) {
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
if(conn!=null){
conn.close();
}
} catch (SQLException e) {
if (!runRecordingThreadStop) {
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
if(isSleep){
dateAligned(20000);
}
}
});
runRecordingThread.setDaemon(true);
runRecordingThread.setName("myth-job#【RunRecordingScanThread】# start");
runRecordingThread.start();
}
/**
* 查询完结 但是没有告警的节点
*/
private void scanRunRec() {
//查询完结且未告警的工作流信息
List<RunRecording> runRecordingByEndAndNotIsAlarm = runRecordingService.findRunRecordingByEndAndNotIsAlarm();
if (CollectionUtil.isNotEmpty(runRecordingByEndAndNotIsAlarm)) {
runRecordingByEndAndNotIsAlarm.forEach(runRecording -> {
//如果设置为完成时告警
switch (runRecording.getAlarmlAction()) {
//设置为完成时告警
case WHEN_DONE:
if (RunRecordingEnum.FLOW_STATUS_IS_END.getCode().equals(runRecording.getFlowStatus())) {
saveEmailAlarms(runRecording);
}
break;
//失败时告警
case FAILURE_DONE:
if (RunRecordingEnum.RUN_FLOW_FAILURE.getCode().equals(runRecording.getFlowRunResult())
|| RunRecordingEnum.RUN_FLOW_RE_FAILURE.getCode().equals(runRecording.getFlowRunResult())) {
saveEmailAlarms(runRecording);
}
break;
//成功时告警
case SUCCESS_DONE:
if (RunRecordingEnum.RUN_FLOW_SUCCESS.getCode().equals(runRecording.getFlowRunResult())
|| RunRecordingEnum.RUN_FLOW_RE_SUCCESS.getCode().equals(runRecording.getFlowRunResult())) {
saveEmailAlarms(runRecording);
}
break;
default:
break;
}
});
}
}
public void doStop() {
this.runRecordingThreadStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
if (runRecordingThread.getState() != Thread.State.TERMINATED) {
runRecordingThread.interrupt();
try {
runRecordingThread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.warn("---------------【运行记录扫描日志线程被注销】-----------------------");
this.judgeFlowIsEndStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
if (judgeFlowIsEndThread.getState() != Thread.State.TERMINATED) {
judgeFlowIsEndThread.interrupt();
try {
judgeFlowIsEndThread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.warn("---------------【扫描工作流结果线程被注销】-----------------------");
}
/**
* 对齐时钟。整秒运行
*/
private void dateAligned(long waitTime) {
try {
TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis() % 1000);
} catch (InterruptedException e) {
log.warn("----------------【线程被中断】-----------------------");
}
}
private void saveEmailAlarms(RunRecording runRecording) {
//这一步是根据flowId和RunId查询对应的节点信息
List<JobTaskRunLogWithBLOBs> jobTaskRunLogByFlowIdAndRunId = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(runRecording.getFlowId(), runRecording.getRunId());
String flowName = runRecording.getFlowName();
String senContentHtml = runMsgHtml(jobTaskRunLogByFlowIdAndRunId, flowName);
EmailAlarm emailAlarm = EmailAlarm.builder()
.flowId(runRecording.getFlowId())
.flowName(flowName)
.runId(runRecording.getRunId())
.versionName(runRecording.getFlowVersionName())
.alarmEmail(runRecording.getAlarmEmail())
.alarmContent(senContentHtml)
.alarmResult("0")
.flowRes(runRecording.getFlowRunResult())
.alarmTitle(flowName)
.build();
//保存邮箱
emailAlarmService.saveEmailAlarm(emailAlarm);
//修改为已告警
runRecording.setIsAlarm("0");
runRecordingService.updateRunRecordingById(runRecording);
}
private String runMsgHtml(List<JobTaskRunLogWithBLOBs> jobTaskRunLogs, String title) {
StringBuilder stringBuilder = new StringBuilder();
stringBuilder.append("<table border='1' width='80%' align='center' cellspacing='0' cellpadding='6'>")
.append(String.format("<h2 style='text-align:center;color:red'>%s</h2>", title))
.append("<thead align='center' style='background: blue;color: #fff'>")
.append("<th width = '10%'>节点名称</th>")
.append("<th width = '10%'>开始时间</th>")
.append("<th width = '10%'>结束时间</th>")
.append("<th width = '10%'>耗费时间</th>")
.append("<th width = '10%'>运行结果</th>")
.append("<th width = '50%'>运行日志</th>")
.append("</thead>")
.append("<tbody>");
if (CollectionUtil.isNotEmpty(jobTaskRunLogs)) {
jobTaskRunLogs.forEach(jobTaskRunLog -> {
long timeConsuming = jobTaskRunLog.getEndTime().getTime() - jobTaskRunLog.getStartTime().getTime();
/*
* String iframeHtml = "<iframe src='%s'></iframe>";
* String logFilePath = FILE_SYSTEM_PRE+fileSystemIp+":"+fileSystemPort+"/"+jobTaskRunLog.getLogRemotelyPath();
* String logFileUrl = String.format(iframeHtml,logFilePath);
*/
String logStr = null;
try {
if(null != jobTaskRunLog.getLogRemotelyPath()){
logStr = new String(fileSystem.downloaderFile(jobTaskRunLog.getLogRemotelyPath()), StandardCharsets.UTF_8);
}
} catch (IOException | MyException e) {
e.printStackTrace();
}
stringBuilder.append("<tr align='center'>")
.append(String.format("<td>%s</td>", jobTaskRunLog.getNodeName()))
.append(String.format("<td>%s</td>", DateUtil.format(jobTaskRunLog.getStartTime(), DATE_FORMAT)))
.append(String.format("<td>%s</td>", DateUtil.format(jobTaskRunLog.getEndTime(), DATE_FORMAT)))
.append(String.format("<td>%s</td>", TimeFormatUtil.timeFormat(timeConsuming)))
.append(String.format("<td>%s</td>", "1".equals(jobTaskRunLog.getRunCode()) ? "成功"
: "3".equals(jobTaskRunLog.getRunCode()) ? "补批成功"
: "4".equals(jobTaskRunLog.getRunCode()) ? "补批失败" : "失败"))
.append("<td>")
.append("<div style='display:inline-block;width:100%;word-break:break-all;height: auto;overflow: auto;text-align: left;'>")
.append(String.format("%s", logStr==null?jobTaskRunLog.getRunMsg():logStr))
.append("</div>")
.append("</td></tr>");
});
}
stringBuilder.append("</tbody>")
.append("</table>");
return stringBuilder.toString();
}
@Autowired
public void setDataSource(DataSource dataSource) {
this.dataSource = dataSource;
}
}
\ 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