Commit e260687e by huangfusuper

java任务自定义时间配置个性化配置

parent 5a63c4a0
...@@ -1308,6 +1308,8 @@ public class ApiFlowServiceImpl implements ApiFlowService { ...@@ -1308,6 +1308,8 @@ public class ApiFlowServiceImpl implements ApiFlowService {
} }
//TODO 替换参数 //TODO 替换参数
waitingTaskList.forEach(waitingTask -> { waitingTaskList.forEach(waitingTask -> {
//补批只替换不是java的节点
if(!NodeTypeEnum.JAVA.getCode().equals( waitingTask.getJobType())){
if (StringUtils.isNotEmpty(waitingTask.getRunParam())){ if (StringUtils.isNotEmpty(waitingTask.getRunParam())){
String param = PlaceholderUtils.paramPlaceholder(waitingTask.getRunParam(), repairTime); String param = PlaceholderUtils.paramPlaceholder(waitingTask.getRunParam(), repairTime);
String runCommand = waitingTask.getRunCommand(); String runCommand = waitingTask.getRunCommand();
...@@ -1315,6 +1317,7 @@ public class ApiFlowServiceImpl implements ApiFlowService { ...@@ -1315,6 +1317,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
waitingTask.setRunCommand(runCommand); waitingTask.setRunCommand(runCommand);
waitingTask.setRunParam(param); waitingTask.setRunParam(param);
} }
}
waitingTaskDateList.add(waitingTask); waitingTaskDateList.add(waitingTask);
}); });
result.put(repairTime, waitingTaskDateList); result.put(repairTime, waitingTaskDateList);
......
...@@ -3,6 +3,7 @@ package com.byit.service.impl; ...@@ -3,6 +3,7 @@ package com.byit.service.impl;
import cn.hutool.http.HttpUtil; import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject; import com.alibaba.fastjson.JSONObject;
import com.byit.dto.personalise.JavaJobConfDto;
import com.byit.dto.plugin.JavaTask; import com.byit.dto.plugin.JavaTask;
import com.byit.dto.plugin.RunLog; import com.byit.dto.plugin.RunLog;
import com.byit.dto.plugin.RunNode; import com.byit.dto.plugin.RunNode;
...@@ -38,7 +39,9 @@ import org.springframework.transaction.annotation.Transactional; ...@@ -38,7 +39,9 @@ import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.text.ParseException;
import java.util.*; import java.util.*;
import java.util.concurrent.TimeUnit;
import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName; import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
...@@ -322,9 +325,22 @@ public class ApiNodeServiceImpl implements ApiNodeService { ...@@ -322,9 +325,22 @@ public class ApiNodeServiceImpl implements ApiNodeService {
ValidationUtil.dataNotBank(param, "请求参数不允许为空!"); ValidationUtil.dataNotBank(param, "请求参数不允许为空!");
JavaTask javaTask = JSON.parseObject(param, JavaTask.class); JavaTask javaTask = JSON.parseObject(param, JavaTask.class);
ValidationUtil.dataNotBank(javaTask.getTaskName(), "任务实现类的名称不允许为空!"); ValidationUtil.dataNotBank(javaTask.getTaskName(), "任务实现类的名称不允许为空!");
ValidationUtil.dataNotBank(javaTask.getCron(), "任务的调度时间不允许为空!");
ValidationUtil.isTrueValidation(!CronExpression.isValidExpression(javaTask.getCron()), "工作流cron表达式不符合规范!"); //当自定义参数不为空时
javaTask.setTriggerTime(new CronExpression(javaTask.getCron()).getNextValidTimeAfter(new Date()).getTime()); String personalise = javaTask.getPersonalise();
if(StringUtils.isNoneBlank(personalise)){
JavaJobConfDto javaJobConfDto = JSON.parseObject(personalise, JavaJobConfDto.class);
Long startTime = javaJobConfDto.getStartTime();
Integer interval = javaJobConfDto.getInterval();
TimeUnit timeUnit = javaJobConfDto.getTimeUnit();
ValidationUtil.dataNotNull(startTime, "检测发现个性化配置的开始时间为空!");
ValidationUtil.dataNotNull(interval, "检测发现个性化配置的间隔时间为空!");
ValidationUtil.dataNotNull(timeUnit,"检测发现个性化配置的时间单位为空!");
javaTask.setTriggerTime(startTime);
}else{
parseCron(javaTask);
}
//如果不为空且不为-1 //如果不为空且不为-1
if(null != javaTask.getRepeatCount() && -1 != javaTask.getRepeatCount()){ if(null != javaTask.getRepeatCount() && -1 != javaTask.getRepeatCount()){
javaTask.setRemainingCount(javaTask.getRemainingCount()); javaTask.setRemainingCount(javaTask.getRemainingCount());
...@@ -334,4 +350,10 @@ public class ApiNodeServiceImpl implements ApiNodeService { ...@@ -334,4 +350,10 @@ public class ApiNodeServiceImpl implements ApiNodeService {
} }
return javaTask; return javaTask;
} }
private void parseCron(JavaTask javaTask) throws ParseException {
ValidationUtil.dataNotBank(javaTask.getCron(), "任务的调度时间不允许为空!");
ValidationUtil.isTrueValidation(!CronExpression.isValidExpression(javaTask.getCron()), "工作流cron表达式不符合规范!");
javaTask.setTriggerTime(new CronExpression(javaTask.getCron()).getNextValidTimeAfter(new Date()).getTime());
}
} }
\ No newline at end of file
package com.byit.service.mapservice.impl; package com.byit.service.mapservice.impl;
import cn.hutool.core.date.DateUtil; import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.byit.dto.personalise.JavaJobConfDto;
import com.byit.dto.plugin.JavaTask; import com.byit.dto.plugin.JavaTask;
import com.byit.job.WorkRoulette; import com.byit.job.WorkRoulette;
import com.byit.job.utils.CronExpression; import com.byit.job.utils.CronExpression;
...@@ -9,11 +11,13 @@ import com.byit.service.JavaTaskService; ...@@ -9,11 +11,13 @@ import com.byit.service.JavaTaskService;
import com.byit.service.mapservice.JavaTaskAndLogService; import com.byit.service.mapservice.JavaTaskAndLogService;
import com.byit.task.JavaTaskJobTask; import com.byit.task.JavaTaskJobTask;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils; import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.text.ParseException; import java.text.ParseException;
import java.util.Date; import java.util.Date;
import java.util.concurrent.TimeUnit;
/** /**
* @author Administrator * @author Administrator
...@@ -32,7 +36,18 @@ public class JavaTaskAndLogServiceServiceImpl implements JavaTaskAndLogService { ...@@ -32,7 +36,18 @@ public class JavaTaskAndLogServiceServiceImpl implements JavaTaskAndLogService {
JavaTask updateJavaTask = new JavaTask(); JavaTask updateJavaTask = new JavaTask();
try { try {
BeanUtils.copyProperties(javaTask,updateJavaTask); BeanUtils.copyProperties(javaTask,updateJavaTask);
String personalise = javaTask.getPersonalise();
if(StringUtils.isNotBlank(personalise)){
JavaJobConfDto javaJobConfDto = JSON.parseObject(personalise, JavaJobConfDto.class);
Integer interval = javaJobConfDto.getInterval();
TimeUnit timeUnit = javaJobConfDto.getTimeUnit();
//使用默认的毫秒级别
long toMillis = timeUnit.toMillis(interval);
updateJavaTask.setTriggerTime(javaTask.getTriggerTime()+toMillis);
}else{
updateJavaTask.setTriggerTime(new CronExpression(javaTask.getCron()).getNextValidTimeAfter(new Date(javaTask.getTriggerTime())).getTime()); updateJavaTask.setTriggerTime(new CronExpression(javaTask.getCron()).getNextValidTimeAfter(new Date(javaTask.getTriggerTime())).getTime());
}
//保存到调度轮 //保存到调度轮
JavaTaskJobTask javaTaskJobTask = new JavaTaskJobTask(javaTask); JavaTaskJobTask javaTaskJobTask = new JavaTaskJobTask(javaTask);
WorkRoulette.addJob(javaTaskJobTask,javaTask.getTriggerTime()); WorkRoulette.addJob(javaTaskJobTask,javaTask.getTriggerTime());
......
...@@ -246,6 +246,7 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask ...@@ -246,6 +246,7 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask
BeanUtils.copyProperties(waitingTask, target); BeanUtils.copyProperties(waitingTask, target);
target.setReRunId(waitingTask.getReRunId()); target.setReRunId(waitingTask.getReRunId());
target.setId(null); target.setId(null);
target.setRunParam(waitingTask.getRunParam());
target.setTriggerTime(0L); target.setTriggerTime(0L);
return target; return target;
}).collect(Collectors.toList()); }).collect(Collectors.toList());
......
...@@ -7,6 +7,7 @@ import com.byit.dto.plugin.RunLog; ...@@ -7,6 +7,7 @@ import com.byit.dto.plugin.RunLog;
import com.byit.enums.NodeRunStatusPropertyEnum; import com.byit.enums.NodeRunStatusPropertyEnum;
import com.byit.enums.ScheduleTypeEnum; import com.byit.enums.ScheduleTypeEnum;
import com.byit.enums.task.RunResultEnum; import com.byit.enums.task.RunResultEnum;
import com.byit.job.utils.DateUtil;
import com.byit.model.JobTaskRunLogWithBLOBs; import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.packet.request.PluginRpcRequestPacket; import com.byit.packet.request.PluginRpcRequestPacket;
import com.byit.packet.response.PluginRpcResponsePacket; import com.byit.packet.response.PluginRpcResponsePacket;
...@@ -54,7 +55,9 @@ public class JavaTaskJobTask implements TimerTask { ...@@ -54,7 +55,9 @@ public class JavaTaskJobTask implements TimerTask {
communicationParam.setLogId(logId+""); communicationParam.setLogId(logId+"");
communicationParam.setCallbackUrl(callbackUrl); communicationParam.setCallbackUrl(callbackUrl);
communicationParam.setBody(javaTask.getParam()); communicationParam.setBody(javaTask.getParam());
communicationParam.setExpand2(String.valueOf(javaTask.getTriggerTime()));
String dateFormat = DateUtil.dateFormat(new Date(javaTask.getTriggerTime()), DateUtil.FORMAT_DATE_TIME);
communicationParam.setExpand2(dateFormat);
request.setParam(communicationParam); request.setParam(communicationParam);
request.setExtension(logId+""); request.setExtension(logId+"");
......
...@@ -16,11 +16,12 @@ ...@@ -16,11 +16,12 @@
<result column="alarml_action" jdbcType="CHAR" property="alarmlAction" /> <result column="alarml_action" jdbcType="CHAR" property="alarmlAction" />
<result column="alarm_email" jdbcType="VARCHAR" property="alarmEmail" /> <result column="alarm_email" jdbcType="VARCHAR" property="alarmEmail" />
<result column="del_mark" jdbcType="CHAR" property="delMark" /> <result column="del_mark" jdbcType="CHAR" property="delMark" />
<result column="personalise" jdbcType="VARCHAR" property="personalise" />
</resultMap> </resultMap>
<sql id="Base_Column_List"> <sql id="Base_Column_List">
<!-- generated @mbg.generated date: 2020-04-14 --> <!-- generated @mbg.generated date: 2020-04-14 -->
id, job_name, task_name, repeat_count, param, trigger_time, cron, remaining_count, id, job_name, task_name, repeat_count, param, trigger_time, cron, remaining_count,
create_time, update_time, alarml_action, alarm_email, del_mark create_time, update_time, alarml_action, alarm_email, del_mark, personalise
</sql> </sql>
<select id="findAll" resultMap="BaseResultMap"> <select id="findAll" resultMap="BaseResultMap">
...@@ -101,6 +102,9 @@ ...@@ -101,6 +102,9 @@
<if test="delMark != null"> <if test="delMark != null">
del_mark, del_mark,
</if> </if>
<if test="personalise != null">
personalise,
</if>
</trim> </trim>
<trim prefix="values (" suffix=")" suffixOverrides=","> <trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="id != null"> <if test="id != null">
...@@ -142,6 +146,9 @@ ...@@ -142,6 +146,9 @@
<if test="delMark != null"> <if test="delMark != null">
#{delMark,jdbcType=CHAR}, #{delMark,jdbcType=CHAR},
</if> </if>
<if test="personalise != null">
#{personalise,jdbcType=VARCHAR},
</if>
</trim> </trim>
</insert> </insert>
<update id="updateByIdSelective" parameterType="com.byit.dto.plugin.JavaTask"> <update id="updateByIdSelective" parameterType="com.byit.dto.plugin.JavaTask">
...@@ -184,6 +191,9 @@ ...@@ -184,6 +191,9 @@
<if test="delMark != null"> <if test="delMark != null">
del_mark = #{delMark,jdbcType=CHAR}, del_mark = #{delMark,jdbcType=CHAR},
</if> </if>
<if test="personalise != null">
personalise = #{personalise,jdbcType=VARCHAR},
</if>
</set> </set>
where id = #{id,jdbcType=INTEGER} where id = #{id,jdbcType=INTEGER}
</update> </update>
...@@ -224,6 +234,9 @@ ...@@ -224,6 +234,9 @@
<if test="delMark != null"> <if test="delMark != null">
del_mark = #{delMark,jdbcType=CHAR}, del_mark = #{delMark,jdbcType=CHAR},
</if> </if>
<if test="personalise != null">
personalise = #{personalise,jdbcType=VARCHAR},
</if>
</set> </set>
where job_name = #{jobName,jdbcType=VARCHAR} where job_name = #{jobName,jdbcType=VARCHAR}
</update> </update>
......
...@@ -12,8 +12,8 @@ import java.util.Date; ...@@ -12,8 +12,8 @@ import java.util.Date;
* @author huangfu * @author huangfu
*/ */
public class DateUtil { public class DateUtil {
private static final String NOT_FORMAT_DATE = "yyyyMMdd"; public static final String NOT_FORMAT_DATE = "yyyyMMdd";
private static final String FORMAT_DATE_TIME = "yyyy-MM-dd HH:mm:ss"; public static final String FORMAT_DATE_TIME = "yyyy-MM-dd HH:mm:ss";
/** /**
* 转换成对应格式的字符串 * 转换成对应格式的字符串
......
package com.byit.dto.personalise;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
import java.util.concurrent.TimeUnit;
/**
* @author huangfu
* java立即执行任务的自定义调度配置
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class JavaJobConfDto implements Serializable {
private static final long serialVersionUID = -8854344978247302329L;
/**
* 任务的开始时间
*/
private Long startTime;
/**
* 任务的间隔时间
*/
private Integer interval;
/**
* 任务的时间单位
*/
private TimeUnit timeUnit;
}
package com.byit.dto.plugin; package com.byit.dto.plugin;
import com.byit.dto.personalise.JavaJobConfDto;
import lombok.AllArgsConstructor; import lombok.AllArgsConstructor;
import lombok.Data; import lombok.Data;
import lombok.NoArgsConstructor; import lombok.NoArgsConstructor;
...@@ -91,6 +92,11 @@ public class JavaTask implements Serializable { ...@@ -91,6 +92,11 @@ public class JavaTask implements Serializable {
private String delMark; private String delMark;
/** /**
* 个性化配置 json {@link JavaJobConfDto}
*/
private String personalise;
/**
*/ */
private static final long serialVersionUID = 1L; private static final long serialVersionUID = 1L;
} }
package com.byit.utils;
import java.io.PrintWriter;
import java.io.StringWriter;
/**
* @author huangfu
*/
public class PluginLogUtils {
/**
* 读取异常信息
* @param obj 异常对象
* @return 返回堆栈异常信息
*/
public static String getMessage(Object obj) {
if (obj == null) {
return "";
}
if (obj instanceof Throwable) {
StringWriter str = new StringWriter();
PrintWriter pw = new PrintWriter(str);
((Throwable) obj).printStackTrace(pw);
return str.toString();
} else {
return obj.toString();
}
}
}
...@@ -10,6 +10,7 @@ import com.byit.packet.request.PluginRpcRequestPacket; ...@@ -10,6 +10,7 @@ import com.byit.packet.request.PluginRpcRequestPacket;
import com.byit.packet.response.PluginRpcResponsePacket; import com.byit.packet.response.PluginRpcResponsePacket;
import com.byit.param.PluginBeat; import com.byit.param.PluginBeat;
import com.byit.task.handler.interfaces.IJobHandler; import com.byit.task.handler.interfaces.IJobHandler;
import com.byit.utils.PluginLogUtils;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler; import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent; import io.netty.handler.timeout.IdleStateEvent;
...@@ -64,11 +65,11 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin ...@@ -64,11 +65,11 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
sendMsg(rpcResponsePacket,msg); sendMsg(rpcResponsePacket,msg);
}catch (Throwable e){ }catch (Throwable e){
rpcResponsePacket.setCode(JobResultEnum.FAIL.getCode()); rpcResponsePacket.setCode(JobResultEnum.FAIL.getCode());
rpcResponsePacket.setMsg(e.getMessage()); rpcResponsePacket.setMsg(PluginLogUtils.getMessage(e));
rpcResponsePacket.setStatus(false); rpcResponsePacket.setStatus(false);
rpcResponsePacket.setExtension(msg.getExtension()); rpcResponsePacket.setExtension(msg.getExtension());
ReturnResult<String> execute = new ReturnResult<>(); ReturnResult<String> execute = new ReturnResult<>();
execute.setMsg(e.getMessage()); execute.setMsg(PluginLogUtils.getMessage(e));
execute.setCode(JobResultEnum.FAIL.getCode()); execute.setCode(JobResultEnum.FAIL.getCode());
rpcResponsePacket.setResult(execute); rpcResponsePacket.setResult(execute);
sendMsg(rpcResponsePacket,msg); sendMsg(rpcResponsePacket,msg);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment