修正任务不遵循cron表达式 重复调用问题

byit-myth-admin/src/main/java/com/byit/AdminApplication.java

@@ -2,8 +2,13 @@ package com.byit;
 
 import com.byit.annotations.EnablePluginClient;
 import com.byit.rpc.remoting.provider.annotation.RpcService;
+
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.web.cors.CorsConfiguration;
+import org.springframework.web.cors.UrlBasedCorsConfigurationSource;
+import org.springframework.web.filter.CorsFilter;
 
 /**
  * @program: byit-myth-job->AdminApplication
@@ -18,4 +23,21 @@ public class AdminApplication {
     public static void main(String[] args) {
         SpringApplication.run(AdminApplication.class,args);
     }
+
+    private CorsConfiguration corsConfiguration(){
+        CorsConfiguration corsConfiguration = new CorsConfiguration();
+        corsConfiguration.addAllowedOrigin("*");
+        corsConfiguration.addAllowedHeader("*");
+        corsConfiguration.addAllowedMethod("*");
+        corsConfiguration.setAllowCredentials(true);
+        corsConfiguration.setMaxAge(3600L);
+        return corsConfiguration;
+    }
+
+    @Bean
+    public CorsFilter corsFilter(){
+        UrlBasedCorsConfigurationSource source  = new UrlBasedCorsConfigurationSource();
+        source.registerCorsConfiguration("/**",corsConfiguration());
+        return new CorsFilter(source);
+    }
 }

byit-myth-admin/src/main/java/com/byit/api/ApiNodeController.java

@@ -30,6 +30,7 @@ public class ApiNodeController {
         return monitorKey;
     }
 
+
     @PostMapping("runHistory")
     public ResponseResult runHistory(String nodeId){
         List<JobTaskRunLog> jobTaskRunLogList = apiNodeService.runHistory(nodeId);
@@ -84,6 +85,12 @@ public class ApiNodeController {
         return ResponseResult.ok("SUCCESS");
     }
 
+    @PostMapping("runTask")
+    public ResponseResult runTask(String jobName){
+        apiNodeService.runTask(jobName);
+        return ResponseResult.ok("SUCCESS");
+    }
+
     @PostMapping("loadCurrentStatusByJobName")
     public ResponseResult loadCurrentStatusByJobName(String jobNames){
         Map<String, JobTaskRunLog> jobTaskRunLogMap = apiNodeService.loadCurrentStatusByJobName(jobNames);

byit-myth-admin/src/main/java/com/byit/service/ApiNodeService.java

@@ -17,6 +17,8 @@ public interface ApiNodeService {
      */
     String runNode(String param);
 
+
+
     /**
      * 根据nodeid查询运行历史
      * @param nodeId
@@ -43,6 +45,13 @@ public interface ApiNodeService {
     void addJavaTask(String param) throws Exception;
 
     /**
+     * 立即运行任务不需要验证是否存在
+     * @param param
+     * @throws Exception
+     */
+    void runTask(String param) ;
+
+    /**
      * 功能描述 更新任务配置信息
      * @author gml
      * @date 2020-04-14 11:28

byit-myth-admin/src/main/java/com/byit/service/impl/ApiNodeServiceImpl.java

@@ -21,6 +21,7 @@ import com.byit.task.JavaTaskJobTask;
 import com.byit.task.ScriptExecutorJobTask;
 import com.byit.utils.ValidationUtil;
 import io.netty.util.TimerTask;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
@@ -143,6 +144,23 @@ public class ApiNodeServiceImpl implements ApiNodeService {
     }
 
     @Override
+    public void runTask(String param) {
+        ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
+        JavaTask javaTask = JSON.parseObject(param, JavaTask.class);
+        ValidationUtil.dataNotBank(javaTask.getTaskName(), "任务实现类的名称不允许为空！");
+        if(StringUtils.isBlank(javaTask.getJobName())){
+            javaTask.setJobName(javaTask.getTaskName());
+        }
+        javaTask.setTriggerTime(0L);
+        if (null != javaTask.getAlarmlAction() && "0".equals(javaTask.getRepeatCount())){
+            ValidationUtil.dataNotBank(javaTask.getAlarmEmail(), "设置为告警时告警邮箱不允许为空");
+        }
+
+        TimerTask timerTask = new JavaTaskJobTask(javaTask);
+        WorkRoulette.addJob(timerTask, System.currentTimeMillis());
+    }
+
+    @Override
     public void updateJavaTask(String param) throws Exception {
         JavaTask javaTask = validate(param);
         ValidationUtil.dataNotBank(javaTask.getJobName(), "jobName不允许为空");

byit-myth-admin/src/main/java/com/byit/view/FlowViewController.java

+package com.byit.view;
+
+import com.byit.service.FlowVersionService;
+import com.byit.vo.FlowVersionVo;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+/**
+ * 工作流视图界面
+ * @author huangfu
+ */
+@RestController
+@RequestMapping("view/flow")
+public class FlowViewController {
+    private final FlowVersionService flowVersionService;
+
+    public FlowViewController(FlowVersionService flowVersionService) {
+        this.flowVersionService = flowVersionService;
+    }
+    @PostMapping("getAllFlowVersion")
+    public List<FlowVersionVo> getAllFlowVersion(){
+        return flowVersionService.findAllFlow();
+    }
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/job/WorkRoulette.java

@@ -27,7 +27,7 @@ public class WorkRoulette {
 
 
     public static void addJob(TimerTask timerTask,long triggerNextTime) {
-        log.info("-----添加任务{}-------",((JavaTaskJobTask)timerTask).getJavaTask().getTaskName());
+        //log.info("-----添加任务{}-------",((JavaTaskJobTask)timerTask).getJavaTask().getTaskName());
         HASHED_WHEEL_TIMER.newTimeout(timerTask, TimeUnit.MILLISECONDS.toNanos(triggerNextTime-System.currentTimeMillis()), TimeUnit.NANOSECONDS);
     }
 

byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/FlowVersionMapper.java

 package com.byit.mapper;
 
 import com.byit.model.FlowVersion;
+import org.springframework.stereotype.Repository;
 
+import java.util.List;
+
+@Repository
 public interface FlowVersionMapper {
+    /**
+     * 查询全部的工作流
+     * @return
+     */
+    List<FlowVersion> findAllFlow();
+
     int deleteById(Integer flowVersionId);
 
     int insertSelective(FlowVersion record);

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/FlowVersionService.java

+package com.byit.service;
+
+import com.byit.model.FlowVersion;
+import com.byit.vo.FlowVersionVo;
+
+import java.util.List;
+
+/**
+ * @author huangfu
+ */
+public interface FlowVersionService {
+    /**
+     * 查询全部的工作流
+     * @return
+     */
+    List<FlowVersionVo> findAllFlow();
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/FlowVersionServiceImpl.java

+package com.byit.service.impl;
+
+import com.byit.mapper.FlowVersionMapper;
+import com.byit.model.FlowVersion;
+import com.byit.service.FlowVersionService;
+import com.byit.vo.FlowVersionVo;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * 工作流版本业务的实现
+ * @author huangfu
+ */
+@Service
+public class FlowVersionServiceImpl implements FlowVersionService {
+    private final FlowVersionMapper flowVersionMapper;
+
+    public FlowVersionServiceImpl(FlowVersionMapper flowVersionMapper) {
+        this.flowVersionMapper = flowVersionMapper;
+    }
+
+    @Override
+    public List<FlowVersionVo> findAllFlow() {
+        List<FlowVersion> allFlow = flowVersionMapper.findAllFlow();
+        Map<String, List<FlowVersion>> versionColl = allFlow.stream().collect(Collectors.groupingBy(FlowVersion::getFlowName));
+        List<FlowVersionVo> flowVersionVos = new ArrayList<>();
+        versionColl.forEach((key,value) ->{
+            FlowVersionVo flowVersionVo = new FlowVersionVo();
+            flowVersionVo.setFlowName(key);
+            flowVersionVo.setChildren(value);
+            flowVersionVos.add(flowVersionVo);
+        });
+        return flowVersionVos;
+    }
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java

@@ -152,6 +152,8 @@ public class ScriptExecutorJobTask implements TimerTask {
                 //上一次的执行日志
                 String runMsg = jobTaskRunLogById.getRunMsg();
                 jobTaskRunLog.setRunMsg(runMsg+"|"+dispatchResponseDto.getMsg());
+            }else{
+                jobTaskRunLog.setRunMsg(dispatchResponseDto.getMsg());
             }
         }
         //获取执行机地址

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/TaskThreadRunHelper.java

@@ -230,7 +230,7 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper {
         }
         for (JobTaskRunLog jobTaskRunLog : errorJobLog) {
             //失败重试次数大于0 而且错误原因不是上级节点执行失败
-            if(jobTaskRunLog.getFailedRemainingCount()>0 && !("6".equals(jobTaskRunLog.getRunCode()) && "5".equals(jobTaskRunLog.getRunCode()))){
+            if(jobTaskRunLog.getFailedRemainingCount()!= null && jobTaskRunLog.getFailedRemainingCount()>0 && !("6".equals(jobTaskRunLog.getRunCode()) || "5".equals(jobTaskRunLog.getRunCode()))){
                 log.debug("----------------【{}节点没有重试完毕】----------------",jobTaskRunLog);
                 return false;
             }

byit-myth-core/myth-admin-core/src/main/java/com/byit/vo/FlowVersionVo.java

+package com.byit.vo;
+
+import com.byit.model.FlowVersion;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * 工作流版本的视图载体
+ * @author huangfu
+ */
+@ApiModel("工作流版本的视图载体")
+@Data
+public class FlowVersionVo implements Serializable {
+    /**
+     * 工作流名字
+     */
+    @ApiModelProperty("工作流的名字")
+    private String flowName;
+    /**
+     * 对应的版本
+     */
+    @ApiModelProperty("对应的工作流的所由版本")
+    private List<FlowVersion> children;
+}

byit-myth-core/myth-admin-core/src/main/resources/mapper/FlowVersionMapper.xml

@@ -31,6 +31,11 @@
     repeat_count, version_mark, workspace_id, author, principal, version_name, is_inner
   </sql>
 
+  <select id="findAllFlow" resultMap="BaseResultMap">
+    select <include refid="Base_Column_List" />
+    from flow_version where remove_mark = '1'
+  </select>
+
   <select id="getById" parameterType="java.lang.Integer" resultMap="BaseResultMap">
     <!-- generated @mbg.generated date: 2019-12-31 -->
     select 

byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml

@@ -349,7 +349,7 @@
         #{triggerCode,jdbcType=VARCHAR},
       </if>
       <if test="triggerTime != null">
-        #{triggerTime,jdbcType=DATE},
+        #{triggerTime,jdbcType=TIMESTAMP},
       </if>
       <if test="jobGroupIp != null">
         #{jobGroupIp,jdbcType=VARCHAR},

byit-myth-executor/myth-executor-plugin/src/main/java/com/byit/utils/JobUtils.java

@@ -139,6 +139,10 @@ public class JobUtils {
      */
     private static final String REQUEST_RUN_JAVATASK = "/api/node/runJavaTask";
     /**
+     * 立即运行quartz任务
+     */
+    private static final String REQUEST_RUN_TASK = "/api/node/runTask";
+    /**
      * 获取当前的运行状态
      */
     private static final String REQUEST_LOADSTATUS_JAVATASK = "/api/node/loadCurrentStatusByJobName";
@@ -597,6 +601,18 @@ public class JobUtils {
     }
 
     /**
+     * 立即运行quartz任务 不验证是否存在数据库中
+     * @param javaTask
+     * @return
+     */
+    public static ResponseResult runTask(JavaTask javaTask){
+        log.info("-------------立即运行quartz任务----------------------");
+        String response = createHttpRequest(REQUEST_RUN_TASK, "param=" + JSON.toJSONString(javaTask,WriteClassName));
+        log.info("--------------------立即运行quartz任务，结果为：{}------------------------",response);
+        return JSON.parseObject(response, ResponseResult.class);
+    }
+
+    /**
      * 获取当前的运行状态
      * @param jobNames
      * @return

demo-client/byit-demo-client/src/main/java/com/byit/job/AddComplexPy1.java

@@ -13,11 +13,13 @@ import java.util.concurrent.TimeUnit;
  */
 public class AddComplexPy1 {
     public static void main(String[] args) {
+
         PluginPackage pluginPackage = new PluginPackage();
         pluginPackage.setWorkspaceName("test");
         pluginPackage.setFlow(createFlow());
         JobUtils.setRequestUrl("http://127.0.0.1:8081/myth-job-admin");
         JobUtils.setTOKEN("test");
+        //JobUtils.addWorkspace("test");
         JobUtils.publish(pluginPackage);
     }