yineng.huang
/
hz
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: 执行计划 

1.添加批次为空的处理逻辑
2.添加批次下总任务数量与生成任务数量比较的逻辑。
hz_1122
李杰应 2024-12-19 03:44:51 +08:00
parent 58c743a889
commit c9fac180e6
1 changed files with 53 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

80
cctp-atu/atu-execute-plan/src/main/java/net/northking/cctp/executePlan/job/PlanBatchTaskDataUpdateJob.java
View File

@ -121,6 +121,8 @@ public class PlanBatchTaskDataUpdateJob {
private final static String LOCK_KEY_BATCH_SUM = "LOCK:PLAN:BATCH-SUM-DATA-UPDATE-METHOD"; private final static String LOCK_KEY_BATCH_SUM = "LOCK:PLAN:BATCH-SUM-DATA-UPDATE-METHOD";
private final static String BATCH_IS_NULL_COUNT_HASH = "PLAN:BATCH:NULL-COUNT-HASH";
@Autowired @Autowired
private AtuPlanConfig atuPlanConfig; private AtuPlanConfig atuPlanConfig;
@ -165,6 +167,37 @@ public class PlanBatchTaskDataUpdateJob {
RedisConstant.CLUSTER_KEY_SUFFIX; RedisConstant.CLUSTER_KEY_SUFFIX;
// 通过数据库进行统计，减少统计数据出错的原因 // 通过数据库进行统计，减少统计数据出错的原因
AtuPlanBatch batch = this.planBatchService.findByPrimaryKey(batchId);
if (batch == null) {
logger.info("批次【{}】信息目前不存在，再等待下一次查询...", batchId);
Boolean existFlag = redisTemplate.opsForHash().hasKey(BATCH_IS_NULL_COUNT_HASH, batchId);
if (existFlag) {
Object nullCount = redisTemplate.opsForHash().get(BATCH_IS_NULL_COUNT_HASH, batchId);
if (nullCount != null && NumberUtil.isNumber(nullCount.toString())) {
int i = Integer.parseInt(nullCount.toString());
if (i < 100) { // 统计100次后依然没有批次信息，批次不存在
redisTemplate.opsForHash().put(BATCH_IS_NULL_COUNT_HASH, batchId, i + 1);
} else {
// 没有对应批次数据，删除缓存数据
logger.debug("没有对应的批次信息[{}]，删除缓存数据", batchId);
redisTemplate.delete(key);
}
}
} else {
redisTemplate.opsForHash().put(BATCH_IS_NULL_COUNT_HASH, batchId, 1);
}
return; // 退出这个批次统计
}
// 批次信息存在
AtuPlanTask totalQ = new AtuPlanTask();
totalQ.setBatchId(batchId);
long currentTaskTotal = planTaskService.count(totalQ);
// 判断批次任务是否已经全部生成
if (batch.getTaskTotal() > currentTaskTotal) {
logger.info("批次[{}]正在生成任务中，结束统计...", batchId);
return;
}
// 以脚本维度统计批次下各脚本的执行情况
AtuPlanTask taskQ = new AtuPlanTask(); AtuPlanTask taskQ = new AtuPlanTask();
taskQ.setBatchId(batchId); taskQ.setBatchId(batchId);
List<ScriptStatusStatistic> resultList = planTaskService.countTaskStatus(taskQ); List<ScriptStatusStatistic> resultList = planTaskService.countTaskStatus(taskQ);
@ -192,36 +225,29 @@ public class PlanBatchTaskDataUpdateJob {
if (waitCount > 0 || ingCount > 0) { // 批次还没有完成 if (waitCount > 0 || ingCount > 0) { // 批次还没有完成
// 检查批次的开始时间 // 检查批次的开始时间
AtuPlanBatch batch = this.planBatchService.findByPrimaryKey(batchId); Date createdTime = batch.getCreatedTime();
if (batch != null) { if (createdTime != null) { // 有创建时间，那么比较开始时间与现在的时间差
Date createdTime = batch.getCreatedTime(); LocalDateTime createDateTime = createdTime.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
if (createdTime != null) { // 有创建时间，那么比较开始时间与现在的时间差 LocalDateTime now = LocalDateTime.now();
LocalDateTime createDateTime = createdTime.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime(); Duration duration = Duration.between(createDateTime, now);
LocalDateTime now = LocalDateTime.now(); long hours = duration.toHours();
Duration duration = Duration.between(createDateTime, now); if (atuPlanConfig.isBatchExecuteTimeoutEnabled() && hours > atuPlanConfig.getBatchExecuteTimeout().longValue()) { // 参数化
long hours = duration.toHours(); logger.info("批次[{}-{}]已创建{}小时，任务没有执行完成，强制结束批次", batch.getId(), batch.getBatch(), 48);
if (atuPlanConfig.isBatchExecuteTimeoutEnabled() && hours > atuPlanConfig.getBatchExecuteTimeout().longValue()) { // 参数化 // 如果批次的创建时间离现在已经过去48小时还没完成，那么判定批次已完成，等待中|执行中 的任务全部算超时处理
logger.info("批次[{}-{}]已创建{}小时，任务没有执行完成，强制结束批次", batch.getId(), batch.getBatch(), 48); timeoutCount += waitCount;
// 如果批次的创建时间离现在已经过去48小时还没完成，那么判定批次已完成，等待中|执行中 的任务全部算超时处理 timeoutCount += ingCount;
timeoutCount += waitCount; waitCount = 0;
timeoutCount += ingCount; ingCount = 0;
waitCount = 0; // 更新批次下 等待中（0）|执行中（1） 的任务全部算超时处理（6）
ingCount = 0; this.planTaskService.batchUpdateTaskStatusToTimeout(batch.getId(), PlanConstant.TASK_TIMEOUT_STATUS, PlanConstant.TASK_WAIT_EXECUTE_STATUS, PlanConstant.TASK_START_EXECUTE_STATUS);
// 更新批次下 等待中（0）|执行中（1） 的任务全部算超时处理（6）
this.planTaskService.batchUpdateTaskStatusToTimeout(batch.getId(), PlanConstant.TASK_TIMEOUT_STATUS, PlanConstant.TASK_WAIT_EXECUTE_STATUS, PlanConstant.TASK_START_EXECUTE_STATUS);
} else {
// 批次还没有完成，退出统计
return;
}
} else { } else {
// 有批次信息，但是没有创建时间，应该是数据有问题 // 批次还没有完成，退出统计
logger.debug("批次[{}]没有创建时间", batch.getId());
return; return;
} }
} else { // 没有对应批次数据，删除缓存数据 } else {
logger.debug("没有对应的批次信息[{}]，删除缓存数据", batchId); // 有批次信息，但是没有创建时间，应该是数据有问题
redisTemplate.delete(key); logger.debug("批次[{}]没有创建时间", batch.getId());
return; // 退出这个批次统计 return;
} }
} }
// 批次已完成 // 批次已完成