Alawn/udf
1
0
Fork 0
mirror of https://gitee.com/wangkang/udf.git synced 2026-11-01 22:30:00 +00:00
Code Issues Releases Wiki Activity

修改rsmjob,禁止并发模式,的分布式锁机制

Browse Source
This commit is contained in:
wangkang
2017-12-21 09:46:34 +08:00
parent 790255f733
commit 2044733a43
7 changed files with 59 additions and 66 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/domain/RmsJobResult.java
+6
View File
@@ -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;
}
udf-starter-scheduler-client/src/main/java/org/itkk/udf/scheduler/client/web/SchClientController.java
+3 -2
View File
@@ -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);
udf-starter-scheduler/pom.xml
+5
View File
@@ -25,6 +25,11 @@
<groupId>org.itkk.udf</groupId>
<artifactId>udf-starter-rms</artifactId>
</dependency>
<!-- udf-starter-cache-redis -->
<dependency>
<groupId>org.itkk.udf</groupId>
<artifactId>udf-starter-cache-redis</artifactId>
</dependency>
<!-- udf-starter-scheduler-client -->
<dependency>
<groupId>org.itkk.udf</groupId>
udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/IRmsJobEvent.java
-15
View File
@@ -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
*
udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/impl/DefRmsJobEvent.java
-11
View File
@@ -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) {
udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/job/AbstractBaseRmsJob.java
+26 -32
View File
@@ -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);
}
}
/**
* 保存
*
udf-starter-scheduler/src/main/java/org/itkk/udf/scheduler/web/SchClientCallbackController.java
+19 -6
View File
@@ -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<String> 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<>();
}
}