diff --git a/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/domain/RmsJobResult.java b/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/domain/RmsJobResult.java index e33ea26..919bf9a 100644 --- a/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/domain/RmsJobResult.java +++ b/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/domain/RmsJobResult.java @@ -65,4 +65,10 @@ public class RmsJobResult implements Serializable { @ApiModelProperty(value = "错误信息", required = true, dataType = "string") private String errorMsg; + /** + * 描述 : param + */ + @ApiModelProperty(value = "参数", required = true, dataType = "string") + private RmsJobParam param; + } diff --git a/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/web/SchClientController.java b/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/web/SchClientController.java index bf12f01..6a5d426 100644 --- a/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/web/SchClientController.java +++ b/udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/web/SchClientController.java @@ -6,8 +6,6 @@ */ package org.itkk.udf.scheduler.client.web; -import java.util.Date; - import org.itkk.udf.core.RestResponse; import org.itkk.udf.scheduler.client.RmsJobStats; import org.itkk.udf.scheduler.client.SchClientHandle; @@ -17,6 +15,8 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RestController; +import java.util.Date; + /** * 描述 : SchClientController * @@ -40,6 +40,7 @@ public class SchClientController implements ISchClientController { result.setClientReceiveTime(receiveTime); result.setId(param.getId()); result.setClientStartExecuteTime(new Date()); + result.setParam(param); //执行(区分同步跟异步) if (param.getAsync()) { schClientHandle.asyncHandle(param, result); diff --git a/udf-starter-scheduler/pom.xml b/udf-starter-scheduler/pom.xml index 5442003..63943bc 100644 --- a/udf-starter-scheduler/pom.xml +++ b/udf-starter-scheduler/pom.xml @@ -25,6 +25,11 @@ org.itkk.udf udf-starter-rms + + + org.itkk.udf + udf-starter-cache-redis + org.itkk.udf diff --git a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/IRmsJobEvent.java b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/IRmsJobEvent.java index f9c19ea..6e0c3ff 100644 --- a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/IRmsJobEvent.java +++ b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/IRmsJobEvent.java @@ -16,21 +16,6 @@ import org.itkk.udf.scheduler.client.domain.RmsJobResult; */ public interface IRmsJobEvent { - /** - * hasRunning - * - * @param rmsJobParam rmsJobParam - * @return boolean - */ - boolean beginRun(RmsJobParam rmsJobParam); - - /** - * endRun - * - * @param rmsJobParam rmsJobParam - */ - void endRun(RmsJobParam rmsJobParam); - /** * 描述 : save * diff --git a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/impl/DefRmsJobEvent.java b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/impl/DefRmsJobEvent.java index a1c97b1..435dfef 100644 --- a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/impl/DefRmsJobEvent.java +++ b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/impl/DefRmsJobEvent.java @@ -11,17 +11,6 @@ import org.itkk.udf.scheduler.client.domain.RmsJobResult; @Slf4j public class DefRmsJobEvent implements IRmsJobEvent { - @Override - public boolean beginRun(RmsJobParam rmsJobParam) { - log.info("----- DefRmsJobEvent.beginRun -----> {}", rmsJobParam.toString()); - return false; - } - - @Override - public void endRun(RmsJobParam rmsJobParam) { - log.info("----- DefRmsJobEvent.endRun -----> {}", rmsJobParam.toString()); - } - @Override public void save(RmsJobParam param, RmsJobResult result) { if (param != null) { 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 index 632005f..a381536 100644 --- 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 @@ -10,6 +10,7 @@ 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.cache.redis.lock.RedisBasedDistributedLock; import org.itkk.udf.core.RestResponse; import org.itkk.udf.rms.Rms; import org.itkk.udf.scheduler.IRmsJobEvent; @@ -22,11 +23,13 @@ import org.itkk.udf.scheduler.client.domain.RmsJobResult; import org.quartz.JobDataMap; import org.quartz.JobExecutionContext; import org.quartz.JobExecutionException; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.core.ParameterizedTypeReference; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import java.util.UUID; +import java.util.concurrent.TimeUnit; /** * 描述 : RmsJob @@ -36,6 +39,12 @@ import java.util.UUID; @Slf4j public abstract class AbstractBaseRmsJob extends AbstractBaseJob { + /** + * redisBasedDistributedLock + */ + @Autowired + private RedisBasedDistributedLock redisBasedDistributedLock; + /** * 获得RmsJobParam * @@ -75,17 +84,25 @@ public abstract class AbstractBaseRmsJob extends AbstractBaseJob { * @throws JobExecutionException JobExecutionException */ protected void disallowConcurrentExecute(RmsJobParam rmsJobParam) throws JobExecutionException { - try { - if (!this.beginRun(rmsJobParam)) { //没有正在运行的任务才能运行 + //定义锁信息 (锁超时时间为60分钟) (异步任务,解锁在callback中完成,同步任务,解锁在本方法完成) + String name = "rmsJobDisallowConcurrent_" + rmsJobParam.getServiceCode() + "_" + rmsJobParam.getBeanName(); + final long timeout = 60; + TimeUnit timeUnit = TimeUnit.MINUTES; + //尝试锁定,锁定成功则触发,锁定失败则跳过 + if (redisBasedDistributedLock.lock(name, timeout, timeUnit, rmsJobParam)) { + try { this.execute(rmsJobParam); - } else { //跳过执行,并且记录 - RmsJobResult result = new RmsJobResult(); - result.setId(rmsJobParam.getId()); - result.setStats(RmsJobStats.SKIP.value()); - save(rmsJobParam, result); + } finally { + //同步任务解锁 + if (!rmsJobParam.getAsync()) { + redisBasedDistributedLock.unlock(name); + } } - } finally { - this.endRun(rmsJobParam); + } else { //跳过执行,并且记录 + RmsJobResult result = new RmsJobResult(); + result.setId(rmsJobParam.getId()); + result.setStats(RmsJobStats.SKIP.value()); + save(rmsJobParam, result); } } @@ -130,29 +147,6 @@ public abstract class AbstractBaseRmsJob extends AbstractBaseJob { } } - /** - * 开始运行任务 - * - * @param rmsJobParam rmsJobParam - * @return 是否正在运行 - */ - private boolean beginRun(RmsJobParam rmsJobParam) { - IRmsJobEvent rmsJobEvent = this.getRmsJobEvent(); - return rmsJobEvent != null && rmsJobEvent.beginRun(rmsJobParam); - } - - /** - * 结束运行任务 - * - * @param rmsJobParam rmsJobParam - */ - private void endRun(RmsJobParam rmsJobParam) { - IRmsJobEvent rmsJobEvent = this.getRmsJobEvent(); - if (rmsJobEvent != null) { - rmsJobEvent.endRun(rmsJobParam); - } - } - /** * 保存 * diff --git a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/web/SchClientCallbackController.java b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/web/SchClientCallbackController.java index c4556b7..34b76fd 100644 --- a/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/web/SchClientCallbackController.java +++ b/udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/web/SchClientCallbackController.java @@ -6,14 +6,14 @@ */ package org.itkk.udf.scheduler.web; +import org.itkk.udf.cache.redis.lock.RedisBasedDistributedLock; +import org.itkk.udf.core.RestResponse; import org.itkk.udf.scheduler.IRmsJobEvent; +import org.itkk.udf.scheduler.client.domain.RmsJobResult; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RestController; -import org.itkk.udf.core.RestResponse; -import org.itkk.udf.scheduler.client.domain.RmsJobResult; - /** * 描述 : JobController * @@ -28,12 +28,25 @@ public class SchClientCallbackController implements ISchClientCallbackController @Autowired(required = false) private IRmsJobEvent rmsJobLog; + /** + * redisBasedDistributedLock + */ + @Autowired + private RedisBasedDistributedLock redisBasedDistributedLock; + @Override public RestResponse callback(@RequestBody RmsJobResult result) { - if (rmsJobLog != null) { - rmsJobLog.save(null, result); + try { + if (rmsJobLog != null) { + rmsJobLog.save(null, result); + } + return new RestResponse<>(); + } finally { + //获得key + String name = result.getParam().getServiceCode() + "_" + result.getParam().getBeanName(); + //解锁 + redisBasedDistributedLock.unlock(name); } - return new RestResponse<>(); } }