gtsoft/gtsoft-snail-job-server
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects 1 Releases Wiki Activity

feat:(1.2.0-beta2): 优化openapi

Browse Source
This commit is contained in:
opensnail 2024-10-19 19:06:45 +08:00
parent 5f355bb1d8
commit 40d16177dd
18 changed files with 582 additions and 366 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pom.xml
View File

@ -21,7 +21,7 @@
<java.version>17</java.version>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<revision>1.2.0-beta1.1</revision>
<revision>1.2.0-beta2</revision>
<netty-all.version>4.1.94.Final</netty-all.version>
<hutool-all.version>5.8.25</hutool-all.version>
<mybatis-plus.version>3.5.8</mybatis-plus.version>

6
snail-job-client/snail-job-client-common/pom.xml
View File

@ -86,12 +86,6 @@
<artifactId>log4j</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.hibernate</groupId>
<artifactId>hibernate-validator</artifactId>
<version>${hibernate.verion}</version>
</dependency>
</dependencies>
</project>

10
snail-job-client/snail-job-client-job-core/pom.xml
View File

@ -18,6 +18,7 @@
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<hibernate.verion>8.0.1.Final</hibernate.verion>
</properties>
<dependencies>
<dependency>
@ -50,6 +51,15 @@
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</dependency>
<dependency>
<groupId>jakarta.validation</groupId>
<artifactId>jakarta.validation-api</artifactId>
</dependency>
<dependency>
<groupId>org.hibernate</groupId>
<artifactId>hibernate-validator</artifactId>
<version>${hibernate.verion}</version>
</dependency>
</dependencies>
</project>

17
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/AbstractRequestHandler.java
View File

@ -1,5 +1,6 @@
package com.aizuda.snailjob.client.job.core.handler;
import cn.hutool.core.lang.Pair;
import com.aizuda.snailjob.client.common.exception.SnailJobClientException;
/**
@ -16,14 +17,22 @@ public abstract class AbstractRequestHandler<R> implements RequestHandler<R> {
*/
@Override
public R execute() {
if (checkRequest()) {
return doExecute();
Pair<Boolean, String> checked = checkRequest();
if (checked.getKey()) {
beforeExecute();
R r = doExecute();
afterExecute(r);
return r;
} else {
throw new SnailJobClientException("snail job openapi check error");
throw new SnailJobClientException("snail job openapi check error. [{}]", checked.getValue());
}
}
protected abstract void afterExecute(R r);
protected abstract void beforeExecute();
protected abstract R doExecute();
protected abstract boolean checkRequest();
protected abstract Pair<Boolean, String> checkRequest();
}

253
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/RequestAddHandler.java
View File

@ -1,253 +0,0 @@
package com.aizuda.snailjob.client.job.core.handler;
import cn.hutool.core.util.StrUtil;
import com.aizuda.snailjob.client.common.exception.SnailJobClientException;
import com.aizuda.snailjob.client.job.core.dto.RequestAddJobDTO;
import com.aizuda.snailjob.client.job.core.enums.AllocationAlgorithmEnum;
import com.aizuda.snailjob.client.job.core.enums.TriggerTypeEnum;
import com.aizuda.snailjob.client.job.core.util.ValidatorUtils;
import com.aizuda.snailjob.common.core.enums.*;
import com.aizuda.snailjob.common.core.util.JsonUtil;
import com.aizuda.snailjob.common.log.SnailJobLog;
import java.util.HashMap;
import java.util.Map;
public class RequestAddHandler extends AbstractRequestHandler<Long> {
private final RequestAddJobDTO requestAddJobDTO;
public RequestAddHandler(JobTaskTypeEnum taskType, Integer shardNum) {
this.requestAddJobDTO = new RequestAddJobDTO();
// 默认创建就开启
requestAddJobDTO.setJobStatus(StatusEnum.YES.getStatus());
// 设置任务类型
requestAddJobDTO.setTaskType(taskType.getType());
// 默认java
requestAddJobDTO.setExecutorType(ExecutorTypeEnum.JAVA.getType());
// 设置分片
if (shardNum != null) {
Map<String, Object> map = new HashMap<>(1);
map.put(SHARD_NUM, shardNum);
requestAddJobDTO.setArgsStr(JsonUtil.toJsonString(map));
}
}
@Override
protected Long doExecute() {
String data = JsonUtil.toJsonString(client.addJob(requestAddJobDTO).getData());
return Long.valueOf(data);
}
@Override
protected boolean checkRequest() {
boolean validated = ValidatorUtils.validateEntity(requestAddJobDTO);
// 如果校验正确，则正对进行相关筛选
if (validated) {
if (requestAddJobDTO.getTaskType() == JobTaskTypeEnum.CLUSTER.getType()) {
// 集群模式只允许并发为 1
setParallelNum(1);
}
if (requestAddJobDTO.getTriggerType() == TriggerTypeEnum.WORK_FLOW.getType()) {
// 工作流没有调度时间
setTriggerInterval("*");
}
}
return validated;
}
/**
* 设置任务名
*
* @param jobName 任务名
* @return
*/
public RequestAddHandler setJobName(String jobName) {
requestAddJobDTO.setJobName(jobName);
return this;
}
/**
* 设置参数
*
* @param argsStr
* @return
*/
private RequestAddHandler setArgsStr(Map<String, Object> argsStr) {
Map<String, Object> args = new HashMap<>();
if (StrUtil.isNotBlank(requestAddJobDTO.getArgsStr())) {
args = JsonUtil.parseHashMap(requestAddJobDTO.getArgsStr());
}
args.putAll(argsStr);
requestAddJobDTO.setArgsStr(JsonUtil.toJsonString(args));
requestAddJobDTO.setArgsType(JobArgsTypeEnum.JSON.getArgsType());
return this;
}
/**
* 添加参数，可支持多次添加
* 静态分片不可使用该方法
*
* @param argsKey 参数名
* @param argsValue 参数值
* @return
*/
public RequestAddHandler addArgsStr(String argsKey, Object argsValue) {
if (requestAddJobDTO.getTaskType().equals(JobTaskTypeEnum.SHARDING.getType())) {
SnailJobLog.LOCAL.warn("静态分片任务，不可使用该方法添加相关任务参数，请使用addShardingArgs");
return this;
}
Map<String, Object> map = new HashMap<>();
if (StrUtil.isNotBlank(requestAddJobDTO.getArgsStr())) {
map = JsonUtil.parseHashMap(requestAddJobDTO.getArgsStr());
}
map.put(argsKey, argsValue);
requestAddJobDTO.setArgsStr(JsonUtil.toJsonString(map));
requestAddJobDTO.setArgsType(JobArgsTypeEnum.JSON.getArgsType());
return this;
}
/**
* 添加静态分片相关参数
*
* @param shardingValue
* @return
*/
public RequestAddHandler addShardingArgs(String[] shardingValue) {
if (!requestAddJobDTO.getTaskType().equals(JobTaskTypeEnum.SHARDING.getType())) {
SnailJobLog.LOCAL.warn("非静态分片任务，不可使用该方法添加相关任务参数，请使用addArgsStr");
return this;
}
requestAddJobDTO.setArgsStr(JsonUtil.toJsonString(shardingValue));
requestAddJobDTO.setArgsType(JobArgsTypeEnum.TEXT.getArgsType());
return this;
}
/**
* 设置路由
*
* @param algorithmEnum 路由算法
* @return
*/
public RequestAddHandler setRouteKey(AllocationAlgorithmEnum algorithmEnum) {
// 非集群模式 路由策略只能为轮询
if (requestAddJobDTO.getTaskType() != JobTaskTypeEnum.CLUSTER.getType()) {
setRouteKey(AllocationAlgorithmEnum.ROUND);
SnailJobLog.LOCAL.warn("非集群模式 路由策略只能为轮询");
return this;
}
requestAddJobDTO.setRouteKey(algorithmEnum.getType());
return this;
}
/**
* 设置执行器信息
*
* @param executorInfo
* @return
*/
public RequestAddHandler setExecutorInfo(String executorInfo) {
requestAddJobDTO.setExecutorInfo(executorInfo);
return this;
}
/**
* 设置调度类型
*
* @param triggerType
* @return
*/
public RequestAddHandler setTriggerType(TriggerTypeEnum triggerType) {
requestAddJobDTO.setTriggerType(triggerType.getType());
if (requestAddJobDTO.getTriggerType() == TriggerTypeEnum.WORK_FLOW.getType()) {
// 工作流没有调度时间
setTriggerInterval("*");
}
return this;
}
/**
* 设置触发间隔；
* 单位：秒
* 工作流无需配置
*
* @param triggerInterval
* @return
*/
public RequestAddHandler setTriggerInterval(String triggerInterval) {
requestAddJobDTO.setTriggerInterval(triggerInterval);
return this;
}
/**
* 设置阻塞策略
*
* @param blockStrategy
* @return
*/
public RequestAddHandler setBlockStrategy(BlockStrategyEnum blockStrategy) {
// 非集群模式 路由策略只能为轮询
if (requestAddJobDTO.getTaskType() == JobTaskTypeEnum.CLUSTER.getType()
&& blockStrategy.getBlockStrategy() == BlockStrategyEnum.CONCURRENCY.getBlockStrategy()) {
throw new SnailJobClientException("集群模式不能使用并行阻塞策略");
}
requestAddJobDTO.setBlockStrategy(blockStrategy.getBlockStrategy());
return this;
}
/**
* 设置执行器超时时间
*
* @param executorTimeout
* @return
*/
public RequestAddHandler setExecutorTimeout(Integer executorTimeout) {
requestAddJobDTO.setExecutorTimeout(executorTimeout);
return this;
}
/**
* 设置任务最大重试次数
*
* @param maxRetryTimes
* @return
*/
public RequestAddHandler setMaxRetryTimes(Integer maxRetryTimes) {
requestAddJobDTO.setMaxRetryTimes(maxRetryTimes);
return this;
}
/**
* 设置重试间隔
*
* @param retryInterval
* @return
*/
public RequestAddHandler setRetryInterval(Integer retryInterval) {
requestAddJobDTO.setRetryInterval(retryInterval);
return this;
}
/**
* 设置并发数量
*
* @param parallelNum
* @return
*/
public RequestAddHandler setParallelNum(Integer parallelNum) {
requestAddJobDTO.setParallelNum(parallelNum);
return this;
}
/**
* 设置定时任务描述
*
* @param description
* @return
*/
public RequestAddHandler setDescription(String description) {
requestAddJobDTO.setDescription(description);
return this;
}
}

221
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/add/AddHandler.java Normal file
View File

@ -0,0 +1,221 @@
package com.aizuda.snailjob.client.job.core.handler.add;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.lang.Pair;
import cn.hutool.core.util.StrUtil;
import com.aizuda.snailjob.client.common.exception.SnailJobClientException;
import com.aizuda.snailjob.client.job.core.dto.RequestAddJobDTO;
import com.aizuda.snailjob.client.job.core.enums.AllocationAlgorithmEnum;
import com.aizuda.snailjob.client.job.core.enums.TriggerTypeEnum;
import com.aizuda.snailjob.client.job.core.handler.AbstractRequestHandler;
import com.aizuda.snailjob.client.job.core.util.ValidatorUtils;
import com.aizuda.snailjob.common.core.enums.*;
import com.aizuda.snailjob.common.core.util.JsonUtil;
import lombok.Setter;
import java.util.HashMap;
import java.util.Map;
import static com.aizuda.snailjob.client.job.core.enums.TriggerTypeEnum.WORK_FLOW;
public abstract class AddHandler<R> extends AbstractRequestHandler<Long> {
private final RequestAddJobDTO reqDTO;
@Setter
private R r;
public AddHandler(JobTaskTypeEnum taskType, Integer shardNum) {
this.reqDTO = new RequestAddJobDTO();
// 默认创建就开启
reqDTO.setJobStatus(StatusEnum.YES.getStatus());
// 设置任务类型
reqDTO.setTaskType(taskType.getType());
// 默认java
reqDTO.setExecutorType(ExecutorTypeEnum.JAVA.getType());
// 设置分片
if (shardNum != null) {
Map<String, Object> map = new HashMap<>(1);
map.put(SHARD_NUM, shardNum);
reqDTO.setArgsStr(JsonUtil.toJsonString(map));
}
}
@Override
protected Long doExecute() {
String data = JsonUtil.toJsonString(client.addJob(reqDTO).getData());
return Long.valueOf(data);
}
@Override
protected void beforeExecute() {
// 此次是兜底覆盖,工作流是没有调度时间
if (reqDTO.getTriggerType() == WORK_FLOW.getType()) {
setTriggerInterval("*");
}
}
@Override
protected void afterExecute(Long aLong) {
}
@Override
protected Pair<Boolean, String> checkRequest() {
return ValidatorUtils.validateEntity(reqDTO);
}
/**
* 设置任务名
*
* @param jobName 任务名
* @return
*/
public R setJobName(String jobName) {
reqDTO.setJobName(jobName);
return r;
}
/**
* 添加参数，可支持多次添加
* 静态分片不可使用该方法
*
* @param argsKey 参数名
* @param argsValue 参数值
* @return
*/
protected R addArgsStr(String argsKey, Object argsValue) {
Map<String, Object> map = new HashMap<>();
if (StrUtil.isNotBlank(reqDTO.getArgsStr())) {
map = JsonUtil.parseHashMap(reqDTO.getArgsStr());
}
map.put(argsKey, argsValue);
reqDTO.setArgsStr(JsonUtil.toJsonString(map));
reqDTO.setArgsType(JobArgsTypeEnum.JSON.getArgsType());
return r;
}
/**
* 添加静态分片相关参数
*
* @param shardingValue 分片参数
* @return r
*/
protected R addShardingArgs(String ...shardingValue) {
reqDTO.setArgsStr(JsonUtil.toJsonString(shardingValue));
reqDTO.setArgsType(JobArgsTypeEnum.TEXT.getArgsType());
return r;
}
/**
* 设置路由
*
* @param algorithmEnum 路由算法
* @return r
*/
protected R setRouteKey(AllocationAlgorithmEnum algorithmEnum) {
reqDTO.setRouteKey(algorithmEnum.getType());
return r;
}
/**
* 设置执行器信息
*
* @param executorInfo 执行器信息
* @return r
*/
public R setExecutorInfo(String executorInfo) {
reqDTO.setExecutorInfo(executorInfo);
return r;
}
/**
* 设置调度类型
*
* @param triggerType 触发类型
* @return r
*/
public R setTriggerType(TriggerTypeEnum triggerType) {
reqDTO.setTriggerType(triggerType.getType());
return r;
}
/**
* 设置触发间隔；
* 单位：秒
* 工作流无需配置
*
* @param triggerInterval 触发间隔
* @return r
*/
public R setTriggerInterval(String triggerInterval) {
// 若是工作流则没有调度时间
Assert.isTrue(reqDTO.getTriggerType() == WORK_FLOW.getType(),
() -> new SnailJobClientException("工作流无需配置"));
reqDTO.setTriggerInterval(triggerInterval);
return r;
}
/**
* 设置阻塞策略
*
* @param blockStrategy 阻塞策略
* @return r
*/
public R setBlockStrategy(BlockStrategyEnum blockStrategy) {
reqDTO.setBlockStrategy(blockStrategy.getBlockStrategy());
return r;
}
/**
* 设置执行器超时时间
*
* @param executorTimeout 超时时间(单位:秒)
* @return r
*/
public R setExecutorTimeout(Integer executorTimeout) {
reqDTO.setExecutorTimeout(executorTimeout);
return r;
}
/**
* 设置任务最大重试次数
*
* @param maxRetryTimes 最大超时时间
* @return r
*/
public R setMaxRetryTimes(Integer maxRetryTimes) {
reqDTO.setMaxRetryTimes(maxRetryTimes);
return r;
}
/**
* 设置重试间隔
*
* @param retryInterval 重试间隔
* @return r
*/
public R setRetryInterval(Integer retryInterval) {
reqDTO.setRetryInterval(retryInterval);
return r;
}
/**
* 设置并发数量
*
* @param parallelNum 并发数量
* @return r
*/
protected R setParallelNum(Integer parallelNum) {
reqDTO.setParallelNum(parallelNum);
return r;
}
/**
* 设置定时任务描述
*
* @param description 任务描述
* @return r
*/
public R setDescription(String description) {
reqDTO.setDescription(description);
return r;
}
}

30
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/add/BroadcastAddHandler.java Normal file
View File

@ -0,0 +1,30 @@
package com.aizuda.snailjob.client.job.core.handler.add;
import com.aizuda.snailjob.client.job.core.enums.AllocationAlgorithmEnum;
import com.aizuda.snailjob.common.core.enums.JobTaskTypeEnum;
/**
* @author opensnail
* @date 2024-10-19 12:25:49
* @since sj_1.1.0
*/
public class BroadcastAddHandler extends AddHandler<BroadcastAddHandler> {
public BroadcastAddHandler() {
this(JobTaskTypeEnum.CLUSTER, null);
}
public BroadcastAddHandler(JobTaskTypeEnum taskType, Integer shardNum) {
super(taskType, shardNum);
// 广播模式只允许并发为 1
setParallelNum(1);
// 广播模式采用轮询模式
setRouteKey(AllocationAlgorithmEnum.ROUND);
setR(this);
}
@Override
public BroadcastAddHandler addArgsStr(String argsKey, Object argsValue) {
return super.addArgsStr(argsKey, argsValue);
}
}

34
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/add/ClusterAddHandler.java Normal file
View File

@ -0,0 +1,34 @@
package com.aizuda.snailjob.client.job.core.handler.add;
import com.aizuda.snailjob.client.job.core.enums.AllocationAlgorithmEnum;
import com.aizuda.snailjob.common.core.enums.JobTaskTypeEnum;
/**
* @author opensnail
* @date 2024-10-19 12:25:49
* @since sj_1.2.0
*/
public class ClusterAddHandler extends AddHandler<ClusterAddHandler> {
public ClusterAddHandler(JobTaskTypeEnum taskType) {
this(taskType, null);
}
public ClusterAddHandler(JobTaskTypeEnum taskType, Integer shardNum) {
super(taskType, shardNum);
// 集群模式只允许并发为 1
setParallelNum(1);
setR(this);
}
@Override
public ClusterAddHandler setRouteKey(AllocationAlgorithmEnum algorithmEnum) {
super.setRouteKey(algorithmEnum);
return this;
}
@Override
public ClusterAddHandler addArgsStr(String argsKey, Object argsValue) {
return super.addArgsStr(argsKey, argsValue);
}
}

32
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/add/MapAddHandler.java Normal file
View File

@ -0,0 +1,32 @@
package com.aizuda.snailjob.client.job.core.handler.add;
import com.aizuda.snailjob.client.job.core.enums.AllocationAlgorithmEnum;
import com.aizuda.snailjob.common.core.enums.JobTaskTypeEnum;
/**
* @author opensnail
* @date 2024-10-19 12:25:49
* @since sj_1.2.0
*/
public class MapAddHandler extends AddHandler<MapAddHandler> {
public MapAddHandler() {
this(JobTaskTypeEnum.MAP, null);
}
public MapAddHandler(JobTaskTypeEnum taskType, Integer shardNum) {
super(taskType, shardNum);
setRouteKey(AllocationAlgorithmEnum.ROUND);
setR(this);
}
@Override
public MapAddHandler addShardingArgs(String... shardingValue) {
return super.addShardingArgs(shardingValue);
}
@Override
public MapAddHandler setParallelNum(Integer parallelNum) {
return super.setParallelNum(parallelNum);
}
}

32
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/add/MapReduceAddHandler.java Normal file
View File

@ -0,0 +1,32 @@
package com.aizuda.snailjob.client.job.core.handler.add;
import com.aizuda.snailjob.client.job.core.enums.AllocationAlgorithmEnum;
import com.aizuda.snailjob.common.core.enums.JobTaskTypeEnum;
/**
* @author opensnail
* @date 2024-10-19 12:25:49
* @since sj_1.2.0
*/
public class MapReduceAddHandler extends AddHandler<MapReduceAddHandler> {
public MapReduceAddHandler(Integer shardNum) {
this(JobTaskTypeEnum.MAP_REDUCE, shardNum);
}
public MapReduceAddHandler(JobTaskTypeEnum taskType, Integer shardNum) {
super(taskType, shardNum);
setRouteKey(AllocationAlgorithmEnum.ROUND);
setR(this);
}
@Override
public MapReduceAddHandler addShardingArgs(String... shardingValue) {
return super.addShardingArgs(shardingValue);
}
@Override
public MapReduceAddHandler setParallelNum(Integer parallelNum) {
return super.setParallelNum(parallelNum);
}
}

32
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/add/ShardingAddHandler.java Normal file
View File

@ -0,0 +1,32 @@
package com.aizuda.snailjob.client.job.core.handler.add;
import com.aizuda.snailjob.client.job.core.enums.AllocationAlgorithmEnum;
import com.aizuda.snailjob.common.core.enums.JobTaskTypeEnum;
/**
* @author opensnail
* @date 2024-10-19 12:25:49
* @since sj_1.2.0
*/
public class ShardingAddHandler extends AddHandler<ShardingAddHandler> {
public ShardingAddHandler() {
this(JobTaskTypeEnum.SHARDING, null);
}
public ShardingAddHandler(JobTaskTypeEnum taskType, Integer shardNum) {
super(taskType, shardNum);
setRouteKey(AllocationAlgorithmEnum.ROUND);
setR(this);
}
@Override
public ShardingAddHandler addShardingArgs(String... shardingValue) {
return super.addShardingArgs(shardingValue);
}
@Override
public ShardingAddHandler setParallelNum(Integer parallelNum) {
return super.setParallelNum(parallelNum);
}
}

20
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/RequestQueryHandler.java → snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/quert/RequestQueryHandler.java
View File

@ -1,9 +1,11 @@
package com.aizuda.snailjob.client.job.core.handler;
package com.aizuda.snailjob.client.job.core.handler.quert;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.lang.Pair;
import com.aizuda.snailjob.client.common.exception.SnailJobClientException;
import com.aizuda.snailjob.client.job.core.dto.JobResponseVO;
import com.aizuda.snailjob.client.job.core.handler.AbstractRequestHandler;
import com.aizuda.snailjob.common.core.util.JsonUtil;
import java.util.Objects;
@ -15,16 +17,26 @@ public class RequestQueryHandler extends AbstractRequestHandler<JobResponseVO> {
this.queryJobId = queryJobId;
}
@Override
protected void afterExecute(JobResponseVO jobResponseVO) {
}
@Override
protected void beforeExecute() {
}
@Override
protected JobResponseVO doExecute() {
Object data = client.getJobDetail(queryJobId).getData();
Assert.isTrue(Objects.nonNull(data),()-> new SnailJobClientException("获取[{}]任务详情失败", queryJobId));
Assert.isTrue(Objects.nonNull(data), () -> new SnailJobClientException("获取[{}]任务详情失败", queryJobId));
return JsonUtil.parseObject(JsonUtil.toJsonString(data), JobResponseVO.class);
}
@Override
protected boolean checkRequest() {
return queryJobId != null && ! Long.valueOf(0).equals(queryJobId);
protected Pair<Boolean, String> checkRequest() {
return Pair.of(queryJobId != null && !Long.valueOf(0).equals(queryJobId), "queryJobId不能为null并且必须大于0");
}
}

20
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/RequestTriggerJobHandler.java → snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/trigger/RequestTriggerJobHandler.java
View File

@ -1,9 +1,11 @@
package com.aizuda.snailjob.client.job.core.handler;
package com.aizuda.snailjob.client.job.core.handler.trigger;
import cn.hutool.core.lang.Pair;
import com.aizuda.snailjob.client.common.exception.SnailJobClientException;
import com.aizuda.snailjob.client.job.core.enums.JobTypeEnum;
import com.aizuda.snailjob.client.job.core.handler.AbstractRequestHandler;
public class RequestTriggerJobHandler extends AbstractRequestHandler<Boolean>{
public class RequestTriggerJobHandler extends AbstractRequestHandler<Boolean> {
private final Long triggerJobId;
// 1: job; 2: workflow
private final int triggerType;
@ -13,6 +15,16 @@ public class RequestTriggerJobHandler extends AbstractRequestHandler<Boolean>{
this.triggerType = triggerType;
}
@Override
protected void afterExecute(Boolean aBoolean) {
}
@Override
protected void beforeExecute() {
}
@Override
protected Boolean doExecute() {
if (triggerType == JobTypeEnum.JOB.getType()) {
@ -25,7 +37,7 @@ public class RequestTriggerJobHandler extends AbstractRequestHandler<Boolean>{
}
@Override
protected boolean checkRequest() {
return triggerJobId != null && !Long.valueOf(0).equals(triggerJobId);
protected Pair<Boolean, String> checkRequest() {
return Pair.of(triggerJobId != null && !Long.valueOf(0).equals(triggerJobId), "triggerJobId不能为null并且必须大于0");
}
}

138
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/RequestUpdateHandler.java → snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/update/RequestUpdateHandler.java
View File

@ -1,10 +1,12 @@
package com.aizuda.snailjob.client.job.core.handler;
package com.aizuda.snailjob.client.job.core.handler.update;
import cn.hutool.core.lang.Pair;
import cn.hutool.core.util.StrUtil;
import com.aizuda.snailjob.client.common.exception.SnailJobClientException;
import com.aizuda.snailjob.client.job.core.dto.RequestUpdateJobDTO;
import com.aizuda.snailjob.client.job.core.enums.AllocationAlgorithmEnum;
import com.aizuda.snailjob.client.job.core.enums.TriggerTypeEnum;
import com.aizuda.snailjob.client.job.core.handler.AbstractRequestHandler;
import com.aizuda.snailjob.client.job.core.util.ValidatorUtils;
import com.aizuda.snailjob.common.core.enums.BlockStrategyEnum;
import com.aizuda.snailjob.common.core.enums.ExecutorTypeEnum;
@ -15,63 +17,81 @@ import com.aizuda.snailjob.common.log.SnailJobLog;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import static com.aizuda.snailjob.common.core.enums.JobTaskTypeEnum.CLUSTER;
public class RequestUpdateHandler extends AbstractRequestHandler<Boolean> {
private final RequestUpdateJobDTO requestUpdateJobDTO;
private final RequestUpdateJobDTO reqDTO;
public RequestUpdateHandler(Long jobId) {
this.requestUpdateJobDTO = new RequestUpdateJobDTO();
this.reqDTO = new RequestUpdateJobDTO();
// 更新必须要id
requestUpdateJobDTO.setId(jobId);
reqDTO.setId(jobId);
// 默认java
requestUpdateJobDTO.setExecutorType(ExecutorTypeEnum.JAVA.getType());
reqDTO.setExecutorType(ExecutorTypeEnum.JAVA.getType());
}
@Override
protected void afterExecute(Boolean aBoolean) {
}
@Override
protected void beforeExecute() {
}
@Override
protected Boolean doExecute() {
return (Boolean) client.updateJob(requestUpdateJobDTO).getData();
return (Boolean) client.updateJob(reqDTO).getData();
}
@Override
protected boolean checkRequest() {
boolean validated = ValidatorUtils.validateEntity(requestUpdateJobDTO);
// 如果校验正确，则正对进行相关筛选
if (validated) {
if (requestUpdateJobDTO.getTaskType() != null
&& requestUpdateJobDTO.getTaskType() == JobTaskTypeEnum.CLUSTER.getType()){
protected Pair<Boolean, String> checkRequest() {
Pair<Boolean, String> validated = ValidatorUtils.validateEntity(reqDTO);
if (!validated.getKey()) {
return validated;
}
// 如果校验正确，下面则进行相关参数填充
Optional.ofNullable(reqDTO.getTaskType()).ifPresent(taskType -> {
if (reqDTO.getTaskType() == CLUSTER.getType()) {
// 集群模式只允许并发为 1
setParallelNum(1);
}
// 非集群模式 路由策略只能为轮询
if (requestUpdateJobDTO.getTaskType() != null
&& requestUpdateJobDTO.getTaskType() != JobTaskTypeEnum.CLUSTER.getType()){
} else {
// 非集群模式 路由策略只能为轮询
setRouteKey(AllocationAlgorithmEnum.ROUND);
}
if (requestUpdateJobDTO.getTriggerType() != null
&& requestUpdateJobDTO.getTriggerType() == TriggerTypeEnum.WORK_FLOW.getType()){
});
Optional.ofNullable(reqDTO.getTriggerType()).ifPresent((triggerType) -> {
if (reqDTO.getTriggerType() == TriggerTypeEnum.WORK_FLOW.getType()) {
// 工作流没有调度时间
setTriggerInterval("*");
}
}
});
return validated;
}
/**
* 修改Reduce的分片数
* 只允许MAP_REDUCE设置
*
* @param shardNum
* @return
*/
public RequestUpdateHandler setShardNum(Integer shardNum){
Integer taskType = requestUpdateJobDTO.getTaskType();
if (taskType != null && taskType.equals(JobTaskTypeEnum.MAP_REDUCE.getType())){
public RequestUpdateHandler setShardNum(Integer shardNum) {
Integer taskType = reqDTO.getTaskType();
if (taskType != null && taskType.equals(JobTaskTypeEnum.MAP_REDUCE.getType())) {
// 设置分片
if (shardNum != null){
if (shardNum != null) {
Map<String, Object> map = new HashMap<>(1);
map.put(SHARD_NUM, shardNum);
requestUpdateJobDTO.setArgsStr(JsonUtil.toJsonString(map));
reqDTO.setArgsStr(JsonUtil.toJsonString(map));
}
}else {
} else {
throw new SnailJobClientException("非MapReduce模式不能设置分片数");
}
return this;
@ -79,28 +99,30 @@ public class RequestUpdateHandler extends AbstractRequestHandler<Boolean> {
/**
* 修改任务名称
*
* @param jobName
* @return
*/
public RequestUpdateHandler setJobName(String jobName) {
requestUpdateJobDTO.setJobName(jobName);
reqDTO.setJobName(jobName);
return this;
}
/**
* 修改时会直接覆盖之前的任务参数
* 修改参数
*
* @param argsStr
* @return
*/
private RequestUpdateHandler setArgsStr(Map<String, Object> argsStr) {
Map<String, Object> args = new HashMap<>();
if (StrUtil.isNotBlank(requestUpdateJobDTO.getArgsStr())){
args = JsonUtil.parseHashMap(requestUpdateJobDTO.getArgsStr());
if (StrUtil.isNotBlank(reqDTO.getArgsStr())) {
args = JsonUtil.parseHashMap(reqDTO.getArgsStr());
}
args.putAll(argsStr);
requestUpdateJobDTO.setArgsStr(JsonUtil.toJsonString(args));
requestUpdateJobDTO.setArgsType(JobArgsTypeEnum.JSON.getArgsType());
reqDTO.setArgsStr(JsonUtil.toJsonString(args));
reqDTO.setArgsType(JobArgsTypeEnum.JSON.getArgsType());
return this;
}
@ -108,69 +130,74 @@ public class RequestUpdateHandler extends AbstractRequestHandler<Boolean> {
* 修改时会直接覆盖之前的任务参数
* 添加参数，可支持多次添加
* 静态分片不可使用该方法
* @param argsKey 参数名
*
* @param argsKey 参数名
* @param argsValue 参数值
* @return
*/
public RequestUpdateHandler addArgsStr(String argsKey, Object argsValue) {
if (requestUpdateJobDTO.getTaskType() != null
&& requestUpdateJobDTO.getTaskType().equals(JobTaskTypeEnum.SHARDING.getType())){
if (reqDTO.getTaskType() != null
&& reqDTO.getTaskType().equals(JobTaskTypeEnum.SHARDING.getType())) {
SnailJobLog.LOCAL.warn("静态分片任务，不可使用该方法添加相关任务参数，请使用addShardingArgs");
return this;
}
Map<String, Object> map = new HashMap<>();
if (StrUtil.isNotBlank(requestUpdateJobDTO.getArgsStr())){
map = JsonUtil.parseHashMap(requestUpdateJobDTO.getArgsStr());
if (StrUtil.isNotBlank(reqDTO.getArgsStr())) {
map = JsonUtil.parseHashMap(reqDTO.getArgsStr());
}
map.put(argsKey, argsValue);
requestUpdateJobDTO.setArgsStr(JsonUtil.toJsonString(map));
requestUpdateJobDTO.setArgsType(JobArgsTypeEnum.JSON.getArgsType());
reqDTO.setArgsStr(JsonUtil.toJsonString(map));
reqDTO.setArgsType(JobArgsTypeEnum.JSON.getArgsType());
return this;
}
/**
* 添加静态分片相关参数
*
* @param shardingValue
* @return
*/
public RequestUpdateHandler addShardingArgs(String[] shardingValue){
if (requestUpdateJobDTO.getTaskType() != null
&& !requestUpdateJobDTO.getTaskType().equals(JobTaskTypeEnum.SHARDING.getType())){
public RequestUpdateHandler addShardingArgs(String[] shardingValue) {
if (reqDTO.getTaskType() != null
&& !reqDTO.getTaskType().equals(JobTaskTypeEnum.SHARDING.getType())) {
SnailJobLog.LOCAL.warn("非静态分片任务，不可使用该方法添加相关任务参数，请使用addArgsStr");
return this;
}
requestUpdateJobDTO.setArgsStr(JsonUtil.toJsonString(shardingValue));
requestUpdateJobDTO.setArgsType(JobArgsTypeEnum.TEXT.getArgsType());
reqDTO.setArgsStr(JsonUtil.toJsonString(shardingValue));
reqDTO.setArgsType(JobArgsTypeEnum.TEXT.getArgsType());
return this;
}
/**
* 修改路由
*
* @param algorithmEnum
* @return
*/
public RequestUpdateHandler setRouteKey(AllocationAlgorithmEnum algorithmEnum) {
requestUpdateJobDTO.setRouteKey(algorithmEnum.getType());
reqDTO.setRouteKey(algorithmEnum.getType());
return this;
}
/**
* 修改相关执行器
*
* @param executorInfo
* @return
*/
public RequestUpdateHandler setExecutorInfo(String executorInfo) {
requestUpdateJobDTO.setExecutorInfo(executorInfo);
reqDTO.setExecutorInfo(executorInfo);
return this;
}
/**
* 修改调度类型
*
* @param triggerType
* @return
*/
public RequestUpdateHandler setTriggerType(TriggerTypeEnum triggerType) {
requestUpdateJobDTO.setTriggerType(triggerType.getType());
reqDTO.setTriggerType(triggerType.getType());
return this;
}
@ -178,71 +205,78 @@ public class RequestUpdateHandler extends AbstractRequestHandler<Boolean> {
* 修改调度时间
* 单位：秒
* 工作流无需配置
*
* @param triggerInterval
* @return
*/
public RequestUpdateHandler setTriggerInterval(String triggerInterval) {
requestUpdateJobDTO.setTriggerInterval(triggerInterval);
reqDTO.setTriggerInterval(triggerInterval);
return this;
}
/**
* 修改阻塞策略
*
* @param blockStrategy
* @return
*/
public RequestUpdateHandler setBlockStrategy(BlockStrategyEnum blockStrategy) {
requestUpdateJobDTO.setBlockStrategy(blockStrategy.getBlockStrategy());
reqDTO.setBlockStrategy(blockStrategy.getBlockStrategy());
return this;
}
/**
* 修改执行器超时时间
*
* @param executorTimeout
* @return
*/
public RequestUpdateHandler setExecutorTimeout(Integer executorTimeout) {
requestUpdateJobDTO.setExecutorTimeout(executorTimeout);
reqDTO.setExecutorTimeout(executorTimeout);
return this;
}
/**
* 修改任务最大重试次数
*
* @param maxRetryTimes
* @return
*/
public RequestUpdateHandler setMaxRetryTimes(Integer maxRetryTimes) {
requestUpdateJobDTO.setMaxRetryTimes(maxRetryTimes);
reqDTO.setMaxRetryTimes(maxRetryTimes);
return this;
}
/**
* 修改重试间隔
*
* @param retryInterval
* @return
*/
public RequestUpdateHandler setRetryInterval(Integer retryInterval) {
requestUpdateJobDTO.setRetryInterval(retryInterval);
reqDTO.setRetryInterval(retryInterval);
return this;
}
/**
* 修改并发数量
*
* @param parallelNum
* @return
*/
public RequestUpdateHandler setParallelNum(Integer parallelNum) {
requestUpdateJobDTO.setParallelNum(parallelNum);
reqDTO.setParallelNum(parallelNum);
return this;
}
/**
* 修改定时任务描述
*
* @param description
* @return
*/
public RequestUpdateHandler setDescription(String description) {
requestUpdateJobDTO.setDescription(description);
reqDTO.setDescription(description);
return this;
}

30
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/RequestUpdateStatusHandler.java → snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/handler/update/RequestUpdateStatusHandler.java
View File

@ -1,13 +1,15 @@
package com.aizuda.snailjob.client.job.core.handler;
package com.aizuda.snailjob.client.job.core.handler.update;
import cn.hutool.core.lang.Pair;
import com.aizuda.snailjob.client.common.exception.SnailJobClientException;
import com.aizuda.snailjob.client.job.core.dto.RequestUpdateStatusDTO;
import com.aizuda.snailjob.client.job.core.enums.JobTypeEnum;
import com.aizuda.snailjob.client.job.core.handler.AbstractRequestHandler;
import com.aizuda.snailjob.client.job.core.util.ValidatorUtils;
import com.aizuda.snailjob.common.core.enums.StatusEnum;
public class RequestUpdateStatusHandler extends AbstractRequestHandler<Boolean>{
public class RequestUpdateStatusHandler extends AbstractRequestHandler<Boolean> {
private final RequestUpdateStatusDTO statusDTO;
// 1: job; 2: workflow
private final int type;
@ -18,38 +20,50 @@ public class RequestUpdateStatusHandler extends AbstractRequestHandler<Boolean>{
setId(id);
}
@Override
protected void afterExecute(Boolean aBoolean) {
}
@Override
protected void beforeExecute() {
}
@Override
protected Boolean doExecute() {
if (type == JobTypeEnum.JOB.getType()){
if (type == JobTypeEnum.JOB.getType()) {
return (Boolean) client.updateJobStatus(statusDTO).getData();
}
if (type == JobTypeEnum.WORKFLOW.getType()){
if (type == JobTypeEnum.WORKFLOW.getType()) {
return (Boolean) client.updateWorkFlowStatus(statusDTO).getData();
}
throw new SnailJobClientException("snail job openapi check error");
}
@Override
protected boolean checkRequest() {
return ValidatorUtils.validateEntity(statusDTO);
protected Pair<Boolean, String> checkRequest() {
return ValidatorUtils.validateEntity(statusDTO);
}
/**
* 设置任务/工作流ID
*
* @param id
* @return
*/
private RequestUpdateStatusHandler setId(Long id){
private RequestUpdateStatusHandler setId(Long id) {
this.statusDTO.setId(id);
return this;
}
/**
* 设置状态
*
* @param status
* @return
*/
public RequestUpdateStatusHandler setStatus(StatusEnum status){
public RequestUpdateStatusHandler setStatus(StatusEnum status) {
this.statusDTO.setJobStatus(status.getStatus());
return this;
}

48
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/openapi/SnailJobOpenApi.java
View File

@ -1,7 +1,11 @@
package com.aizuda.snailjob.client.job.core.openapi;
import com.aizuda.snailjob.client.job.core.enums.JobTypeEnum;
import com.aizuda.snailjob.client.job.core.handler.*;
import com.aizuda.snailjob.client.job.core.handler.add.*;
import com.aizuda.snailjob.client.job.core.handler.quert.RequestQueryHandler;
import com.aizuda.snailjob.client.job.core.handler.trigger.RequestTriggerJobHandler;
import com.aizuda.snailjob.client.job.core.handler.update.RequestUpdateHandler;
import com.aizuda.snailjob.client.job.core.handler.update.RequestUpdateStatusHandler;
import com.aizuda.snailjob.common.core.enums.JobTaskTypeEnum;
/**
@ -16,54 +20,54 @@ public final class SnailJobOpenApi {
/**
* 添加集群定时任务
*
* @return
* @return {@link ClusterAddHandler}
*/
public static RequestAddHandler addClusterJob() {
return new RequestAddHandler(JobTaskTypeEnum.CLUSTER, null);
public static ClusterAddHandler addClusterJob() {
return new ClusterAddHandler(JobTaskTypeEnum.CLUSTER);
}
/**
* 添加广播定时任务
*
* @return
* @return {@link BroadcastAddHandler}
*/
public static RequestAddHandler addBroadcastJob() {
return new RequestAddHandler(JobTaskTypeEnum.BROADCAST, null);
public static BroadcastAddHandler addBroadcastJob() {
return new BroadcastAddHandler();
}
/**
* 添加固定分片定时任务
*
* @return
* @return {@link ShardingAddHandler}
*/
public static RequestAddHandler addShardingJob() {
return new RequestAddHandler(JobTaskTypeEnum.SHARDING, null);
public static ShardingAddHandler addShardingJob() {
return new ShardingAddHandler();
}
/**
* 添加Map定时任务
*
* @return
* @return {@link MapAddHandler}
*/
public static RequestAddHandler addMapJob() {
return new RequestAddHandler(JobTaskTypeEnum.MAP, null);
public static MapAddHandler addMapJob() {
return new MapAddHandler();
}
/**
* 添加MapReduce定时任务
*
* @param shardNum Reduce数量
* @return
* @return {@link MapReduceAddHandler}
*/
public static RequestAddHandler addMapReduceJob(Integer shardNum) {
return new RequestAddHandler(JobTaskTypeEnum.MAP_REDUCE, shardNum);
public static MapReduceAddHandler addMapReduceJob(Integer shardNum) {
return new MapReduceAddHandler(shardNum);
}
/**
* 更新定时任务
*
* @param jobId 定时任务ID
* @return
* @return {@link RequestUpdateHandler}
*/
public static RequestUpdateHandler updateJob(Long jobId) {
return new RequestUpdateHandler(jobId);
@ -73,7 +77,7 @@ public final class SnailJobOpenApi {
* 获取定时任务详情
*
* @param jobId 定时任务ID
* @return
* @return {@link RequestQueryHandler}
*/
public static RequestQueryHandler getJobDetail(Long jobId) {
return new RequestQueryHandler(jobId);
@ -83,7 +87,7 @@ public final class SnailJobOpenApi {
* 手动触发定时任务
*
* @param jobId 定时任务ID
* @return
* @return {@link RequestTriggerJobHandler}
*/
public static RequestTriggerJobHandler triggerJob(Long jobId) {
return new RequestTriggerJobHandler(jobId, JobTypeEnum.JOB.getType());
@ -93,7 +97,7 @@ public final class SnailJobOpenApi {
* 手动触发工作流任务
*
* @param id 工作流任务ID
* @return
* @return {@link RequestTriggerJobHandler}
*/
public static RequestTriggerJobHandler triggerWorkFlow(Long id) {
return new RequestTriggerJobHandler(id, JobTypeEnum.WORKFLOW.getType());
@ -103,7 +107,7 @@ public final class SnailJobOpenApi {
* 更新定时任务状态
*
* @param jobId 任务ID
* @return
* @return {@link RequestUpdateStatusHandler}
*/
public static RequestUpdateStatusHandler updateJobStatus(Long jobId) {
return new RequestUpdateStatusHandler(jobId, JobTypeEnum.JOB.getType());
@ -113,7 +117,7 @@ public final class SnailJobOpenApi {
* 更新工作流任务状态
*
* @param workFlowId 工作流ID
* @return
* @return {@link RequestUpdateStatusHandler}
*/
public static RequestUpdateStatusHandler updateWorkFlowStatus(Long workFlowId) {
return new RequestUpdateStatusHandler(workFlowId, JobTypeEnum.WORKFLOW.getType());

21
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/util/ValidatorUtils.java
View File

@ -1,7 +1,7 @@
package com.aizuda.snailjob.client.job.core.util;
import cn.hutool.core.lang.Pair;
import com.aizuda.snailjob.client.common.exception.SnailJobClientException;
import com.aizuda.snailjob.common.log.SnailJobLog;
import jakarta.validation.ConstraintViolation;
import jakarta.validation.Validation;
import jakarta.validation.Validator;
@ -9,7 +9,7 @@ import jakarta.validation.Validator;
import java.util.Set;
public class ValidatorUtils {
private static Validator validator;
private static final Validator validator;
static {
validator = Validation.buildDefaultValidatorFactory().getValidator();
@ -17,21 +17,20 @@ public class ValidatorUtils {
/**
* 校验对象
* @param object 待校验对象
* @throws SnailJobClientException 校验不通过，则报SnailJobClientException异常
*
* @param object 待校验对象
* @throws SnailJobClientException 校验不通过，则报SnailJobClientException异常
*/
public static boolean validateEntity(Object object)
throws SnailJobClientException {
public static Pair<Boolean, String> validateEntity(Object object) {
Set<ConstraintViolation<Object>> constraintViolations = validator.validate(object);
if (!constraintViolations.isEmpty()) {
StringBuilder msg = new StringBuilder();
for(ConstraintViolation<Object> constraint: constraintViolations){
for (ConstraintViolation<Object> constraint : constraintViolations) {
msg.append(constraint.getMessage()).append("\n");
}
SnailJobLog.LOCAL.error(msg.toString());
return false;
}else {
return true;
return Pair.of(Boolean.FALSE, msg.toString());
} else {
return Pair.of(Boolean.TRUE, null);
}
}
}

2
snail-job-server/snail-job-server-starter/src/main/resources/application.yml
View File

@ -82,7 +82,7 @@ snail-job:
netty-port: 1788 # 服务端netty端口
limiter: 1000 # 一个客户端每秒最多接收的重试数量指令
step: 100 # 号段模式下步长配置
log-storage: 90 # 日志保存时间(单位: day)
log-storage: 45 # 日志保存时间(单位: day)
callback: # 回调配置
max-count: 288 #回调最大执行次数
trigger-interval: 900 #间隔时间