diff --git a/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/RmsJobStats.java b/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/RmsJobStats.java index d8e9e73..1c98824 100644 --- a/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/RmsJobStats.java +++ b/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/RmsJobStats.java @@ -23,7 +23,11 @@ public enum RmsJobStats { /** * 描述 : 错误 */ - ERROR(-1); + ERROR(-1), + /** + * 描述 : 跳过 + */ + SKIP(-2); /** *

diff --git a/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/domain/RmsJobParam.java b/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/domain/RmsJobParam.java index d76357f..99f07d7 100644 --- a/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/domain/RmsJobParam.java +++ b/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/domain/RmsJobParam.java @@ -51,6 +51,12 @@ public class RmsJobParam implements Serializable { @ApiModelProperty(value = "触发实例ID", required = true, dataType = "string") private String triggerType; + /** + * 描述 : 服务代码 + */ + @ApiModelProperty(value = "服务代码", required = true, dataType = "string") + private String serviceCode; + /** * 描述 : bean名称 */ diff --git a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/IRmsJobLog.java b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/IRmsJobLog.java index c5f6dc6..2ef8ebf 100644 --- a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/IRmsJobLog.java +++ b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/IRmsJobLog.java @@ -17,10 +17,17 @@ import org.itkk.udf.scheduler.client.domain.RmsJobResult; public interface IRmsJobLog { /** - * + * hasRunning + * + * @param rmsJobParam rmsJobParam + * @return boolean + */ + boolean hasRunning(RmsJobParam rmsJobParam); + + /** * 描述 : save * - * @param param param + * @param param param * @param result result */ void save(RmsJobParam param, RmsJobResult result); diff --git a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/AbstractBaseRmsJob.java b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/AbstractBaseRmsJob.java new file mode 100644 index 0000000..30348b2 --- /dev/null +++ b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/AbstractBaseRmsJob.java @@ -0,0 +1,158 @@ +/** + * RmsJob.java + * Created at 2017-06-04 + * Created by Administrator + * Copyright (C) 2016 itkk.org, All rights reserved. + */ +package org.itkk.udf.scheduler.job; + +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.ArrayUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.commons.lang3.exception.ExceptionUtils; +import org.itkk.udf.core.RestResponse; +import org.itkk.udf.rms.Rms; +import org.itkk.udf.scheduler.IRmsJobLog; +import org.itkk.udf.scheduler.client.RmsJobStats; +import org.itkk.udf.scheduler.client.SchException; +import org.itkk.udf.scheduler.client.TriggerDataMapKey; +import org.itkk.udf.scheduler.client.TriggerType; +import org.itkk.udf.scheduler.client.domain.RmsJobParam; +import org.itkk.udf.scheduler.client.domain.RmsJobResult; +import org.quartz.JobDataMap; +import org.quartz.JobExecutionContext; +import org.quartz.JobExecutionException; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; + +import java.util.UUID; + +/** + * 描述 : RmsJob + * + * @author Administrator + */ +@Slf4j +public abstract class AbstractBaseRmsJob extends AbstractBaseJob { + + /** + * 获得RmsJobParam + * + * @param jobExecutionContext jobExecutionContext + * @return RmsJobParam + */ + protected RmsJobParam getRmsJobParam(JobExecutionContext jobExecutionContext) { + //从trigger中获得jobDataMap + JobDataMap jobDataMap = jobExecutionContext.getTrigger().getJobDataMap(); + //获得必要字段 + String triggerId = jobDataMap.getString(TriggerDataMapKey.TRIGGER_ID.value()); + triggerId = StringUtils.isBlank(triggerId) ? UUID.randomUUID().toString() : triggerId; + String parentTriggerId = jobDataMap.getString(TriggerDataMapKey.PARENT_TRIGGER_ID.value()); + String triggerType = jobDataMap.getString(TriggerDataMapKey.TRIGGER_TYPE.value()); + triggerType = StringUtils.isBlank(triggerType) ? TriggerType.TRIGGER.value() : triggerType; + String serviceCode = jobDataMap.getString(TriggerDataMapKey.SERVICE_CODE.value()); + String beanName = jobDataMap.getString(TriggerDataMapKey.BEAN_NAME.value()); + Boolean async = jobDataMap.getBoolean(TriggerDataMapKey.ASYNC.value()); + //封装请求参数 + RmsJobParam rmsJobParam = new RmsJobParam(); + rmsJobParam.setId(triggerId); + rmsJobParam.setParentId(parentTriggerId); + rmsJobParam.setTriggerType(triggerType); + rmsJobParam.setFireInstanceId(jobExecutionContext.getFireInstanceId()); + rmsJobParam.setServiceCode(serviceCode); + rmsJobParam.setBeanName(beanName); + rmsJobParam.setAsync(async); + rmsJobParam.setJobDataMap(jobDataMap.getWrappedMap()); + //返回 + return rmsJobParam; + } + + /** + * 锁定执行 + * + * @param rmsJobParam rmsJobParam + * @throws JobExecutionException JobExecutionException + */ + protected void disallowConcurrentExecute(RmsJobParam rmsJobParam) throws JobExecutionException { + if (!this.hasRunning(rmsJobParam)) { //没有正在运行的任务才能运行 + this.execute(rmsJobParam); + } else { //跳过执行,并且记录 + RmsJobResult result = new RmsJobResult(); + result.setId(rmsJobParam.getId()); + result.setStats(RmsJobStats.SKIP.value()); + saveRmsJobLog(rmsJobParam, result); + } + } + + /** + * 执行 + * + * @param rmsJobParam rmsJobParam + * @throws JobExecutionException JobExecutionException + */ + protected void execute(RmsJobParam rmsJobParam) throws JobExecutionException { + try { + //校验 + if (StringUtils.isBlank(rmsJobParam.getServiceCode())) { + throw new SchException("serviceCode can not be empty"); + } + if (StringUtils.isBlank(rmsJobParam.getBeanName())) { + throw new SchException("beanName can not be empty"); + } + //获得RMS + Rms rms = this.getApplicationContext().getBean(Rms.class); + //请求 + ResponseEntity> result = rms.call(rmsJobParam.getServiceCode(), rmsJobParam, null, new ParameterizedTypeReference>() { + }, null); + //判断http状态 + if (result.getStatusCode() != HttpStatus.OK) { + //抛出异常 + throw new SchException("http状态:" + result.getStatusCode().toString()); + } + //记录 + saveRmsJobLog(rmsJobParam, result.getBody().getResult()); + } catch (Exception e) { + //定义返回值 + RmsJobResult result = new RmsJobResult(); + result.setId(rmsJobParam.getId()); + result.setStats(RmsJobStats.ERROR.value()); + result.setErrorMsg(ExceptionUtils.getStackTrace(e)); + //记录 + saveRmsJobLog(rmsJobParam, result); + //抛出异常 + log.error("RmsJob error:", e); + throw new JobExecutionException(e); + } + } + + /** + * 检查是否有正在运行 + * + * @param rmsJobParam rmsJobParam + * @return 是否正在运行 + */ + private boolean hasRunning(RmsJobParam rmsJobParam) { + String[] beanNames = this.getApplicationContext().getBeanNamesForType(IRmsJobLog.class); + if (ArrayUtils.isNotEmpty(beanNames)) { + IRmsJobLog rmsJobLog = this.getApplicationContext().getBean(IRmsJobLog.class); + return rmsJobLog.hasRunning(rmsJobParam); + } + return false; + } + + /** + * 保存 + * + * @param param param + * @param result result + */ + private void saveRmsJobLog(RmsJobParam param, RmsJobResult result) { + String[] beanNames = this.getApplicationContext().getBeanNamesForType(IRmsJobLog.class); + if (ArrayUtils.isNotEmpty(beanNames)) { + IRmsJobLog rmsJobLog = this.getApplicationContext().getBean(IRmsJobLog.class); + rmsJobLog.save(param, result); + } + } + +} diff --git a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/RmsJob.java b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/RmsJob.java index f974111..bba1255 100644 --- a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/RmsJob.java +++ b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/RmsJob.java @@ -7,26 +7,8 @@ package org.itkk.udf.scheduler.job; import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.ArrayUtils; -import org.apache.commons.lang3.StringUtils; -import org.apache.commons.lang3.exception.ExceptionUtils; -import org.itkk.udf.core.RestResponse; -import org.itkk.udf.rms.Rms; -import org.itkk.udf.scheduler.IRmsJobLog; -import org.itkk.udf.scheduler.client.RmsJobStats; -import org.itkk.udf.scheduler.client.SchException; -import org.itkk.udf.scheduler.client.TriggerDataMapKey; -import org.itkk.udf.scheduler.client.TriggerType; -import org.itkk.udf.scheduler.client.domain.RmsJobParam; -import org.itkk.udf.scheduler.client.domain.RmsJobResult; -import org.quartz.JobDataMap; import org.quartz.JobExecutionContext; import org.quartz.JobExecutionException; -import org.springframework.core.ParameterizedTypeReference; -import org.springframework.http.HttpStatus; -import org.springframework.http.ResponseEntity; - -import java.util.UUID; /** * 描述 : RmsJob @@ -34,76 +16,11 @@ import java.util.UUID; * @author Administrator */ @Slf4j -public class RmsJob extends AbstractBaseJob { +public class RmsJob extends AbstractBaseRmsJob { @Override protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException { //NOSONAR - //从trigger中获得jobDataMap - JobDataMap jobDataMap = jobExecutionContext.getTrigger().getJobDataMap(); - //获得必要字段 - String triggerId = jobDataMap.getString(TriggerDataMapKey.TRIGGER_ID.value()); - triggerId = StringUtils.isBlank(triggerId) ? UUID.randomUUID().toString() : triggerId; - String parentTriggerId = jobDataMap.getString(TriggerDataMapKey.PARENT_TRIGGER_ID.value()); - String triggerType = jobDataMap.getString(TriggerDataMapKey.TRIGGER_TYPE.value()); - triggerType = StringUtils.isBlank(triggerType) ? TriggerType.TRIGGER.value() : triggerType; - String serviceCode = jobDataMap.getString(TriggerDataMapKey.SERVICE_CODE.value()); - String beanName = jobDataMap.getString(TriggerDataMapKey.BEAN_NAME.value()); - Boolean async = jobDataMap.getBoolean(TriggerDataMapKey.ASYNC.value()); - //封装请求参数 - RmsJobParam rmsJobParam = new RmsJobParam(); - rmsJobParam.setId(triggerId); - rmsJobParam.setParentId(parentTriggerId); - rmsJobParam.setTriggerType(triggerType); - rmsJobParam.setFireInstanceId(jobExecutionContext.getFireInstanceId()); - rmsJobParam.setBeanName(beanName); - rmsJobParam.setAsync(async); - rmsJobParam.setJobDataMap(jobDataMap.getWrappedMap()); - try { - //校验 - if (StringUtils.isBlank(serviceCode)) { - throw new SchException("serviceCode can not be empty"); - } - if (StringUtils.isBlank(rmsJobParam.getBeanName())) { - throw new SchException("beanName can not be empty"); - } - //获得RMS - Rms rms = this.getApplicationContext().getBean(Rms.class); - //请求 - ResponseEntity> result = rms.call(serviceCode, rmsJobParam, null, new ParameterizedTypeReference>() { - }, null); - //判断http状态 - if (result.getStatusCode() != HttpStatus.OK) { - //抛出异常 - throw new SchException("http状态:" + result.getStatusCode().toString()); - } - //记录 - saveRmsJobLog(rmsJobParam, result.getBody().getResult()); - } catch (Exception e) { - //定义返回值 - RmsJobResult result = new RmsJobResult(); - result.setId(rmsJobParam.getId()); - result.setStats(RmsJobStats.ERROR.value()); - result.setErrorMsg(ExceptionUtils.getStackTrace(e)); - //记录 - saveRmsJobLog(rmsJobParam, result); - //抛出异常 - log.error("RmsJob error:", e); - throw new JobExecutionException(e); - } - } - - /** - * 保存 - * - * @param param param - * @param result result - */ - private void saveRmsJobLog(RmsJobParam param, RmsJobResult result) { - String[] beanNames = this.getApplicationContext().getBeanNamesForType(IRmsJobLog.class); - if (ArrayUtils.isNotEmpty(beanNames)) { - IRmsJobLog rmsJobLog = this.getApplicationContext().getBean(IRmsJobLog.class); - rmsJobLog.save(param, result); - } + this.execute(this.getRmsJobParam(jobExecutionContext)); } } diff --git a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/RmsJobDisallowConcurrent.java b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/RmsJobDisallowConcurrent.java index 69105f2..139b427 100644 --- a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/RmsJobDisallowConcurrent.java +++ b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/RmsJobDisallowConcurrent.java @@ -7,14 +7,20 @@ package org.itkk.udf.scheduler.job; import org.quartz.DisallowConcurrentExecution; +import org.quartz.JobExecutionContext; +import org.quartz.JobExecutionException; /** * 描述 : RmsJobDisallowConcurrent * * @author Administrator - * */ @DisallowConcurrentExecution -public class RmsJobDisallowConcurrent extends RmsJob { +public class RmsJobDisallowConcurrent extends AbstractBaseRmsJob { + + @Override + protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException { //NOSONAR + this.disallowConcurrentExecute(this.getRmsJobParam(jobExecutionContext)); + } }