- 总体思路
- 利用kafka单播模式：开启一个定时任务
- 利用kafka广播模式：关闭某个定时任务
- 实现代码
-
WcsTimerManager类
package com.isoftstone.hig.wmsck.adapters.jd.common.timermanager; import com.alibaba.fastjson.JSONObject; import com.isoftstone.hig.common.constants.KafkaConstant; import com.isoftstone.hig.common.model.MQEvent; import com.isoftstone.hig.common.model.ResultMode; import com.isoftstone.hig.common.utils.*; import com.isoftstone.hig.wmsck.adapters.jd.api.param.timermanager.TimerTaskParam; import com.isoftstone.hig.wmsck.adapters.jd.api.param.timermanager.WcsTimerTaskEnum; import com.isoftstone.hig.wmsck.adapters.jd.common.kafka.KafkaClientManager; import org.quartz.CronExpression; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.support.CronTrigger; import org.springframework.stereotype.Component; import javax.validation.constraints.NotNull; import java.util.Date; import java.util.HashMap; import java.util.Map; import java.util.concurrent.ScheduledFuture; @Component public class WcsTimerManager { private ScheduledFuture future; private static Map<String,ScheduledFuture> map = new HashMap<>(); @Autowired private TimerTask timerTask; @Autowired private ThreadPoolTaskScheduler threadPoolTaskScheduler; @Bean public ThreadPoolTaskScheduler threadPoolTaskScheduler() { return new ThreadPoolTaskScheduler(); } /** * 开启定时器 */ public ResultMode startTask(TimerTaskParam timerTask) { ResultMode resultMode = new ResultMode(); try { if (PublicUtil.isEmpty(timerTask.getTaskNameNo())){ resultMode.setSucceed(false); resultMode.setErrMsg("任务名称编号不能为空"); return resultMode; } if (PublicUtil.isEmpty(timerTask.getTimeCycle())){ resultMode.setSucceed(false); resultMode.setErrMsg("任务时间周期不能为空"); return resultMode; } if (PublicUtil.isEmpty(timerTask.getTimeType())){ resultMode.setSucceed(false); resultMode.setErrMsg("cron表达式类型不能为空"); return resultMode; } //获取cron表达式 String cronTrigger = this.getCron(timerTask); //校验cron表达式是否合法 if (!CronExpression.isValidExpression(cronTrigger)){ LogHelper.writeInfo("cron表达式非法！"); resultMode.setSucceed(false); resultMode.setErrMsg("cron表达式非法!"); return resultMode; } //如果存在Redis标识 if (RedisUtil.hexists("future",timerTask.getTaskNameNo())){ //判断任务是否关闭,已关闭=true boolean bool = Boolean.valueOf(RedisUtil.hget("future",timerTask.getTaskNameNo())); //如果未关闭，等待其关闭 if (!bool){ //先关闭上个任务，开启新的任务！！！ this.stopTask(timerTask); LogHelper.writeInfo("先关闭上个任务，开启新的任务！！！"); //如果值为：false-未关闭，一直等待其关闭 while (!Boolean.valueOf(RedisUtil.hget("future",timerTask.getTaskNameNo()))){ Thread.sleep(1000); LogHelper.writeWarn("休眠1秒！"); } } } ExecuteTimerTask executeTimerTask = new ExecuteTimerTask(); //任务名称编号 executeTimerTask.setTaskNameNo(timerTask.getTaskNameNo()); //入参对象 executeTimerTask.setData(timerTask.getData()); future = threadPoolTaskScheduler.schedule(executeTimerTask,new CronTrigger(cronTrigger)); //放入map中,停止任务时需要释放对应的任务 map.put(timerTask.getTaskNameNo(),future); //写入标识到Redis long result = RedisUtil.hset("future",timerTask.getTaskNameNo(), "false"); LogHelper.writeInfo("Redis标识改为false-未关闭:"+result); LogHelper.writeInfo("策略已经启动"); resultMode.setErrMsg("策略已经启动！"); } catch (Exception e) { LogHelper.writeError("策略启动失败{}", e); resultMode.setSucceed(false); resultMode.setErrMsg("策略启动失败！"); } return resultMode; } /** * 关闭定时器-发送kafka消息 */ public ResultMode stopTask(TimerTaskParam timerTask) { ResultMode resultMode = new ResultMode(); try { if (PublicUtil.isEmpty(timerTask.getTaskNameNo())) { resultMode.setSucceed(false); resultMode.setErrMsg("任务名称编号不能为空"); return resultMode; } //发布kafka MQEvent mqEvent = new MQEvent<>(UtilityClass.uuid(), KafkaConstant.EVENT_WCS_TIMER_MANAGER, timerTask); String topic = SpringContextUtil.getKafkaTopicPrefix() + UtilityEnum.KafkaTopicNameEnum.TOPIC_WMSCK_JDA_TIMER_MANAGER.getTopicName(); LogHelper.writeInfo("生产消息：" + JSONObject.toJSONString(mqEvent)); KafkaClientManager.sendMessage(topic, null, JSONObject.toJSONString(mqEvent)); resultMode.setErrMsg("策略关闭请求发送成功"); }catch (Exception e){ LogHelper.writeError("策略关闭失败{}", e); resultMode.setSucceed(false); resultMode.setErrMsg("策略关闭失败！"); } return resultMode; } /** * 关闭定时器-消费kafka消息 */ public void stopTimerManager(MQEvent mqEvent) { LogHelper.writeInfo("定时器,监听测试---------" + JSONObject.toJSONString(mqEvent)); LogHelper.writeInfo("消费消息：" + JSONObject.toJSONString(mqEvent)); TimerTaskParam timerTask = JSONObject.parseObject(JSONObject.toJSONString(mqEvent.getData()), TimerTaskParam.class); //关闭定时器 if (!this.stop(timerTask).getSucceed()){ LogHelper.writeInfo("关闭定时器失败！"); } LogHelper.writeInfo("关闭定时器成功！"); } //关闭定时任务 private ResultMode stop(TimerTaskParam timerTask) { ResultMode resultMode = new ResultMode(); try { if (future != null) { future=map.get(timerTask.getTaskNameNo()); future.cancel(true); LogHelper.writeInfo("future:"+future); } //Redis标识改为true-已关闭 long result = RedisUtil.hset("future",timerTask.getTaskNameNo(),"true"); LogHelper.writeInfo("Redis标识改为true-已关闭:"+result); LogHelper.writeInfo("策略已经停止"); resultMode.setErrMsg("策略已经停止"); return resultMode; } catch (Exception e) { LogHelper.writeError("策略停止失败{}", e); resultMode.setSucceed(false); resultMode.setErrMsg("策略停止失败！"); return resultMode; } } //获取cron表达式 private String getCron(TimerTaskParam timerTask){ //间隔时间或cron表达式 String time = timerTask.getTimeCycle(); //按秒执行 if (WcsTimerTaskEnum.TimerTypeEnum.WCS_TimerType_10.getCode().equals(timerTask.getTimeType())){ return "0/"+time+" * * * * ?"; } //按分钟执行 if (WcsTimerTaskEnum.TimerTypeEnum.WCS_TimerType_20.getCode().equals(timerTask.getTimeType())){ return "0 0/"+time+" * * * ?"; } //按小时执行 if (WcsTimerTaskEnum.TimerTypeEnum.WCS_TimerType_30.getCode().equals(timerTask.getTimeType())){ return "0 0 0/"+time+" * * ?"; } //自定义时间cron表达式 return time; } //执行任务 private class ExecuteTimerTask implements Runnable{ /** * 任务名称编号 */ @NotNull(message = "任务名称编号不能为空") private String taskNameNo; /** * 入参对象 */ private Object data; public String getTaskNameNo() { return taskNameNo; } public void setTaskNameNo(String taskNameNo) { this.taskNameNo = taskNameNo; } public Object getData() { return data; } public void setData(Object data) { this.data = data; } @Override public void run() { if (this.taskNameNo.equals(WcsTimerTaskEnum.TimerTaskEnum.WCS_TIMER_TASK_10.getCode())) { LogHelper.writeInfo("定时生成设备摘果单-启动中:" + new Date()); //定时生成设备摘果单 timerTask.buildPickingTask(this.data); } else { LogHelper.writeInfo("待定任务-我在启动中:" + new Date()); } } } } -
TimerTask类
package com.isoftstone.hig.wmsck.adapters.jd.common.timermanager; import com.alibaba.fastjson.JSON; import com.isoftstone.hig.common.model.ResultMode; import com.isoftstone.hig.wmsck.adapters.jd.client.WcsLocalCallServiceBusiness; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; //定时任务类 @Service public class TimerTask { @Autowired private WcsLocalCallServiceBusiness wcsLocalCallServiceBusiness; /** * 自动生成设备摘果单 */ public ResultMode<String> buildPickingTask(Object data){ //LogHelper.writeInfo("入参："+JSON.toJSONString(data)); String warehouseNo = JSON.parseObject(JSON.toJSONString(data),String.class); //生成设备摘果单 return wcsLocalCallServiceBusiness.buildPickingTask(warehouseNo); } } -
TimerTaskParam类
package com.isoftstone.hig.wmsck.adapters.jd.api.param.timermanager; import io.swagger.annotations.ApiModel; import io.swagger.annotations.ApiModelProperty; import java.io.Serializable; @ApiModel("定时器入参") public class TimerTaskParam implements Serializable { private static final long serialVersionUID = 2168115660376016205L; /** * 任务名称编号 */ @ApiModelProperty(value = "任务名称编号", name ="taskNameNo") private String taskNameNo; /** * 任务时间周期或cron表达式 */ @ApiModelProperty(value = "任务时间周期或cron表达式", name ="timeCycle") private String timeCycle; /** * cron表达式类型:参考WcsTimerTaskEnum.TimerTypeEnum枚举类型 */ @ApiModelProperty(value = "cron表达式类型", name ="timeType") private String timeType; /** * 入参对象 */ private Object data; public String getTaskNameNo() { return taskNameNo; } public void setTaskNameNo(String taskNameNo) { this.taskNameNo = taskNameNo; } public String getTimeCycle() { return timeCycle; } public void setTimeCycle(String timeCycle) { this.timeCycle = timeCycle; } public String getTimeType() { return timeType; } public void setTimeType(String timeType) { this.timeType = timeType; } public Object getData() { return data; } public void setData(Object data) { this.data = data; } } -
WcsTimerTaskEnum
package com.isoftstone.hig.wmsck.adapters.jd.api.param.timermanager; public class WcsTimerTaskEnum { /** * 定时任务列表枚举类 */ public enum TimerTaskEnum { WCS_TIMER_TASK_10("10", "定时生成设备摘果单"), WCS_TIMER_TASK_20("20", "待定"); private String code; private String desc; public String getCode() { return code; } public String getDesc() { return desc; } private TimerTaskEnum(String code, String desc) { this.code = code; this.desc = desc; } } /** * cron表达式类型 */ public enum TimerTypeEnum { WCS_TimerType_10("10", "秒"), WCS_TimerType_20("20", "分钟"), WCS_TimerType_30("30", "小时"), WCS_TimerType_40("40", "自定义cron表达式"); private String code; private String desc; public String getCode() { return code; } public String getDesc() { return desc; } private TimerTypeEnum(String code, String desc) { this.code = code; this.desc = desc; } } }
