Skip to content

B
byit-myth-job
Commit a569cdae by huangfusuper
Options

特殊的补批接口

parent a8834421
Showing with 275 additions and 136 deletions
byit-myth-admin/src/main/java/com/byit/api/ApiFlowOperatingController.java 0 → 100644
package com.byit.api;
import com.byit.dto.specials.SpecialJobParam;
import io.swagger.annotations.Api;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
*
* 需求：节点操作执行的Controllrt
*
* @author huangfu
* @date 2020年9月22日18:37:41
*/
@Api(tags = "工作流操作执行api")
@RestController
@RequestMapping("api/operating/node/")
public class ApiFlowOperatingController {
/**
* 特殊的补批接口 可以依照魔衣工作流下的某一个节点，自动补批其下所由的节点
* @param specialJobParam 采纳数信息
*/
@PostMapping("specialRunBatch")
public void specialRunBatch(@RequestBody SpecialJobParam specialJobParam){}
}
byit-myth-admin/src/main/java/com/byit/api/ApiNodeOperatingController.java 0 → 100644
package com.byit.api;
import io.swagger.annotations.Api;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
*
* 需求：节点操作执行的Controllrt
*
* @author huangfu
* @date 2020年9月22日18:37:41
*/
@Api(tags = "节点操作执行api")
@RestController
@RequestMapping("api/operating/flow")
public class ApiNodeOperatingController {
}
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/plugin/QualityCheckVo.java → byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/specials/SpecialJavaNode.java
package com.byit.dto.plugin;
package com.byit.dto.specials;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
/**
* 数据质量审核专用DTO
* 特殊节点重新执行
*
* 需求: ddmp需求，当该java节点的工作流任务执行完毕后
* 游离节点再次执行后，将游离节点执行的之日拼装在对应工作流上的节点日志
* @author huangfu
* @date 2020年9月15日10:04:40
* @date 2020年9月22日18:35:05
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class QualityCheckVo {
private String workspaceName;
public class SpecialJavaNode implements Serializable {
/**
* 工作流名称
*/
private String flowName;
/**
* 节点名称
*/
private String nodeName;
/**
* 当前执行的日志的url
*/
private String thisLogUrlPath;
}
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/specials/SpecialJobParam.java 0 → 100644
package com.byit.dto.specials;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.Map;
/**
* 特殊的补批作业
* demand：该参数应对与特殊的补批需求，
* demand One: 可以基于某一节点，补批下方所有节点（局限于该节点之下的节点，不包含同级节点以及下方节点）
* demand two：可以重新给补批节点设置参数
* demand three: 是否运行本节点
* @author huangfu
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class SpecialJobParam {
/**
* 根据工作流名称
*/
private String flowName;
/**
* 是否补批本节点 1：是 2：否
*/
private String supplementStatus;
/**
* 开始节点的名称
*/
private String startNodeTaskName;
/**
* 任务名称 -> 任务参数载体
*/
Map<String,String> paramCarrier;
}
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/web/ReturnResult.java
......@@ -8,6 +8,8 @@ import lombok.NoArgsConstructor;
import java.io.Serializable;
/**
* 当前规定的所有的返回结果集都必须是{@link com.byit.dto.specials.SpecialJavaNode} 类型的
*
* @program: byit-myth-job->ReturnResult
* @description: 服务调用的返回结果
* @author: huangfu
......@@ -22,6 +24,10 @@ public class ReturnResult<T> implements Serializable {
public static final ReturnResult<String> FAIL = new ReturnResult<>(JobResultEnum.FAIL.getCode(), JobResultEnum.FAIL.getMsg());
private String code;
private String msg;
/**
* 只是一个介绍，对于类 content 的类型 JSON化
*/
private String contentTypeName = "com.byit.dto.specials.SpecialJavaNode";
private T content;
/**
......
byit-myth-executor/myth-executor-plugin/src/main/java/com/byit/utils/JobUtils.java
......@@ -5,6 +5,7 @@ import com.alibaba.fastjson.JSON;
import com.byit.dto.api.JavaCallbackLogDto;
import com.byit.dto.executor.PluginBeanJobInfo;
import com.byit.dto.plugin.*;
import com.byit.dto.specials.SpecialJobParam;
import com.byit.dto.web.ResponseResult;
import com.byit.executor.handler.interfaces.IJobHandler;
import lombok.extern.slf4j.Slf4j;
......@@ -125,6 +126,10 @@ public class JobUtils {
* 补批工作流
*/
public static final String REQUEST_REPAIRFLOW = "/api/flow/repairFlow";
/**
* 特殊的补批接口
*/
public static final String SPECIAL_REQUEST_REPAIRFLOW = "/api/operating/node/specialRunBatch";
private static final String REQUEST_REAL_EXECT = "/api/node/runNode";
......@@ -176,32 +181,34 @@ public class JobUtils {
/**
* 任务的缓存
*/
public static final Map<String,Class<? extends IJobHandler>> jobCache = new ConcurrentHashMap<>();
public static final Map<String, Class<? extends IJobHandler>> jobCache = new ConcurrentHashMap<>();
public static void setRequestUrl(String requestUrl){
public static void setRequestUrl(String requestUrl) {
REQUEST_URL = requestUrl;
}
/**
* 存放token
*
* @param token
*/
public static void setTOKEN(String token){
public static void setTOKEN(String token) {
TOKEN_THREADLOCAL.set(token);
}
/**
* 构建http请求
*
* @param requestUrl
* @param body
* @return
*/
private static String createHttpRequest(String requestUrl, String body){
if (StringUtils.isBlank(REQUEST_URL)){
private static String createHttpRequest(String requestUrl, String body) {
if (StringUtils.isBlank(REQUEST_URL)) {
throw new RuntimeException("请求调度中心的地址尚未设置！");
}
String token = TOKEN_THREADLOCAL.get();
if (StringUtils.isBlank(token)){
if (StringUtils.isBlank(token)) {
throw new RuntimeException("请求的token尚未设置！");
}
HttpRequest httpRequest = HttpRequest.post(REQUEST_URL + requestUrl);
......@@ -211,52 +218,46 @@ public class JobUtils {
}
/**
* 基于工作空间和工作流名称暂停工作流
* @param qualityCheckVo
* 特殊的补批接口
*
* @return
*/
public static ResponseResult pauseTaskByFlowName(QualityCheckVo qualityCheckVo){
log.debug("-------------基于工作流名称暂停：{}----------------------",qualityCheckVo);
return null;
}
public static ResponseResult specialRequestRepairflow(SpecialJobParam specialJobParam) {
log.debug("-------------特殊的补批工作流运行{}----------------------",specialJobParam);
/**
* 基于工作空间和工作流名称开始工作流
* @param qualityCheckVo
* @return
*/
public static ResponseResult startTaskByFlowName(QualityCheckVo qualityCheckVo){
log.debug("-------------基于工作流名称开始：{}----------------------",qualityCheckVo);
return null;
String response = createHttpRequest(FIND_FLOW_RESULT, "param=" + JSON.toJSONString(specialJobParam, WriteClassName));
log.debug("-------------特殊的补批工作流运行完成：{}----------------------",response);
return JSON.parseObject(response, ResponseResult.class);
}
public static ResponseResult findFlowResult(List<JobStatusDto> jobStatusDtos){
public static ResponseResult findFlowResult(List<JobStatusDto> jobStatusDtos) {
log.debug("-------------查询对应工作流的运行状态----------------------");
String response = createHttpRequest(FIND_FLOW_RESULT, "param=" + JSON.toJSONString(jobStatusDtos, WriteClassName));
log.debug("--------------------查询对应工作流的运行状态，结果为：{}------------------------",response);
log.debug("--------------------查询对应工作流的运行状态，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 暂停指定的工作流
*
* @param runId
* @return
*/
public static ResponseResult stopScheduleByRunId (String runId) {
public static ResponseResult stopScheduleByRunId(String runId) {
log.debug("-------------暂停当前运行的节点----------------------");
String response = createHttpRequest(STOP_SCHEDULE_BY_RUNID, "runId=" + runId);
log.debug("--------------------获取当前的运行状态，结果为：{}------------------------",response);
log.debug("--------------------获取当前的运行状态，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 添加一个任务节点
*
* @param pluginBeanJobInfo 任务节点的详尽配置
* @return 添加结果
*/
public static ResponseResult addJob(PluginBeanJobInfo pluginBeanJobInfo){
public static ResponseResult addJob(PluginBeanJobInfo pluginBeanJobInfo) {
log.debug("---------------开始添加一个任务,jobHandelName:{}---------------------", pluginBeanJobInfo.getJobHandelName());
//发送请求 添加任务
String response = createHttpRequest(REQUEST_ADD_JOB_RESOURCES_SUFFIX, JSON.toJSONString(pluginBeanJobInfo));
......@@ -264,18 +265,20 @@ public class JobUtils {
return JSON.parseObject(response, ResponseResult.class);
}
public static ResponseResult javaUpdateStatusRunResult(JavaCallbackLogDto javaCallbackLogDto){
public static ResponseResult javaUpdateStatusRunResult(JavaCallbackLogDto javaCallbackLogDto) {
log.debug("---------------开始修改任务{}---------------------", javaCallbackLogDto);
String httpRequest = createHttpRequest(JAVA_LOG_CALLBACK_DDMP, JSON.toJSONString(javaCallbackLogDto));
log.debug("--------------------修改任务完成，添加结果为：{}------------------------", httpRequest);
return JSON.parseObject(httpRequest, ResponseResult.class);
}
/**
* 发布一个工作流
*
* @param pluginPackage
* @return
*/
public static ResponseResult publish(PluginPackage pluginPackage){
public static ResponseResult publish(PluginPackage pluginPackage) {
log.debug("---------------开始发布工作流,flowName:{}---------------------", pluginPackage.getFlow().getName());
//发送请求 添加任务
......@@ -286,90 +289,97 @@ public class JobUtils {
/**
* 开始工作流
*
* @param flowName
* @param workspaceName
* @return
*/
public static ResponseResult startFlow(String flowName, String workspaceName){
Map<String,String> map = new HashMap<>(5);
map.put("flowName",flowName);
map.put("workspaceName",workspaceName);
public static ResponseResult startFlow(String flowName, String workspaceName) {
Map<String, String> map = new HashMap<>(5);
map.put("flowName", flowName);
map.put("workspaceName", workspaceName);
//发送请求 添加任务
String response = createHttpRequest(REQUEST_FLOW_START, "param="+JSON.toJSONString(map,WriteClassName));
String response = createHttpRequest(REQUEST_FLOW_START, "param=" + JSON.toJSONString(map, WriteClassName));
log.debug("--------------------开始接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 校验工作流是否存在
*
* @param flowName
* @param workspaceName
* @return
*/
public static ResponseResult existFlow(String flowName, String workspaceName){
Map<String,String> map = new HashMap<>(5);
map.put("flowName",flowName);
map.put("workspaceName",workspaceName);
public static ResponseResult existFlow(String flowName, String workspaceName) {
Map<String, String> map = new HashMap<>(5);
map.put("flowName", flowName);
map.put("workspaceName", workspaceName);
//发送请求 添加任务
String response = createHttpRequest(REQUEST_FLOW_EXIST, "param="+JSON.toJSONString(map));
String response = createHttpRequest(REQUEST_FLOW_EXIST, "param=" + JSON.toJSONString(map));
log.debug("--------------------校验工作流是否存在接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 暂停工作流
*
* @param flowName
* @param workspaceName
* @return
*/
public static ResponseResult stopFlow(String flowName, String workspaceName){
Map<String,String> map = new HashMap<>(5);
map.put("flowName",flowName);
map.put("workspaceName",workspaceName);
public static ResponseResult stopFlow(String flowName, String workspaceName) {
Map<String, String> map = new HashMap<>(5);
map.put("flowName", flowName);
map.put("workspaceName", workspaceName);
//发送请求 添加任务
String response = createHttpRequest(REQUEST_FLOW_STOP,"param="+JSON.toJSONString(map,WriteClassName));
log.debug("--------------------暂停接口调用成功，结果为：{}------------------------",response);
String response = createHttpRequest(REQUEST_FLOW_STOP, "param=" + JSON.toJSONString(map, WriteClassName));
log.debug("--------------------暂停接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 删除工作流
*
* @param flowName
* @param workspaceName
* @return
*/
public static ResponseResult deleteFlow(String flowName, String workspaceName){
Map<String,String> map = new HashMap<>(5);
map.put("flowName",flowName);
map.put("workspaceName",workspaceName);
public static ResponseResult deleteFlow(String flowName, String workspaceName) {
Map<String, String> map = new HashMap<>(5);
map.put("flowName", flowName);
map.put("workspaceName", workspaceName);
//发送请求 添加任务
String response = createHttpRequest(REQUEST_FLOW_DELETE,"param="+JSON.toJSONString(map,WriteClassName));
String response = createHttpRequest(REQUEST_FLOW_DELETE, "param=" + JSON.toJSONString(map, WriteClassName));
log.debug("--------------------删除接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 删除工作流
*
* @param flowName
* @param workspaceName
* @return
*/
public static ResponseResult repealSchedule(String flowName, String workspaceName){
Map<String,String> map = new HashMap<>(5);
map.put("flowName",flowName);
map.put("workspaceName",workspaceName);
public static ResponseResult repealSchedule(String flowName, String workspaceName) {
Map<String, String> map = new HashMap<>(5);
map.put("flowName", flowName);
map.put("workspaceName", workspaceName);
//发送请求 添加任务
String response = createHttpRequest(REQUEST_FLOW_REPEAL,"param="+JSON.toJSONString(map,WriteClassName));
log.debug("--------------------删除接口调用成功，结果为：{}------------------------",response);
String response = createHttpRequest(REQUEST_FLOW_REPEAL, "param=" + JSON.toJSONString(map, WriteClassName));
log.debug("--------------------删除接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 重新开始某次调度
*
* @param runIds
* @return
*/
public static ResponseResult startSchedule(String runIds){
log.debug("重新开始调度接口 runids:{}",runIds);
public static ResponseResult startSchedule(String runIds) {
log.debug("重新开始调度接口 runids:{}", runIds);
//发送请求 添加任务
String response = createHttpRequest(REQUEST_FLOW_REREPEAL, "runIds=" + runIds);
log.debug("--------------------重新开始调度接口调用成功，结果为：{}------------------------", response);
......@@ -378,75 +388,81 @@ public class JobUtils {
/**
* 杀死任务
*
* @param runId
* @param flowName
* @param nodeName
* @return
*/
public static ResponseResult killJob(String runId, String flowName, String nodeName){
public static ResponseResult killJob(String runId, String flowName, String nodeName) {
//请求的路径
Map<String,String> map = new HashMap<>(5);
Map<String, String> map = new HashMap<>(5);
map.put("runId", runId);
map.put("flowName", flowName);
map.put("nodeName", nodeName);
//发送请求 添加任务
String response = createHttpRequest(REQUEST_FLOW_KILL_JOB,"param="+JSON.toJSONString(map,WriteClassName));
log.debug("--------------------杀死任务接口调用成功，结果为：{}------------------------",response);
String response = createHttpRequest(REQUEST_FLOW_KILL_JOB, "param=" + JSON.toJSONString(map, WriteClassName));
log.debug("--------------------杀死任务接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 杀死工作流
*
* @param runId
* @return
*/
public static ResponseResult killFlow(String runId, String flowName, String workspaceName){
Map<String,String> map = new HashMap<>(5);
public static ResponseResult killFlow(String runId, String flowName, String workspaceName) {
Map<String, String> map = new HashMap<>(5);
map.put("runId", runId);
map.put("flowName", flowName);
map.put("workspaceName", workspaceName);
//发送请求 添加任务
String response = createHttpRequest(REQUEST_FLOW_KILL_FLOW, "param="+JSON.toJSONString(map,WriteClassName));
log.debug("--------------------杀死工作流接口调用成功，结果为：{}------------------------",response);
String response = createHttpRequest(REQUEST_FLOW_KILL_FLOW, "param=" + JSON.toJSONString(map, WriteClassName));
log.debug("--------------------杀死工作流接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 重跑节点
*
* @return
*/
public static ResponseResult reRunJob(RunInfo runInfo){
public static ResponseResult reRunJob(RunInfo runInfo) {
//发送请求 添加任务
String response = createHttpRequest(REQUEST_FLOW_RERUNJOB,"param="+JSON.toJSONString(runInfo, WriteClassName));
log.debug("--------------------重跑节点接口调用成功，结果为：{}------------------------",response);
String response = createHttpRequest(REQUEST_FLOW_RERUNJOB, "param=" + JSON.toJSONString(runInfo, WriteClassName));
log.debug("--------------------重跑节点接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 重跑工作流
*
* @return
*/
public static ResponseResult reRunFlow(RunInfo runInfo){
public static ResponseResult reRunFlow(RunInfo runInfo) {
//发送请求 添加任务
String response = createHttpRequest(REQUEST_FLOW_RERUNFLOW,"param="+JSON.toJSONString(runInfo, WriteClassName));
log.debug("--------------------重跑节点接口调用成功，结果为：{}------------------------",response);
String response = createHttpRequest(REQUEST_FLOW_RERUNFLOW, "param=" + JSON.toJSONString(runInfo, WriteClassName));
log.debug("--------------------重跑节点接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 手动置为成功
*
* @return
*/
public static ResponseResult makeSuccess(RunInfo runInfo){
public static ResponseResult makeSuccess(RunInfo runInfo) {
//发送请求 添加任务
String response = createHttpRequest(REQUEST_FLOW_MAKESUCCESS,"param="+JSON.toJSONString(runInfo, WriteClassName));
log.debug("--------------------手动置为成功接口调用成功，结果为：{}------------------------",response);
String response = createHttpRequest(REQUEST_FLOW_MAKESUCCESS, "param=" + JSON.toJSONString(runInfo, WriteClassName));
log.debug("--------------------手动置为成功接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 获取运行实例接口
*
* @param startTime 开始时间 毫秒级别时间
* @param endTime 结束时间 毫秒级别时间
* @param workspaceName 工作空间名称
......@@ -455,7 +471,7 @@ public class JobUtils {
* @param executeStatus 执行结果 1 成功 2 失败 3 补批成功 4 补批失败 5.kill 多个以“,”隔开
* @return
*/
public static ResponseResult loadScheduleResult(Long startTime, Long endTime, String workspaceName, String flowName, String scheduleStatus, String executeStatus){
public static ResponseResult loadScheduleResult(Long startTime, Long endTime, String workspaceName, String flowName, String scheduleStatus, String executeStatus) {
Map<String, Object> param = new HashMap<>();
param.put("startTime", startTime);
param.put("endTime", endTime);
......@@ -464,86 +480,91 @@ public class JobUtils {
param.put("scheduleStatus", scheduleStatus);
param.put("executeStatus", executeStatus);
String response = createHttpRequest(REQUEST_LOADSCHEDULE, "param=" + JSON.toJSONString(param));
log.debug("--------------------获取运行实例接口调用成功，结果为：{}------------------------",response);
log.debug("--------------------获取运行实例接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 获取工作流运行实例的节点日志
*
* @param runId
* @param flowName
* @return
*/
public static ResponseResult loadScheduleLog(String runId, String flowName){
public static ResponseResult loadScheduleLog(String runId, String flowName) {
Map<String, Object> param = new HashMap<>();
param.put("runId", runId);
param.put("flowName", flowName);
String response = createHttpRequest(REQUEST_LOADSCHEDULELOG, "param=" + JSON.toJSONString(param));
log.debug("--------------------获取工作流运行实例的节点日志接口调用成功，结果为：{}------------------------",response);
log.debug("--------------------获取工作流运行实例的节点日志接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 获取运行的统计数据接口
*
* @param startTime 开始时间 毫秒级别时间
* @param endTime 开始时间 毫秒级别时间
* @param workspaceName 工作空间名称
* @return
*/
public static ResponseResult loadStatisticData(Long startTime, Long endTime, String workspaceName){
public static ResponseResult loadStatisticData(Long startTime, Long endTime, String workspaceName) {
Map<String, Object> param = new HashMap<>();
param.put("startTime", startTime);
param.put("endTime", endTime);
param.put("workspaceName", workspaceName);
String response = createHttpRequest(REQUEST_LOADSTATISTICDATA, "param=" + JSON.toJSONString(param));
log.debug("--------------------获取运行的统计数据接口调用成功，结果为：{}------------------------",response);
log.debug("--------------------获取运行的统计数据接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 获取节点运行的统计数据
*
* @param startTime 开始时间 毫秒级别时间
* @param endTime 结束时间 毫秒级别时间
* @param workspaceName 工作空间名称
* @param flowName 工作流名称
* @return
*/
public static ResponseResult loadNodeStatisticData(Long startTime, Long endTime, String workspaceName, String flowName){
public static ResponseResult loadNodeStatisticData(Long startTime, Long endTime, String workspaceName, String flowName) {
Map<String, Object> param = new HashMap<>();
param.put("startTime", startTime);
param.put("endTime", endTime);
param.put("workspaceName", workspaceName);
param.put("flowName", flowName);
String response = createHttpRequest(REQUEST_LOADNODESTATISTICDATA, "param=" + JSON.toJSONString(param));
log.debug("--------------------获取节点运行的统计数据接口调用成功，结果为：{}------------------------",response);
log.debug("--------------------获取节点运行的统计数据接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 获取当前的工作流运行状态
*
* @param workspaceName 工作空间名称
* @param flowNames 工作流名称，多个以","分割
* @return
*/
public static ResponseResult loadCurrentStatus(String workspaceName, String flowNames){
public static ResponseResult loadCurrentStatus(String workspaceName, String flowNames) {
Map<String, Object> param = new HashMap<>();
param.put("workspaceName", workspaceName);
param.put("flowNames", flowNames);
String response = createHttpRequest(REQUEST_LOADCURRENTSTATUS, "param=" + JSON.toJSONString(param));
log.debug("--------------------获取当前的工作流运行状态接口调用成功，结果为：{}------------------------",response);
log.debug("--------------------获取当前的工作流运行状态接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 补批工作流
*
* @param workspaceName 工作空间名称
* @param flowName 工作流名称
* @param nodeNames 节点名称 多个节点以","隔开
* @param repairTimes 补批日期 多个日期以","隔开
* @return
*/
public static ResponseResult repairFlow(String workspaceName, String flowName, String nodeNames, String repairTimes){
public static ResponseResult repairFlow(String workspaceName, String flowName, String nodeNames, String repairTimes) {
Map<String, Object> param = new HashMap<>();
param.put("workspaceName", workspaceName);
param.put("flowName", flowName);
......@@ -551,197 +572,210 @@ public class JobUtils {
param.put("repairTimes", repairTimes);
String response = createHttpRequest(REQUEST_REPAIRFLOW, "param=" + JSON.toJSONString(param));
log.debug("--------------------补批工作流接口调用成功，结果为：{}------------------------",response);
log.debug("--------------------补批工作流接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 创建工作空间
*
* @param workspaceName
* @return
*/
public static ResponseResult addWorkspace(String workspaceName){
public static ResponseResult addWorkspace(String workspaceName) {
log.debug("---------------开始创建工作空间,workspaceName:{}---------------------", workspaceName);
//发送请求 添加任务
String response = createHttpRequest(REQUEST_WORKSPACE_ADD, "workspaceName="+ workspaceName);
log.debug("--------------------创建工作空间接口调用成功，结果为：{}------------------------",response);
String response = createHttpRequest(REQUEST_WORKSPACE_ADD, "workspaceName=" + workspaceName);
log.debug("--------------------创建工作空间接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 判断工作空间是否存在
*
* @param workspaceName
* @return
*/
public static ResponseResult existWorkspace(String workspaceName){
public static ResponseResult existWorkspace(String workspaceName) {
log.debug("---------------判断工作空间是否存在,workspaceName:{}---------------------", workspaceName);
//发送请求 添加任务
String response = createHttpRequest(REQUEST_WORKSPACE_EXIST, "workspaceName="+ workspaceName);
log.debug("--------------------判断工作空间是否存在接口调用成功，结果为：{}------------------------",response);
String response = createHttpRequest(REQUEST_WORKSPACE_EXIST, "workspaceName=" + workspaceName);
log.debug("--------------------判断工作空间是否存在接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 立即运行节点
*
* @return
*/
public static ResponseResult realExectNode(RunNode runNode){
public static ResponseResult realExectNode(RunNode runNode) {
log.debug("---------------立即运行节点---------------------");
//立即运行节点
String response = createHttpRequest(REQUEST_REAL_EXECT, "param=" + JSON.toJSONString(runNode));
log.debug("--------------------立即运行节点，结果为：{}------------------------",response);
log.debug("--------------------立即运行节点，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
public static ResponseResult runHistroy(String nodeId){
public static ResponseResult runHistroy(String nodeId) {
log.debug("-------------获取运行历史----------------------");
String response = createHttpRequest(REQUEST_RUN_HISTORY, "nodeId=" + nodeId);
log.debug("--------------------获取运行历史，结果为：{}------------------------",response);
log.debug("--------------------获取运行历史，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 功能描述 添加任务
* @author gml
* @date 2020-04-14 16:50
*
* @param javaTask
* @return com.byit.dto.web.ResponseResult
* @author gml
* @date 2020-04-14 16:50
*/
public static ResponseResult addJavaTask(JavaTask javaTask){
public static ResponseResult addJavaTask(JavaTask javaTask) {
log.debug("-------------添加任务----------------------");
String response = createHttpRequest(REQUEST_ADD_JAVATASK, "param=" + JSON.toJSONString(javaTask, WriteClassName));
log.debug("--------------------添加任务，结果为：{}------------------------",response);
log.debug("--------------------添加任务，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 功能描述 修改任务
* @author gml
* @date 2020-04-14 16:50
*
* @param javaTask
* @return com.byit.dto.web.ResponseResult
* @author gml
* @date 2020-04-14 16:50
*/
public static ResponseResult updateJavaTask(JavaTask javaTask){
public static ResponseResult updateJavaTask(JavaTask javaTask) {
log.debug("-------------修改任务----------------------");
String response = createHttpRequest(REQUEST_UPDATE_JAVATASK, "param=" + JSON.toJSONString(javaTask, WriteClassName));
log.debug("--------------------修改任务，结果为：{}------------------------",response);
log.debug("--------------------修改任务，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 功能描述 删除任务
* @author gml
* @date 2020-04-14 16:50
*
* @param jobName
* @return com.byit.dto.web.ResponseResult
* @author gml
* @date 2020-04-14 16:50
*/
public static ResponseResult deleteJavaTask(String jobName){
public static ResponseResult deleteJavaTask(String jobName) {
log.debug("-------------删除任务----------------------");
String response = createHttpRequest(REQUEST_DELETE_JAVATASK, "jobName=" + jobName);
log.debug("--------------------删除任务，结果为：{}------------------------",response);
log.debug("--------------------删除任务，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 功能描述 判断是否存在任务 存在 true 不存在 false
* @author gml
* @date 2020-04-14 16:50
*
* @param jobName
* @return com.byit.dto.web.ResponseResult
* @author gml
* @date 2020-04-14 16:50
*/
public static ResponseResult existJavaTask(String jobName){
public static ResponseResult existJavaTask(String jobName) {
log.debug("-------------判断是否存在任务----------------------");
String response = createHttpRequest(REQUEST_EXIST_JAVATASK, "jobName=" + jobName);
log.debug("--------------------判断是否存在任务，结果为：{}------------------------",response);
log.debug("--------------------判断是否存在任务，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 功能描述 根据jobName获取运行日志
* @author gml
* @date 2020-04-14 16:53
*
* @param jobName jobName
* @param startTime 开始时间 毫秒级时间戳
* @param endTime 结束时间 毫秒级时间戳
* @return com.byit.dto.web.ResponseResult
* @author gml
* @date 2020-04-14 16:53
*/
public static ResponseResult loadLogByJobName(String jobName, Long startTime, Long endTime){
public static ResponseResult loadLogByJobName(String jobName, Long startTime, Long endTime) {
log.debug("-------------根据jobName获取运行日志----------------------");
Map<String, Object> map = new HashMap<>();
map.put("jobName", jobName);
map.put("startTime", startTime);
map.put("endTime", endTime);
String response = createHttpRequest(REQUEST_LOADLOG_JOBNAME, "param=" + JSON.toJSONString(map));
log.debug("--------------------根据jobName获取运行日志，结果为：{}------------------------",response);
log.debug("--------------------根据jobName获取运行日志，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 功能描述 根据taskName获取运行日志
* @author gml
* @date 2020-04-14 16:53
*
* @param taskName taskName
* @param startTime 开始时间 毫秒级时间戳
* @param endTime 结束时间 毫秒级时间戳
* @return com.byit.dto.web.ResponseResult
* @author gml
* @date 2020-04-14 16:53
*/
public static ResponseResult loadLogByTaskName(String taskName, Long startTime, Long endTime){
public static ResponseResult loadLogByTaskName(String taskName, Long startTime, Long endTime) {
log.debug("-------------根据taskName获取运行日志----------------------");
Map<String, Object> map = new HashMap<>();
map.put("taskName", taskName);
map.put("startTime", startTime);
map.put("endTime", endTime);
String response = createHttpRequest(REQUEST_LOADLOG_TASKNAME, "param=" + JSON.toJSONString(map));
log.debug("--------------------根据taskName获取运行日志，结果为：{}------------------------",response);
log.debug("--------------------根据taskName获取运行日志，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 立即运行quartz任务
*
* @param jobName
* @return
*/
public static ResponseResult runJavaTask(String jobName){
public static ResponseResult runJavaTask(String jobName) {
log.debug("-------------立即运行quartz任务----------------------");
String response = createHttpRequest(REQUEST_RUN_JAVATASK, "jobName=" + jobName);
log.debug("--------------------立即运行quartz任务，结果为：{}------------------------",response);
log.debug("--------------------立即运行quartz任务，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 立即运行quartz任务 不验证是否存在数据库中
*
* @param javaTask
* @return
*/
public static ResponseResult runTask(JavaTask javaTask){
public static ResponseResult runTask(JavaTask javaTask) {
log.debug("-------------立即运行quartz任务----------------------");
String response = createHttpRequest(REQUEST_RUN_TASK, "param=" + JSON.toJSONString(javaTask,WriteClassName));
log.debug("--------------------立即运行quartz任务，结果为：{}------------------------",response);
String response = createHttpRequest(REQUEST_RUN_TASK, "param=" + JSON.toJSONString(javaTask, WriteClassName));
log.debug("--------------------立即运行quartz任务，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 获取当前的运行状态
*
* @param jobNames
* @return
*/
public static ResponseResult loadCurrentStatusByJobName(String jobNames){
public static ResponseResult loadCurrentStatusByJobName(String jobNames) {
log.debug("-------------获取当前的运行状态----------------------");
String response = createHttpRequest(REQUEST_LOADSTATUS_JAVATASK, "jobNames=" + jobNames);
log.debug("--------------------获取当前的运行状态，结果为：{}------------------------",response);
log.debug("--------------------获取当前的运行状态，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 杀死正在运行中的节点
*
* @param logId
* @return
*/
public static ResponseResult killRunNode(Integer logId) {
log.debug("-------------杀死当前运行的节点----------------------");
String response = createHttpRequest(KILL_NODE_RUN_ING, "logId=" + logId);
log.debug("--------------------获取当前的运行状态，结果为：{}------------------------",response);
log.debug("--------------------获取当前的运行状态，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
......
byit-myth-executor/myth-executor-server/src/main/java/com/byit/service/ScriptExecutorServiceImpl.java
......@@ -169,7 +169,6 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
returnResult.setCode(ReturnResult.FAIL.getCode());
returnResult.setMsg(ExecutorLogUtil.getMessage(e));
jobRunResultDto.setReturnResult(returnResult);
}
//设置结束时间
jobRunResultDto.setEndTime(new Date());
......
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