Alawn/udf
1
0
Fork 0
mirror of https://gitee.com/wangkang/udf.git synced 2026-11-03 17:23:40 +00:00
Code Issues Releases Wiki Activity

调整RmsJobDisallowConcurrent处理逻辑

Browse Source
This commit is contained in:
wangkang
2017-08-28 19:35:52 +08:00
parent a63b78e130
commit a1c1c6f560
6 changed files with 188 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/RmsJobStats.java
+5 -1
View File
@@ -23,7 +23,11 @@ public enum RmsJobStats {
/**
* 描述 : 错误
*/
ERROR(-1);
ERROR(-1),
/**
* 描述 : 跳过
*/
SKIP(-2);
/**
* <p>
udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/domain/RmsJobParam.java
+6
View File
@@ -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名称
*/
udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/IRmsJobLog.java
+9 -2
View File
@@ -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);
udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/AbstractBaseRmsJob.java
+158
View File
@@ -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<RestResponse<RmsJobResult>> result = rms.call(rmsJobParam.getServiceCode(), rmsJobParam, null, new ParameterizedTypeReference<RestResponse<RmsJobResult>>() {
}, 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);
}
}
}
udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/RmsJob.java
+2 -85
View File
@@ -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<RestResponse<RmsJobResult>> result = rms.call(serviceCode, rmsJobParam, null, new ParameterizedTypeReference<RestResponse<RmsJobResult>>() {
}, 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));
}
}
udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/RmsJobDisallowConcurrent.java
+8 -2
View File
@@ -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));
}
}