Commit 3406bf80 by guominglei

Merge remote-tracking branch 'origin/developer' into developer

parents d9c55ff7 250d9109
......@@ -138,7 +138,6 @@
<overwrite>true</overwrite>
<skip>false</skip>
</configuration>
</plugin>
</plugins>
</build>
......
......@@ -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);
}
}
......@@ -15,6 +15,7 @@ import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.text.ParseException;
import java.util.List;
import java.util.Map;
/**
* @description: 工作流操作的API接口
......@@ -185,4 +186,11 @@ public class ApiFlowController {
return ResponseResult.ok(collectData);
}
@PostMapping("/loadCurrentStatus")
@ApiOperation("获取当前的工作流运行状态")
public ResponseResult loadCurrentStatus(String param){
Map<String, RunRecording> statusMap = apiFlowService.loadCurrentStatus(param);
return ResponseResult.ok(statusMap);
}
}
......@@ -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 param){
apiNodeService.runTask(param);
return ResponseResult.ok("SUCCESS");
}
@PostMapping("loadCurrentStatusByJobName")
public ResponseResult loadCurrentStatusByJobName(String jobNames){
Map<String, JobTaskRunLog> jobTaskRunLogMap = apiNodeService.loadCurrentStatusByJobName(jobNames);
......
......@@ -8,7 +8,6 @@ import com.byit.model.JobTask;
import com.byit.service.JobTaskService;
import com.byit.service.RunScriptService;
import com.byit.thread.LogCallbackThread;
import com.byit.thread.RunRecordingScanHelper;
import com.byit.util.SourceObj2TargetObjUtil;
import com.byit.utils.ValidationUtil;
import org.springframework.beans.factory.annotation.Autowired;
......@@ -55,8 +54,6 @@ public class JobController {
return jobTaskService.findJobTaskByTriggerNextTimeLessThanEqual(11);
}
@Autowired
private RunRecordingScanHelper runRecordingScanThread;
@GetMapping("test")
public DispatchResponseDto test(){
DispatchResponseDto dispatchResponseDto = runScriptService.runScript(null);
......
......@@ -7,6 +7,7 @@ import com.byit.model.vo.RunRecordingVo;
import java.text.ParseException;
import java.util.List;
import java.util.Map;
/**
* @description: 工作流的api请求业务处理接口
......@@ -95,4 +96,11 @@ public interface ApiFlowService {
* @return
*/
CollectData loadNodeStatisticData(String param);
/**
* 获取工作流的当前状态
* @param param
* @return
*/
Map<String, RunRecording> loadCurrentStatus(String param);
}
......@@ -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
......
......@@ -13,6 +13,6 @@ public class TestServiceImpl {
public DispatchResponseDto test(){
return scriptExecutorService.runPythonScript(null);
return scriptExecutorService.runScript(null);
}
}
......@@ -137,7 +137,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
flow.setWorkspaceId(workspaceId);
flow.setStartUp(FlowPropertyEnum.IS_START.getCode());
flow.setExecType(pluginFlow.getConfig().getExecType());
flow.setPriority(pluginFlow.getConfig().getPriority());
flow.setPriority(StringUtils.isEmpty(pluginFlow.getConfig().getPriority()) || !"2".equals(pluginFlow.getConfig().getPriority()) ? "1" : pluginFlow.getConfig().getPriority());
flow.setAlarmlAction(pluginFlow.getConfig().getAlarmlAction());
flow.setAlarmEmail(pluginFlow.getConfig().getAlarmEmail());
//设置可执行次数
......@@ -275,7 +275,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
node.setTriggerNextTime(StringUtils.isEmpty(((PluginNode)pluginNode).getConfig().getNodeCron()) ? null : new CronExpression(((PluginNode)pluginNode).getConfig().getNodeCron()).getNextValidTimeAfter(new Date()).getTime());
node.setPluginUrls(((PluginNode)pluginNode).getConfig().getPluginUrls());
node.setRoutingStrategy(StringUtils.isEmpty(((PluginNode)pluginNode).getConfig().getRoutingStrategy()) ? "RANDOM" : ((PluginNode)pluginNode).getConfig().getRoutingStrategy());
node.setPriority(StringUtils.isEmpty(((PluginNode)pluginNode).getConfig().getPriority()) ? "1" : ((PluginNode)pluginNode).getConfig().getPriority());
node.setPriority(StringUtils.isEmpty(((PluginNode)pluginNode).getConfig().getPriority()) || !"2".equals(((PluginNode)pluginNode).getConfig().getPriority()) ? "1" : ((PluginNode)pluginNode).getConfig().getPriority());
node.setScriptUrls(((PluginNode)pluginNode).getScriptUrls());
//设置失败重试
if (null != ((PluginNode)pluginNode).getConfig().getFailedRetryCount()){
......@@ -893,6 +893,36 @@ public class ApiFlowServiceImpl implements ApiFlowService {
return collectData;
}
@Override
public Map<String, RunRecording> loadCurrentStatus(String param) {
ValidationUtil.dataNotBank(param, "请求参数不允许为空!");
JSONObject jsonObject = JSON.parseObject(param);
//获取工作空间名称
String workspaceName = jsonObject.getString("workspaceName");
ValidationUtil.dataNotBank(workspaceName, "工作空间名称不允许为空!");
Workspace workspace = workspaceMapper.getByName(workspaceName);
ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
//获取工作流名称
String flowNames = jsonObject.getString("flowNames");
ValidationUtil.dataNotBank(flowNames, "工作流名称不允许为空!");
List<String> flowNameList = Arrays.asList(flowNames.split(","));
List<Flow> flowList = new ArrayList<>();
flowNameList.forEach(flowName->{
Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
flowList.add(flow);
});
Map<String, RunRecording> statusMap = new HashMap<>();
flowList.forEach(flow -> {
RunRecording runRecording = runRecordingMapper.findNewStatus(flow.getFlowId());
statusMap.put(flow.getFlowName(), runRecording);
});
return statusMap;
}
/**
* runState 补批机制 1 补批当前节点 2 补批当前节点及以下节点
* @param param
......
......@@ -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不允许为空");
......
package com.byit.view;
import com.byit.model.vo.FlowImportantAllVo;
import com.byit.model.vo.FlowVersionDetailedVo;
import com.byit.model.vo.FlowVersionVo;
import com.byit.service.FlowVersionService;
import com.byit.model.vo.FlowVersionVoDep;
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/version")
public class FlowVersionViewController {
private final FlowVersionService flowVersionService;
public FlowVersionViewController(FlowVersionService flowVersionService) {
this.flowVersionService = flowVersionService;
}
@PostMapping("getAllFlowVersion")
public List<FlowVersionVoDep> getAllFlowVersion(){
return flowVersionService.findAllFlow();
}
@PostMapping("findAllVersion")
public List<FlowVersionVo> findAllVersion(){
return flowVersionService.findAllVersion();
}
@RequestMapping("findAllByFlowId")
public List<FlowVersionDetailedVo> findAllByFlowId(Integer flowId) {
return flowVersionService.findAllByFlowId(flowId);
}
}
package com.byit.view;
import com.byit.model.Flow;
import com.byit.model.vo.FlowImportantAllVo;
import com.byit.model.vo.FlowViewVo;
import com.byit.service.FlowService;
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 FlowService flowService;
public FlowViewController(FlowService flowService) {
this.flowService = flowService;
}
@PostMapping("findAllThisVersionFlow")
public List<Flow> findAllThisVersionFlow(){
return flowService.findAllFlow();
}
@PostMapping("findAllFlowViewVo")
public List<FlowViewVo> findAllFlowViewVo(){
return flowService.findAllFlowViewVo();
}
/**
* 查询工作流摘要
* @return
*/
@PostMapping("findAllFlowImportant")
public List<FlowImportantAllVo> findAllFlowImportant() {
return flowService.findAllFlowImportant();
}
}
package com.byit.view;
import com.byit.model.NodeVersion;
import com.byit.model.vo.FlowNodeVersionVo;
import com.byit.service.NodeVersionService;
import org.apache.ibatis.annotations.Param;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
/**
* @author huangfu
*/
@RestController
@RequestMapping("view/node")
public class NodeVersionViewController {
private final NodeVersionService nodeVersionService;
public NodeVersionViewController(NodeVersionService nodeVersionService) {
this.nodeVersionService = nodeVersionService;
}
@RequestMapping("/getFlowNodeVersionVo")
public List<FlowNodeVersionVo> getFlowNodeVersionVo(){
return nodeVersionService.findAll();
}
@RequestMapping("/findAllNodeVersionByFlowId")
public List<NodeVersion> findAllNodeVersionByFlowId(Integer id){
return nodeVersionService.findAllByFlowId(id);
}
}
package com.byit.view;
import com.byit.model.Node;
import com.byit.service.FlowService;
import com.byit.service.NodeService;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
/**
* @author huangfu
*/
@RestController
@RequestMapping("view/node")
public class NodeViewController {
private final NodeService nodeService;
public NodeViewController(NodeService nodeService) {
this.nodeService = nodeService;
}
@RequestMapping("findNodes")
public List<Node> findNodes(Integer flowId){
return nodeService.findNodeByFlowIdAndVersionName(flowId);
}
}
package com.byit.view;
import com.byit.model.vo.RunLogVo;
import com.byit.service.JobTaskRunLogService;
import org.apache.ibatis.annotations.Param;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
/**
* @author huangfu
*/
@RestController
@RequestMapping("view/log")
public class RunLogViewController {
private final JobTaskRunLogService jobTaskRunLogService;
public RunLogViewController(JobTaskRunLogService jobTaskRunLogService) {
this.jobTaskRunLogService = jobTaskRunLogService;
}
@RequestMapping("findAllRunLog")
public List<RunLogVo> findAllRunLogIdByFlowIdAndRunId(@Param("flowId") Integer flowId, @Param("runId") String runId){
return jobTaskRunLogService.findAllRunLogIdByFlowIdAndRunId(flowId,runId);
}
}
package com.byit.view;
import com.byit.model.RunRecording;
import com.byit.model.vo.RunRecordingViewVo;
import com.byit.service.RunRecordingService;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
/**
* 运行记录视图查询
* @author huangfu
*/
@RestController
@RequestMapping("view/runRecording")
public class RunRecordingViewController {
private final RunRecordingService runRecordingService;
public RunRecordingViewController(RunRecordingService runRecordingService) {
this.runRecordingService = runRecordingService;
}
@RequestMapping("findAll")
public List<RunRecording> findAll(){
return runRecordingService.findAll();
}
@RequestMapping("findAllRunIng")
public List<RunRecording> findAllRunIng() {
return runRecordingService.findAllRunIng();
}
@RequestMapping("findAllRunRecordingViewVo")
public List<RunRecordingViewVo> findAllRunRecordingViewVo(){
return runRecordingService.findAllRunRecordingViewVo();
}
@RequestMapping("findAllError")
public List<RunRecordingViewVo> findAllError(){
return runRecordingService.findAllError();
}
}
......@@ -32,7 +32,7 @@ mybatis:
myth-rpc:
registry:
address: http://localhost:8080/myth-register
address: http://10.0.120.208:8080/myth-register
env: dev
biz: byit-myth-job
logging:
......@@ -54,6 +54,14 @@ myth-job:
filestystem: FASTDFS
snapshoot-date: 60 #快照的保存时间 单位天
fdfs:
so-timeout: 1500
connect-timeout: 600
pool:
jmx-enabled: false
tracker-list:
- 10.0.120.2:22122
myth:
plugin:
env: ${myth-rpc.registry.env}
......
......@@ -57,6 +57,13 @@ file:
myth-job:
filestystem: FASTDFS
snapshoot-date: 60 #快照的保存时间 单位天
fdfs:
so-timeout: 1500
connect-timeout: 600
pool:
jmx-enabled: false
tracker-list:
- 10.0.120.2:22122
myth:
plugin:
......
spring:
datasource:
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://${CM_IP}/myth-job?Unicode=true&characterEncoding=UTF-8&useSSL=true
username: ${CM_USER}
password: ${CM_PWD}
url: jdbc:mysql://10.0.120.30:3307/myth-job?Unicode=true&characterEncoding=UTF-8&useSSL=true
username: root
password: root
redis:
database: 0
host: ${Redis_IP}
port: ${Redis_port}
host: 10.0.120.208
port: 6379
password:
timeout: 3000
pool:
......@@ -31,7 +31,7 @@ mybatis:
myth-rpc:
registry:
address: http://${Eureka_IP}/myth-register
address: http://10.0.120.208:8080/myth-register
env: pro
biz: byit-myth-job
logging:
......@@ -52,3 +52,13 @@ file:
myth-job:
filestystem: FASTDFS
snapshoot-date: 60 #快照的保存时间 单位天
fdfs:
so-timeout: 1500
connect-timeout: 600
pool:
jmx-enabled: false
tracker-list:
- 10.0.120.216:22122
- 10.0.120.217:22122
- 10.0.120.218:22122
\ No newline at end of file
......@@ -53,3 +53,19 @@ file:
myth-job:
filestystem: FASTDFS
snapshoot-date: 60 #快照的保存时间 单位天
fdfs:
so-timeout: 1500
connect-timeout: 600
pool:
jmx-enabled: false
tracker-list:
- 10.0.120.216:22122
- 10.0.120.217:22122
- 10.0.120.218:22122
myth:
plugin:
env: ${myth-rpc.registry.env}
biz: ${myth-rpc.registry.biz}
register:
url: ${myth-rpc.registry.address}
\ No newline at end of file
......@@ -21,15 +21,6 @@ import java.util.Map;
@Slf4j
public class MythJobScheduler implements InitializingBean, DisposableBean {
/**
* private final JobScheduleHelper jobScheduleHelper
* private final EmailScanHelper emailScanHelper
* private final FlowScanHelper flowScanHelper
* private final LogScanHelper logScanHelper
* private final RunRecordingScanHelper runRecordingScanHelper
*/
private final ApplicationContext applicationContext;
......
......@@ -5,8 +5,8 @@ package com.byit.enums;
*/
public enum EmailEnum {
IS_ALARM_YES("0","已经告警"),
IS_ALARM_NO("1","没有告警")
IS_ALARM_YES("1","已经告警"),
IS_ALARM_NO("0","没有告警")
;
private String code;
private String msg;
......
......@@ -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);
}
......
......@@ -6,7 +6,11 @@ import org.apache.ibatis.annotations.Param;
import java.util.List;
public interface FlowMapper {
/**
* 查询全部的任务流数据
* @return
*/
List<Flow> findAllFlow();
/**
* 根据ID查询
* @param id
......@@ -54,4 +58,5 @@ public interface FlowMapper {
* @return
*/
List<Flow> findAll();
}
\ No newline at end of file
package com.byit.mapper;
import com.byit.model.FlowVersion;
import org.springframework.stereotype.Repository;
import java.util.List;
@Repository
public interface FlowVersionMapper {
List<FlowVersion> findAllByFlowId(Integer flowId);
/**
* 查询全部的工作流
* @return
*/
List<FlowVersion> findAllFlow();
int deleteById(Integer flowVersionId);
int insertSelective(FlowVersion record);
......
package com.byit.mapper;
import com.byit.model.NodeVersion;
import org.springframework.stereotype.Repository;
import java.util.List;
@Repository
public interface NodeVersionMapper {
/**
* 查询全部数据 根据工作里ID
* @param flowId
* @return
*/
List<NodeVersion> findAllByFlowId(Integer flowId);
/**
* 查询全部的节点版本
* @return
*/
List<NodeVersion> findAll();
int deleteById(Integer nodeVersionId);
int insertSelective(NodeVersion record);
......
......@@ -15,6 +15,23 @@ import java.util.List;
@Repository
public interface RunRecordingMapper {
/**
* 查询全部的数据
* @return
*/
List<RunRecording> findAll();
/**
* 查询全部错误的节点
* @return
*/
List<RunRecording> findAllError();
/**
* 查询运行中的数据
* @return
*/
List<RunRecording> findAllRunIng();
/**
* 查询已经完结的,并且没有告警的任务流
* @return
*/
......@@ -180,4 +197,11 @@ public interface RunRecordingMapper {
* @return
*/
List<RunRecording> findByPreTime(@Param("preHourDate")Date preHourDate, @Param("flowId")Integer flowId);
/**
* 根据flowid获取最新的运行实例
* @param flowId
* @return
*/
RunRecording findNewStatus(Integer flowId);
}
\ No newline at end of file
......@@ -124,7 +124,7 @@ public class RunRecording implements Serializable {
/**
* 是否是内嵌工作流 0 否, 1 是
*/
@ApiModelProperty("是否是内嵌工作流 0 否, 1 是")
@ApiModelProperty("是否是内嵌工作流 1 否, 0 是")
private String isInner;
/**
......
package com.byit.model.vo;
import lombok.Data;
/**
* 版本管理 使用的vo
* @author huangfu
*/
@Data
public class FlowImportantAllVo {
private String rowKey;
private String flowName;
private Integer flowId;
private boolean hasChildren;
}
package com.byit.model.vo;
import com.byit.model.NodeVersion;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.util.List;
/**
* 工作流对应节点版本的工作流
* @author huangfu
*/
@Data
@ApiModel("工作流对应节点版本的工作流")
@Deprecated
public class FlowNodeVersionVo {
@ApiModelProperty("工作流名称")
private String flowName;
@ApiModelProperty("版本名称")
private String versionName;
@ApiModelProperty("对应的节点的集合")
private List<NodeVersion> nodeVersions;
}
package com.byit.model.vo;
import com.byit.model.FlowVersion;
import lombok.Data;
import lombok.EqualsAndHashCode;
/**
*
* @author huangfu
*/
@EqualsAndHashCode(callSuper = true)
@Data
public class FlowVersionDetailedVo extends FlowVersion {
private String rowKey;
}
package com.byit.model.vo;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
/**
* 工作流版本名称
* @author huangfu
*/
@Data
@ApiModel("工作流版本名称")
public class FlowVersionVo {
@ApiModelProperty("ROW-KEY VUE使用")
private String nodeId;
@ApiModelProperty("工作流版本呢的id")
private Integer flowVersionId;
@ApiModelProperty("工作流名称")
private String flowName;
@ApiModelProperty("工作流版本名称")
private String versionName;
@ApiModelProperty("是否存在节点")
private boolean hasChildren;
}
package com.byit.model.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
* Search {@link FlowVersionVo}
*/
@ApiModel("工作流版本的视图载体")
@Data
@Deprecated
public class FlowVersionVoDep implements Serializable {
/**
* 工作流名字
*/
@ApiModelProperty("工作流的名字")
private String flowName;
/**
* 对应的版本
*/
@ApiModelProperty("对应的工作流的所由版本")
private List<FlowVersion> children;
}
package com.byit.model.vo;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
/**
* 工作流视图的试图映射
* @author huangfu
*/
@Data
@ApiModel("工作流视图的试图映射")
public class FlowViewVo {
@ApiModelProperty("row-key vue使用")
private String nodeId;
@ApiModelProperty("工作流id")
private Integer flowId;
@ApiModelProperty("工作流名字")
private String flowName;
@ApiModelProperty("版本名称")
private String versionName;
@ApiModelProperty("是否存在节点")
private boolean hasChildren;
}
package com.byit.model.vo;
import lombok.Data;
import java.util.Date;
/**
* @author huangfu
*/
@Data
public class RunLogVo {
private String rowKey;
private String nodeName;
private String jobType;
private Integer failedRemainingCount;
private String isVirtual;
private Integer mapFlowId;
private String triggerCode;
private String triggerMsg;
private String runCode;
private String runMsg;
private Date triggerTime;
private String jobGroupIp;
private Date startTime;
private Date endTime;
private String logRemotelyPath;
private String scriptUrls;
private Integer scheduleType;
private String operator;
private boolean hasTrigger;
}
package com.byit.model.vo;
import lombok.Data;
/**
* 运行实例
* @author huangfu
*/
@Data
public class RunRecordingViewVo {
private String rowKey;
private Integer flowId;
private String runId;
private String flowName;
private String flowVersionName;
private String flowStatus;
private String flowRunResult;
private boolean hasChildren;
}
package com.byit.service;
import com.byit.model.Flow;
import com.byit.model.vo.FlowImportantAllVo;
import com.byit.model.vo.FlowViewVo;
import com.byit.model.vo.FlowVo;
import com.byit.model.vo.NodeVo;
import org.apache.ibatis.annotations.Param;
......@@ -13,6 +15,19 @@ import java.util.List;
* @create: 2019-12-23 17:33
*/
public interface FlowService {
/**
* 查询全部的任务流数据
* @return
*/
List<FlowViewVo> findAllFlowViewVo();
List<FlowImportantAllVo> findAllFlowImportant();
/**
* 查询全部的任务流数据
* @return
*/
List<Flow> findAllFlow();
/**
* 根据ID查询
* @param id
......
package com.byit.service;
import com.byit.model.FlowVersion;
import com.byit.model.vo.FlowImportantAllVo;
import com.byit.model.vo.FlowVersionDetailedVo;
import com.byit.model.vo.FlowVersionVo;
import com.byit.model.vo.FlowVersionVoDep;
import java.util.List;
/**
* @author huangfu
*/
public interface FlowVersionService {
/**
* 查询全部的工作流
* @return
*/
List<FlowVersionVoDep> findAllFlow();
/**
* 查询全部的工作流
* @return
*/
List<FlowVersionVo> findAllVersion();
List<FlowVersionDetailedVo> findAllByFlowId(Integer flowId);
}
......@@ -2,6 +2,7 @@ package com.byit.service;
import com.byit.model.JobTaskRunLog;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.vo.RunLogVo;
import java.util.List;
......@@ -13,6 +14,8 @@ import java.util.List;
**/
public interface JobTaskRunLogService {
/**
* 查询没有结束的节点
* @param nodIds
......@@ -33,6 +36,15 @@ public interface JobTaskRunLogService {
* @return
*/
List<JobTaskRunLogWithBLOBs> findJobTaskRunLogWithBLOBsByFlowIdAndRunId(Integer flowId,String runId);
/**
* 根据工作流id和日志ID
* @param flowId
* @param runId
* @return
*/
List<RunLogVo> findAllRunLogIdByFlowIdAndRunId(Integer flowId,String runId);
/**
* 查询已经结束或者失败的节点
* @return
......
package com.byit.service;
import com.byit.model.NodeVersion;
import com.byit.model.vo.FlowNodeVersionVo;
import java.util.List;
/**
* 节点版本的业务
* @author huangfu
*/
public interface NodeVersionService {
/**
* 查询全部的节点版本
* @return 全部
*/
List<FlowNodeVersionVo> findAll();
/**
* 查询全部数据 根据工作里ID
* @param flowId
* @return
*/
List<NodeVersion> findAllByFlowId(Integer flowId);
}
package com.byit.service;
import com.byit.model.RunRecording;
import com.byit.model.vo.RunRecordingViewVo;
import java.util.List;
......@@ -10,6 +11,28 @@ import java.util.List;
*/
public interface RunRecordingService {
/**
* 查询全部数据 映射成VO
* @return
*/
List<RunRecordingViewVo> findAllRunRecordingViewVo();
/**
* 查询全部的数据
* @return
*/
List<RunRecording> findAll();
/**
* 查询全部错误的节点
* @return
*/
List<RunRecordingViewVo> findAllError();
/**
* 查询运行中的数据
* @return
*/
List<RunRecording> findAllRunIng();
/**
* 查询已经完结的,并且没有告警的任务流
* @return
*/
......
......@@ -10,6 +10,8 @@ import com.byit.enums.NodePropertyEnum;
import com.byit.job.utils.CronExpression;
import com.byit.mapper.*;
import com.byit.model.*;
import com.byit.model.vo.FlowImportantAllVo;
import com.byit.model.vo.FlowViewVo;
import com.byit.model.vo.FlowVo;
import com.byit.model.vo.NodeVo;
import com.byit.service.FlowService;
......@@ -23,6 +25,7 @@ import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;
/**
* @description: 工作流业务逻辑实现类
......@@ -49,6 +52,39 @@ public class FlowServiceImpl implements FlowService {
@Resource
private WorkspaceMapper workspaceMapper;
@Override
public List<FlowViewVo> findAllFlowViewVo() {
List<Flow> allFlow = findAllFlow();
return allFlow.stream().map(flow -> {
FlowViewVo flowViewVo = new FlowViewVo();
flowViewVo.setFlowId(flow.getFlowId());
flowViewVo.setFlowName(flow.getFlowName());
flowViewVo.setNodeId(UUID.randomUUID().toString().replace("", "-"));
flowViewVo.setVersionName(flow.getVersionName());
flowViewVo.setHasChildren(true);
return flowViewVo;
}).collect(Collectors.toList());
}
@Override
public List<FlowImportantAllVo> findAllFlowImportant() {
List<Flow> all = flowMapper.findAll();
return all.stream().map(flow -> {
FlowImportantAllVo flowImportantAllVo = new FlowImportantAllVo();
flowImportantAllVo.setRowKey(UUID.randomUUID().toString().replace("-",""));
flowImportantAllVo.setHasChildren(true);
flowImportantAllVo.setFlowName(flow.getFlowName());
flowImportantAllVo.setFlowId(flow.getFlowId());
return flowImportantAllVo;
}).collect(Collectors.toList());
}
@Override
public List<Flow> findAllFlow() {
return flowMapper.findAll();
}
@Override
public Flow findFlowById(Integer id) {
return flowMapper.getById(id);
......
package com.byit.service.impl;
import cn.hutool.core.util.IdcardUtil;
import com.byit.mapper.FlowVersionMapper;
import com.byit.model.FlowVersion;
import com.byit.model.vo.FlowImportantAllVo;
import com.byit.model.vo.FlowVersionDetailedVo;
import com.byit.model.vo.FlowVersionVo;
import com.byit.service.FlowVersionService;
import com.byit.model.vo.FlowVersionVoDep;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* 工作流版本业务的实现
* @author huangfu
*/
@Service
public class FlowVersionServiceImpl implements FlowVersionService {
private final FlowVersionMapper flowVersionMapper;
public FlowVersionServiceImpl(FlowVersionMapper flowVersionMapper) {
this.flowVersionMapper = flowVersionMapper;
}
@Override
public List<FlowVersionVoDep> findAllFlow() {
List<FlowVersion> allFlow = flowVersionMapper.findAllFlow();
Map<String, List<FlowVersion>> versionColl = allFlow.stream().collect(Collectors.groupingBy(FlowVersion::getFlowName));
List<FlowVersionVoDep> flowVersionVoDeps = new ArrayList<>();
versionColl.forEach((key,value) ->{
FlowVersionVoDep flowVersionVoDep = new FlowVersionVoDep();
flowVersionVoDep.setFlowName(key);
flowVersionVoDep.setChildren(value);
flowVersionVoDeps.add(flowVersionVoDep);
});
return flowVersionVoDeps;
}
@Override
public List<FlowVersionVo> findAllVersion() {
List<FlowVersion> allFlow = flowVersionMapper.findAllFlow();
return allFlow.stream().map(flowVersion -> {
FlowVersionVo flowVersionVo = new FlowVersionVo();
flowVersionVo.setNodeId(UUID.randomUUID().toString().replace("-",""));
flowVersionVo.setFlowVersionId(flowVersion.getFlowVersionId());
flowVersionVo.setFlowName(flowVersion.getFlowName());
flowVersionVo.setVersionName(flowVersion.getVersionName());
flowVersionVo.setHasChildren(true);
return flowVersionVo;
}).collect(Collectors.toList());
}
@Override
public List<FlowVersionDetailedVo> findAllByFlowId(Integer flowId) {
List<FlowVersion> allByFlowId = flowVersionMapper.findAllByFlowId(flowId);
return allByFlowId.stream().map(e ->{
FlowVersionDetailedVo flowVersionDetailedVo = new FlowVersionDetailedVo();
flowVersionDetailedVo.setRowKey(UUID.randomUUID().toString().replace("-",""));
BeanUtils.copyProperties(e,flowVersionDetailedVo);
return flowVersionDetailedVo;
}).collect(Collectors.toList());
}
}
package com.byit.service.impl;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.mapper.JobTaskRunLogMapper;
import com.byit.model.JobTask;
import com.byit.model.JobTaskRunLog;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.vo.RunLogVo;
import com.byit.service.JobTaskRunLogService;
import com.byit.service.JobTaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
/**
* @program: byit-myth-job->JobTaskRunLogServiceImpl
......@@ -24,10 +32,12 @@ import java.util.List;
public class JobTaskRunLogServiceImpl implements JobTaskRunLogService {
private final JobTaskRunLogMapper jobTaskRunLogMapper;
private final JobTaskService jobTaskService;
@Autowired
public JobTaskRunLogServiceImpl(JobTaskRunLogMapper jobTaskRunLogMapper) {
public JobTaskRunLogServiceImpl(JobTaskRunLogMapper jobTaskRunLogMapper, JobTaskService jobTaskService) {
this.jobTaskRunLogMapper = jobTaskRunLogMapper;
this.jobTaskService = jobTaskService;
}
@Override
......@@ -46,6 +56,27 @@ public class JobTaskRunLogServiceImpl implements JobTaskRunLogService {
}
@Override
public List<RunLogVo> findAllRunLogIdByFlowIdAndRunId(Integer flowId, String runId) {
List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBsByFlowIdAndRunId = jobTaskRunLogMapper.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
List<RunLogVo> collect = jobTaskRunLogWithBLOBsByFlowIdAndRunId.stream().map(log -> {
RunLogVo runLogVo = new RunLogVo();
runLogVo.setHasTrigger(true);
runLogVo.setRowKey(UUID.randomUUID().toString().replace("-", ""));
BeanUtils.copyProperties(log, runLogVo);
return runLogVo;
}).collect(Collectors.toList());
//这个是还未运行的信息
List<JobTask> jobTaskByRunId = jobTaskService.findJobTaskByRunId(runId, flowId);
jobTaskByRunId.forEach(jobTask -> {
RunLogVo runLogVo = new RunLogVo();
runLogVo.setHasTrigger(false);
runLogVo.setRowKey(UUID.randomUUID().toString().replace("-",""));
collect.add(runLogVo);
});
return collect;
}
@Override
public List<JobTaskRunLog> findJobTaskRunLogEndOrFailureNode() {
return jobTaskRunLogMapper.findJobTaskRunLogEndOrFailureNode();
}
......
package com.byit.service.impl;
import com.byit.mapper.FlowVersionMapper;
import com.byit.mapper.NodeVersionMapper;
import com.byit.model.FlowVersion;
import com.byit.model.NodeVersion;
import com.byit.model.vo.FlowNodeVersionVo;
import com.byit.service.FlowVersionService;
import com.byit.service.NodeVersionService;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
/**
* 节点版本的业务
* @author huangfu
*/
@Service
public class NodeVersionServiceImpl implements NodeVersionService {
private final NodeVersionMapper nodeVersionMapper;
private final FlowVersionMapper flowVersionMapper;
public NodeVersionServiceImpl(NodeVersionMapper nodeVersionMapper, FlowVersionMapper flowVersionMapper) {
this.nodeVersionMapper = nodeVersionMapper;
this.flowVersionMapper = flowVersionMapper;
}
@Override
public List<FlowNodeVersionVo> findAll() {
List<FlowNodeVersionVo> flowNodeVersionVos = new ArrayList<>();
List<NodeVersion> nodeVersionMapperAll = nodeVersionMapper.findAll();
List<FlowVersion> allFlow = flowVersionMapper.findAllFlow();
allFlow.forEach(flow ->{
FlowNodeVersionVo flowNodeVersionVo = new FlowNodeVersionVo();
flowNodeVersionVo.setFlowName(flow.getFlowName());
flowNodeVersionVo.setVersionName(flow.getVersionName());
List<NodeVersion> nodeVersions = nodeVersionMapperAll.stream()
.filter(nodeVersion -> flow.getFlowId().equals(nodeVersion.getFlowId()))
.collect(Collectors.toList());
flowNodeVersionVo.setNodeVersions(nodeVersions);
flowNodeVersionVos.add(flowNodeVersionVo);
});
return flowNodeVersionVos;
}
@Override
public List<NodeVersion> findAllByFlowId(Integer flowId) {
return nodeVersionMapper.findAllByFlowId(flowId);
}
}
......@@ -3,11 +3,15 @@ package com.byit.service.impl;
import com.byit.enums.RunRecordingEnum;
import com.byit.mapper.RunRecordingMapper;
import com.byit.model.RunRecording;
import com.byit.model.vo.RunRecordingViewVo;
import com.byit.service.RunRecordingService;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
/**
* @program: byit-myth-job->RunRecordingServiceImpl
......@@ -25,6 +29,40 @@ public class RunRecordingServiceImpl implements RunRecordingService {
}
@Override
public List<RunRecordingViewVo> findAllRunRecordingViewVo() {
List<RunRecording> runRecordings = findAll();
return runRecordings.stream().map(runRecording -> {
RunRecordingViewVo runRecordingViewVo = new RunRecordingViewVo();
runRecordingViewVo.setHasChildren(true);
BeanUtils.copyProperties(runRecording,runRecordingViewVo);
runRecordingViewVo.setRowKey(UUID.randomUUID().toString().replace("-",""));
return runRecordingViewVo;
}).collect(Collectors.toList());
}
@Override
public List<RunRecording> findAll() {
return runRecordingMapper.findAll();
}
@Override
public List<RunRecordingViewVo> findAllError() {
List<RunRecording> allError = runRecordingMapper.findAllError();
return allError.stream().map(runRecording -> {
RunRecordingViewVo runRecordingViewVo = new RunRecordingViewVo();
runRecordingViewVo.setHasChildren(true);
BeanUtils.copyProperties(runRecording,runRecordingViewVo);
runRecordingViewVo.setRowKey(UUID.randomUUID().toString().replace("-",""));
return runRecordingViewVo;
}).collect(Collectors.toList());
}
@Override
public List<RunRecording> findAllRunIng() {
return runRecordingMapper.findAllRunIng();
}
@Override
public List<RunRecording> findRunRecordingByEndAndNotIsAlarm() {
return runRecordingMapper.findRunRecordingByEndAndNotIsAlarm();
}
......
......@@ -16,6 +16,6 @@ public class RunScriptServiceImpl implements RunScriptService {
private ScriptExecutorService scriptExecutorService;
@Override
public DispatchResponseDto runScript(ScriptDto scriptDto) {
return scriptExecutorService.runPythonScript(scriptDto);
return scriptExecutorService.runScript(scriptDto);
}
}
......@@ -12,7 +12,6 @@ import com.byit.service.RunRecordingService;
import com.byit.service.mapservice.RunRecordingAndEmailService;
import com.byit.util.TimeFormatUtil;
import lombok.extern.slf4j.Slf4j;
import org.csource.common.MyException;
import org.springframework.stereotype.Component;
import java.io.IOException;
......@@ -108,7 +107,7 @@ public class RunRecordingAndEmailServiceImpl implements RunRecordingAndEmailServ
if(null != jobTaskRunLog.getLogRemotelyPath()){
logStr = new String(fileSystem.downloaderFile(jobTaskRunLog.getLogRemotelyPath()), StandardCharsets.UTF_8);
}
} catch (IOException | MyException e) {
} catch (IOException e) {
e.printStackTrace();
}
stringBuilder.append("<tr align='center'>")
......
......@@ -4,6 +4,7 @@ import com.byit.conf.MythJobAutoConfigure;
import com.byit.dto.plugin.JavaTask;
import com.byit.enums.NodeRunStatusPropertyEnum;
import com.byit.enums.ScheduleTypeEnum;
import com.byit.enums.task.RunResultEnum;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.packet.request.PluginRpcRequestPacket;
import com.byit.packet.response.PluginRpcResponsePacket;
......@@ -22,6 +23,7 @@ import java.util.Date;
* @author huangfu
*/
public class JavaTaskJobTask implements TimerTask {
public static final String JAVA_SYNC = "JAVA_SYNC";
private final JavaTask javaTask;
public JavaTaskJobTask(JavaTask javaTask) {
......@@ -29,7 +31,7 @@ public class JavaTaskJobTask implements TimerTask {
}
@Override
public void run(Timeout timeout) throws Exception {
public void run(Timeout timeout) {
MythJobAutoConfigure.LOW_LEVEL_JOB_THREAD_POOL.execute(this::runJob);
}
......@@ -55,15 +57,14 @@ public class JavaTaskJobTask implements TimerTask {
try {
PluginRpcResponsePacket pluginRpcResponsePacket = service.runJava(request);
if(pluginRpcResponsePacket.isStatus()){
log.setTriggerCode(NodeRunStatusPropertyEnum.RUN_SUCCESS.getCode());
log.setTriggerCode(RunResultEnum.TRIGGER_SUCCESS.getCode());
}else{
log.setTriggerCode(NodeRunStatusPropertyEnum.RUN_FAILURE.getCode());
log.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
}
log.setTriggerMsg(pluginRpcResponsePacket.getMsg());
}catch (Exception e){
log.setTriggerCode("2");
log.setRunCode(NodeRunStatusPropertyEnum.RUN_FAILURE.getCode());
log.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
log.setRunCode(RunResultEnum.RUN_ERROR.getCode());
log.setRunMsg(javaTask.getTaskName()+":"+e.getMessage());
log.setTriggerMsg(javaTask.getTaskName()+":"+e.getMessage());
}
......@@ -81,14 +82,9 @@ public class JavaTaskJobTask implements TimerTask {
log.setHandlerName(javaTask.getTaskName());
log.setRunParams(javaTask.getParam());
log.setTriggerTime(new Date());
log.setJobType("JAVA_SYNC");
log.setJobType(JAVA_SYNC);
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
int i = jobTaskRunLogService.saveJobTaskRunLog(log);
return log.getLogId();
}
public JavaTask getJavaTask() {
return javaTask;
}
}
......@@ -2,17 +2,18 @@ package com.byit.task;
import com.alibaba.fastjson.JSON;
import com.byit.conf.MythJobAutoConfigure;
import com.byit.dto.executor.DispatchResponseDto;
import com.byit.dto.executor.ScriptDto;
import com.byit.dto.plugin.RunLog;
import com.byit.enums.EmailEnum;
import com.byit.enums.JobResultEnum;
import com.byit.enums.NodePropertyEnum;
import com.byit.enums.NodeRunStatusPropertyEnum;
import com.byit.dto.executor.DispatchResponseDto;
import com.byit.dto.executor.ScriptDto;
import com.byit.enums.JobResultEnum;
import com.byit.enums.task.RunResultEnum;
import com.byit.enums.task.RunTypeEnum;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.JobTaskSchedule;
import com.byit.model.RunRecording;
import com.byit.rpc.util.RpcException;
import com.byit.service.RunScriptService;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.service.impl.RunRecordingServiceImpl;
......@@ -28,13 +29,15 @@ import java.util.Date;
import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
/**
* 脚本的执行调用器
* @author huangfu
*/
@Slf4j
public class ScriptExecutorJobTask implements TimerTask {
private final String HTTP_PRE = "http://";
private final String HTTP_SUFFIX = "/myth-job-admin/job/callbackRes";
private static final String HTTP_PRE = "http://";
private static final String HTTP_SUFFIX = "/myth-job-admin/job/callbackRes";
public static final String LINE = "\n";
private JobTaskSchedule mythJobTaskSchedule;
......@@ -42,8 +45,9 @@ public class ScriptExecutorJobTask implements TimerTask {
this.mythJobTaskSchedule = mythJobTaskSchedule;
}
@Override
public void run(Timeout timeout) throws Exception {
public void run(Timeout timeout) {
log.debug("---------开始交验工作流时否正在运行中------------");
if(checkFlowStatusIsKill(mythJobTaskSchedule.getFlowId(),mythJobTaskSchedule.getRunId())){
log.warn("--------------该工作流已经被杀死,执行快速失败!-------------------");
......@@ -56,15 +60,9 @@ public class ScriptExecutorJobTask implements TimerTask {
String priority = mythJobTaskSchedule.getPriority();
if(NodePropertyEnum.ADVANCED_NODE.getCode().equals(priority)){
log.debug("--------检测到高级节点--------");
MythJobAutoConfigure.ADVANCED_JOB_THREAD_POOL.execute(()->{
runJob(mythJobTaskSchedule);
});
MythJobAutoConfigure.ADVANCED_JOB_THREAD_POOL.execute(()-> runJob(mythJobTaskSchedule));
}else{
MythJobAutoConfigure.LOW_LEVEL_JOB_THREAD_POOL.execute(()->{
log.debug("--------检测到低级节点--------");
runJob(mythJobTaskSchedule);
});
MythJobAutoConfigure.LOW_LEVEL_JOB_THREAD_POOL.execute(()-> runJob(mythJobTaskSchedule));
}
}
......@@ -106,13 +104,13 @@ public class ScriptExecutorJobTask implements TimerTask {
DispatchResponseDto dispatchResponseDto = new DispatchResponseDto();
try{
dispatchResponseDto = runScriptService.runScript(scriptDto);
}catch (Exception rpcException){
}catch (Exception e){
StringRedisTemplate stringRedisTemplate = (StringRedisTemplate) SpringUtil.getBean("stringRedisTemplate");
RunLog runLog = RunLog.builder().isEnd(true).isSuccess(false).runLog("执行资源异常" + rpcException.getMessage()).build();
RunLog runLog = RunLog.builder().isEnd(true).isSuccess(false).runLog("执行资源异常" + e.getMessage()).build();
stringRedisTemplate.convertAndSend("REAL-EXEC-" + mythJobTaskSchedule.getLogId() , JSON.toJSONString(runLog, WriteClassName));
log.error("------执行机异常{}-----", rpcException.getMessage());
dispatchResponseDto.setMsg(rpcException.getMessage());
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_FAIL.getRes());
log.error("------执行机异常{}-----", e.getMessage());
dispatchResponseDto.setMsg(e.getMessage());
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_FAIL.getCode());
}
saveLog(mythJobTaskSchedule,dispatchResponseDto);
......@@ -129,73 +127,87 @@ public class ScriptExecutorJobTask implements TimerTask {
if (jobTaskRunLogById.getRunCount()>1) {
//第一次调度的日志
String triggerMsg = jobTaskRunLogById.getTriggerMsg();
jobTaskRunLog.setTriggerMsg(triggerMsg+"|"+dispatchResponseDto.getMsg());
jobTaskRunLog.setTriggerMsg(triggerMsg + dispatchResponseDto.getMsg() + "\n");
}else{
jobTaskRunLog.setTriggerMsg(dispatchResponseDto.getMsg());
jobTaskRunLog.setTriggerMsg(dispatchResponseDto.getMsg() + "\n");
}
jobTaskRunLog.setLogId(mythJobTaskSchedule.getLogId());
jobTaskRunLog.setVersionName(mythJobTaskSchedule.getVersionName());
jobTaskRunLog.setRunType("1");
jobTaskRunLog.setRunType(RunTypeEnum.EXECUTIVE_MACHINE_RUN.getCode());
jobTaskRunLog.setJobType(mythJobTaskSchedule.getJobType());
jobTaskRunLog.setHandlerName(mythJobTaskSchedule.getHandlerName());
jobTaskRunLog.setTriggerTime(new Date());
jobTaskRunLog.setTriggerCode(JobResultEnum.DISPATCH_SUCCESS.getCode().equals(dispatchResponseDto.getCode())?"1":"2");
//放置执行记录
if(!(JobResultEnum.DISPATCH_SUCCESS.getCode().equals(dispatchResponseDto.getCode()))){
String code = RunResultEnum.TRIGGER_SUCCESS.getCode();
//放置调度码 调度失败 那么执行肯定也失败
if(JobResultEnum.DISPATCH_FAIL.getCode().equals(dispatchResponseDto.getCode())){
code = RunResultEnum.TRIGGER_ERROR.getCode();
Date thisTime = new Date();
jobTaskRunLog.setStartTime(thisTime);
jobTaskRunLog.setEndTime(thisTime);
jobTaskRunLog.setRunCode(NodeRunStatusPropertyEnum.RUN_FAILURE.getCode());
jobTaskRunLog.setRunCode(RunResultEnum.RUN_ERROR.getCode());
jobTaskRunLog.setAlertEnd(EmailEnum.IS_ALARM_NO.getCode());
if (jobTaskRunLogById.getRunCount()>1) {
//上一次的执行日志
String runMsg = jobTaskRunLogById.getRunMsg();
jobTaskRunLog.setRunMsg(runMsg+"|"+dispatchResponseDto.getMsg());
jobTaskRunLog.setRunMsg(runMsg + dispatchResponseDto.getMsg()+LINE);
}else{
jobTaskRunLog.setRunMsg(dispatchResponseDto.getMsg()+LINE);
}
}
//失败
jobTaskRunLog.setTriggerCode(code);
jobTaskRunLog.setRunCount(jobTaskRunLogById.getRunCount()+1);
//获取执行机地址
jobTaskRunLog.setJobGroupIp(dispatchResponseDto.getUrl());
jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
log.info("-----------saveLog--保存脚本调度日志结束------------");
}
/**
* 保存失败的节点执行信息
* @param mythJobTaskSchedule 节点信息
*/
private void saveErrorLog(JobTaskSchedule mythJobTaskSchedule){
log.debug("-----------saveLog--保存脚本调度日志开始------------");
log.debug("-----------saveErrorLog--保存KILL脚本调度日志开始------------");
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
//获取值放入调度轮之前保存的日志
JobTaskRunLogWithBLOBs jobTaskRunLogById = jobTaskRunLogService.findJobTaskRunLogById(mythJobTaskSchedule.getLogId());
JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
//放置调度记录
//放置调度记录 如果执行次数大于一
if (jobTaskRunLogById.getRunCount()>1) {
//一次调度的日志
//一次调度的日志
String triggerMsg = jobTaskRunLogById.getTriggerMsg();
jobTaskRunLog.setTriggerMsg(triggerMsg+"| 工作流被kill");
jobTaskRunLog.setTriggerMsg(triggerMsg+RunResultEnum.KILL_SUCCESS.getMsg()+LINE);
}else{
jobTaskRunLog.setTriggerMsg("工作流被kill");
jobTaskRunLog.setTriggerMsg(RunResultEnum.KILL_SUCCESS.getMsg()+LINE);
}
jobTaskRunLog.setLogId(mythJobTaskSchedule.getLogId());
jobTaskRunLog.setVersionName(mythJobTaskSchedule.getVersionName());
jobTaskRunLog.setRunType("1");
jobTaskRunLog.setRunType(RunTypeEnum.EXECUTIVE_MACHINE_RUN.getCode());
jobTaskRunLog.setJobType(mythJobTaskSchedule.getJobType());
jobTaskRunLog.setHandlerName(mythJobTaskSchedule.getHandlerName());
jobTaskRunLog.setTriggerTime(new Date());
jobTaskRunLog.setTriggerCode("2");
//流被kill他的调度肯定失败
jobTaskRunLog.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
Date thisTime = new Date();
jobTaskRunLog.setStartTime(thisTime);
jobTaskRunLog.setEndTime(thisTime);
//将剩余失败重试次数刷新为0
jobTaskRunLog.setFailedRemainingCount(0);
jobTaskRunLog.setRunCode(NodeRunStatusPropertyEnum.RUN_FAILURE.getCode());
//将执行状态更改为被杀死
jobTaskRunLog.setRunCode(RunResultEnum.KILL_SUCCESS.getCode());
jobTaskRunLog.setAlertEnd(EmailEnum.IS_ALARM_NO.getCode());
if (jobTaskRunLogById.getRunCount()>1) {
//上一次的执行日志
String runMsg = jobTaskRunLogById.getRunMsg();
jobTaskRunLog.setRunMsg(runMsg+"|工作流被kill!");
jobTaskRunLog.setRunMsg(runMsg+RunResultEnum.KILL_SUCCESS.getMsg()+LINE);
}else{
jobTaskRunLog.setRunMsg(RunResultEnum.KILL_SUCCESS.getMsg()+ LINE);
}
jobTaskRunLog.setRunCount(jobTaskRunLogById.getRunCount()+1);
jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
log.info("-----------saveLog--保存脚本调度日志结束------------");
log.info("-----------saveErrorLog--保存KILL脚本调度日志结束------------");
}
}
package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.model.EmailAlarm;
import com.byit.model.vo.EmailAlarmVo;
import com.byit.service.EmailAlarmService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* @program: com.byit.thread.EmailScanHelper
* @description: 对于邮箱的扫描
* @author: huangfu
* @date: 2020年1月7日10:44:01
**/
@Component
@Slf4j
@Deprecated
public class EmailScanHelper {
private final DataSource dataSource;
private final EmailAlarmService emailAlarmService;
private volatile boolean emailThreadIsStop = false;
private Thread emailThread;
public EmailScanHelper(DataSource dataSource, EmailAlarmService emailAlarmService) {
this.dataSource = dataSource;
this.emailAlarmService = emailAlarmService;
}
public void start(){
startScanNotSentEmailFlow();
}
public void startScanNotSentEmailFlow(){
emailThread = new Thread(() ->{
dateAligned(5000);
log.info("--------------【com.byit.thread.EmailScanHelper.startScanNotSentEmailFlow】init success---------------");
while (!emailThreadIsStop){
//是否需要睡眠
boolean isSleep = false;
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
try {
/**
* 添加行锁
*/
conn = dataSource.getConnection();
connAutoCommit = conn.getAutoCommit();
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'emali_alarm_lock' FOR UPDATE ");
preparedStatement.execute();
//查询未告警的邮箱
List<EmailAlarm> emailAlarmByAlarmResults = emailAlarmService.findEmailAlarmByAlarmResult();
if(CollectionUtil.isNotEmpty(emailAlarmByAlarmResults)){
emailAlarmByAlarmResults.forEach(emailAlarm ->{
EmailAlarmVo emailAlarmVo = new EmailAlarmVo();
BeanUtils.copyProperties(emailAlarm,emailAlarmVo);
emailAlarmService.sendEmail(emailAlarmVo);
});
}else{
isSleep = true;
}
}catch (Exception e){
e.printStackTrace();
}finally {
//释放资源
if(conn != null){
try {
conn.commit();
} catch (SQLException e) {
if(!emailThreadIsStop){
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if(conn != null){
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if(!emailThreadIsStop){
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if(preparedStatement != null){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!emailThreadIsStop){
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
conn.close();
} catch (SQLException e) {
if(!emailThreadIsStop){
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
if(isSleep){
dateAligned(20000);
}
}
});
emailThread.setDaemon(true);
emailThread.setName("myth-job#【EmailScanHelper】#startScanNotSentEmailFlow");
emailThread.start();
}
public void doStop(){
this.emailThreadIsStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (emailThread.getState() != Thread.State.TERMINATED) {
emailThread.interrupt();
try {
emailThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【日志扫描线程被注销】-----------------------");
}
/**
* 对齐时钟。整秒运行
*/
private void dateAligned(long waitTime){
try {
TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis()%1000);
} catch (InterruptedException e) {
log.warn("----------------【线程被中断】-----------------------");
}
}
}
package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.job.utils.CronExpression;
import com.byit.model.Flow;
import com.byit.model.Node;
import com.byit.service.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.ParseException;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* 扫描任务流表的线程
* 目的:扫描即将要执行的任务流表 半个小时
* @author huangfu
*/
@Component
@Slf4j
@Deprecated
public class FlowScanHelper {
private final Long PRE_TEST_TIME = System.currentTimeMillis()+TimeUnit.HOURS.toMillis(30);
private final FlowService flowService;
private final DataSource dataSource;
private final NodeService nodeService;
private final RunNodeServer runNodeServer;
private Thread flowThread;
private volatile boolean flowThreadIsStop = false;
public FlowScanHelper(FlowService flowService, DataSource dataSource, NodeService nodeService, RunNodeServer runNodeServer) {
this.flowService = flowService;
this.dataSource = dataSource;
this.nodeService = nodeService;
this.runNodeServer = runNodeServer;
}
public void start(){
flowThreadStart();
}
/**
* 运行扫描工作流的线程
*/
private void flowThreadStart(){
flowThread = new Thread(() ->{
dateAligned(5000);
log.info("--------------【com.byit.thread.FlowScanHelper#flowThreadStart】init success---------------");
while (!flowThreadIsStop){
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
//是否需要睡眠
boolean isSleep = false;
try {
/**
* 添加行锁
*/
conn = dataSource.getConnection();
connAutoCommit = conn.getAutoCommit();
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'flow_lock' FOR UPDATE ");
preparedStatement.execute();
//这个查询时有一个条件是 剩余次数不等于0也就是说 等于0的就查询不出来
List<Flow> halfAnHourFlow = flowService.findHalfAnHourFlow(PRE_TEST_TIME);
if (CollectionUtil.isNotEmpty(halfAnHourFlow)) {
for(Flow flow : halfAnHourFlow ){
log.debug("-----------------【工作流{}的执行次数大于0,放行】-------------------------",flow.getFlowName());
//String versionName = flow.getVersionName()
Integer flowId = flow.getFlowId();
//根据工作流查询工作流下所有的节点
List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(flowId);
if(CollectionUtil.isNotEmpty(nodeByFlowIdAndVersionName)){
//保存到运行记录表和任务表
runNodeServer.saveRunRec(flow,nodeByFlowIdAndVersionName);
if (flow.getRemainingCount()>0) {
flow.setRemainingCount(flow.getRemainingCount()-1);
}
//获取cron表达式
String flowCron = flow.getFlowCron();
//设置下一周期的时间
Date nextValidTime = new CronExpression(flowCron).getNextValidTimeAfter(new Date(flow.getTriggerNextTime()));
flow.setTriggerNextTime(nextValidTime.getTime());
flowService.updateByIdSelective(flow);
}
}
}else{
isSleep = true;
}
}catch (Exception e){
e.printStackTrace();
}finally {
//释放资源
if(conn != null){
try {
conn.commit();
} catch (SQLException e) {
if(!flowThreadIsStop){
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if(conn != null){
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if(!flowThreadIsStop){
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if(preparedStatement != null){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!flowThreadIsStop){
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
conn.close();
} catch (SQLException e) {
if(!flowThreadIsStop){
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
if(isSleep){
try {
log.info("------------------【未扫描到要执行的工作流】-------------------");
TimeUnit.MINUTES.sleep(1);
} catch (InterruptedException e) {
log.warn("----------------------【扫描工作流的线程被关闭了】----------------------------");
}
}
}
});
flowThread.setDaemon(true);
flowThread.setName("myth-job#【FlowScanHelper】#flowThreadStart");
flowThread.start();
}
public void doStop(){
this.flowThreadIsStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (flowThread.getState() != Thread.State.TERMINATED) {
flowThread.interrupt();
try {
flowThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【工作扫描线程被注销】-----------------------");
}
/**
* 对齐时钟。整秒运行
*/
private void dateAligned(long waitTime){
try {
TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis()%1000);
} catch (InterruptedException e) {
log.warn("----------------【线程被中断】-----------------------");
}
}
}
package com.byit.thread;
import com.byit.dto.executor.JobRunResultDto;
import com.byit.dto.web.ReturnResult;
import com.byit.enums.RunRecordingEnum;
import com.byit.enums.JobResultEnum;
import com.byit.enums.task.RunResultEnum;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.packet.response.PluginRpcResponsePacket;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
......@@ -26,8 +25,14 @@ public class JavaTaskCallbackThread implements Runnable {
JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
Map<String,String> result = (Map<String,String>)pluginRpcResponsePacket.getResult();
JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
String code;
if(JobResultEnum.SUCCESS.getCode().equals(result.get("code"))){
code = RunResultEnum.RUN_SUCCESS.getCode();
}else {
code = RunResultEnum.RUN_ERROR.getCode();
}
jobTaskRunLog.setRunMsg(result.get("msg"));
jobTaskRunLog.setRunCode(result.get("code"));
jobTaskRunLog.setRunCode(code);
jobTaskRunLog.setLogId(logId);
mythJobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
......
package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.JobTriggerStatusEnums;
import com.byit.job.WorkRoulette;
import com.byit.model.*;
import com.byit.service.*;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.service.mapservice.RunRecordingAndJobTaskService;
import com.byit.service.mapservice.TaskAndLogServer;
import com.byit.task.JavaBeanJobTask;
import com.byit.task.ScriptExecutorJobTask;
import com.byit.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/**
* @program: byit-myth-job->JobScheduleHeloer
* @description: 工作时间排期表,这里开启了两条线程:
* 一条线程去任务节点读取七秒内将要执行的任务添加进任务排期表
* 一条线程去任务排期表中预读五秒将要执行的任务添加进任务调度轮盘
* @author: huangfu
* @date: 2019/12/9 11:16
**/
@Slf4j
@Component
@Deprecated
public class JobScheduleHelper{
private DataSource dataSource;
private final RunRecordingAndJobTaskService runRecordingAndJobTaskService;
private final JobTaskService jobTaskService;
private final JobTaskScheduleService jobTaskScheduleService;
private final NodeDependencyService nodeDependencyService;
private final JobTaskRunLogService jobTaskRunLogService;
private final TaskAndLogServer taskAndLogServer;
private final RunRecordingService runRecordingService;
/**
* 读取任务节点的预读
*/
private static final long PRE_READ_MS = 7000;
/**
* 读取任务排期表的预读
*/
private static final long SCHEDULE_READ_MS=5000;
/**
* 任务节点线程
*/
private Thread jobInfoThread;
/**
* 任务排期表线程
*/
private Thread scheduleThread;
/**
* 是否停止扫描任务节点表
*/
private volatile boolean jobInfoThreadToStop = false;
/**
* 是否停止排期表的扫描,停止向任务调度轮盘添加任务
*/
private volatile boolean scheduleThreadToStop = false;
@Autowired
public JobScheduleHelper(RunRecordingAndJobTaskService runRecordingAndJobTaskService, JobTaskScheduleService jobTaskScheduleService, JobTaskService jobTaskService, NodeDependencyService nodeDependencyService, JobTaskRunLogService jobTaskRunLogService, TaskAndLogServer taskAndLogServer, RunRecordingService runRecordingService) {
this.runRecordingAndJobTaskService = runRecordingAndJobTaskService;
this.jobTaskScheduleService = jobTaskScheduleService;
this.jobTaskService = jobTaskService;
this.nodeDependencyService = nodeDependencyService;
this.jobTaskRunLogService = jobTaskRunLogService;
this.taskAndLogServer = taskAndLogServer;
this.runRecordingService = runRecordingService;
}
/**
* 启动两条线程
*/
public void start(){
jobInfoThreadStart();
scheduleThreadStart();
}
/**
* 任务节点扫描
*/
public void jobInfoThreadStart(){
jobInfoThread = new Thread(()->{
try {
TimeUnit.MILLISECONDS.sleep(PRE_READ_MS - System.currentTimeMillis()%1000 );
} catch (InterruptedException e) {
if (!jobInfoThreadToStop) {
log.error(e.getMessage(), e);
}
}
log.info("---------------------【init myth-job admin jobInfoThread success】------------------------");
boolean preReadSuc;
while (!jobInfoThreadToStop) {
//开始去扫描任务节点
long start = System.currentTimeMillis();
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
preReadSuc = true;
try{
conn = dataSource.getConnection();
//记录当前的自动提交状态
connAutoCommit = conn.getAutoCommit();
//修改为不自动提交
conn.setAutoCommit(false);
//先添加扫描任务节点的行锁
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'job_task_lock' FOR UPDATE ");
preparedStatement.execute();
//行锁已经加上 后续处理
long nowTime = System.currentTimeMillis();
//开始寻找此时 不是暂停状态,而且七秒内即将运行的任务 而且还不是暂停的节点
List<JobTask> jobTasks = jobTaskService.findJobTaskByTriggerNextTimeLessThanEqual(nowTime + PRE_READ_MS);
if(CollectionUtil.isNotEmpty(jobTasks)){
List<JobTaskSchedule> jobTaskSchedules = new ArrayList<JobTaskSchedule>(15);
//遍历七秒内将要运行的节点数据
for(JobTask jobTask : jobTasks){
/**
* 判断节点状态
* 1.虚节点状态,虚节点状态是映射了一个工作流,需要将该节点映射的工作流下所由的几点拉取到任务表
* 2.普通节点也有两种状态:
* I.开始节点:开始节点不需要验证上级工作流,直接放行执行
* II.正常节点:正常节点需要验证上级节点,首先判断自己是否收弱引用,如果是弱引用那么需要判断
* 上级节点是否已经全部都执行完了,执行完后不论成功与否都执行,同时工作流的运行结果
* 只与end节点关联
*/
//虚节点的状态
if(FlowPropertyEnum.IS_INNER.getCode().equals(jobTask.getIsVirtual())){
try {
runRecordingAndJobTaskService.saveRunRecordingAndTask(jobTask);
log.info("----------------【虚节点保存成功,删除虚节点】--------------------");
jobTaskService.removeMythJobTaskById(jobTask.getId());
} catch (Exception e) {
e.printStackTrace();
}
}else{
if("start".equals(jobTask.getNodeName()) ){
log.debug("任务:{}", jobTask);
JobTaskSchedule jobTaskSchedule = new JobTaskSchedule();
BeanUtils.copyProperties(jobTask,jobTaskSchedule);
jobTaskSchedules.add(jobTaskSchedule);
//更改运行记录为运行中
String runId = jobTask.getRunId();
Integer flowId = jobTask.getFlowId();
RunRecording runRecordingByFlowIdAndRunId = runRecordingService.findRunRecordingByFlowIdAndRunId(flowId, runId);
runRecordingByFlowIdAndRunId.setFlowStatus(FlowPropertyEnum.FLOW_RUN_ING.getCode());
runRecordingService.updateRunRecordingById(runRecordingByFlowIdAndRunId);
}else{
//查询该节点的依赖节点
List<Integer> dependIdByNodeId = nodeDependencyService.findDependIdByNodeId(jobTask.getNodeId());
//这里返回的是上级节点的日志执行情况 把运行中的数据给过滤掉了
List<JobTaskRunLog> jobTaskRunLogList = jobTaskRunLogService.findJobTaskRunLogNotEndNodeByRunCodeCount(dependIdByNodeId, jobTask.getRunId());
if (CollectionUtil.isNotEmpty(jobTaskRunLogList)) {
//判断父类节点是否已经全部完成,只需要判断依赖节点的数目和查询出来的日志数据是否相同
if(dependIdByNodeId.size() == jobTaskRunLogList.size()){
//过滤失败的节点
List<JobTaskRunLog> errorJobLog = jobTaskRunLogList.stream().
filter(jobTaskRunLog -> ("2".equals(jobTaskRunLog.getRunCode()) || "4".equals(jobTaskRunLog.getRunCode()) || "6".equals(jobTaskRunLog.getRunCode())))
.collect(Collectors.toList());
//判断剩余执行次数是否为0
if (parentNodeErrorCount(errorJobLog)) {
log.debug("--------------【{}的上级节点的失败节点已经全部重试完毕】------------------",jobTask);
//该节点如果为弱引用
if ("1".equals(jobTask.getSuperSuccessRun())) {
log.info("-------------【查询到有弱引用节点】-----------------");
//执行代码
runJobTask(jobTask,jobTaskSchedules);
}else{
//如果有失败的节点 就把该节点置为失败
if(CollectionUtil.isNotEmpty(errorJobLog)){
//删除这个数据 并且添加到日志
taskAndLogServer.addRunLogAndRemoveTask(jobTask);
}else{
//执行代码
runJobTask(jobTask,jobTaskSchedules);
}
}
}
}
}
}
}
}
if(CollectionUtil.isNotEmpty(jobTaskSchedules)) {
jobTaskScheduleService.saveAllData(jobTaskSchedules);
jobTaskService.removeMythJobTaskInIds(jobTaskSchedules);
}
}else{
//log.info("-------------------空轮转-------------------");
preReadSuc = false;
}
}catch (Exception e){
if(!jobInfoThreadToStop){
e.printStackTrace();
log.error("------------------扫描任务表出现异常:{}-----------------",e.getMessage());
}
}finally {
//commit
if (conn!=null) {
try {
conn.commit();
}catch (SQLException e){
if(!jobInfoThreadToStop){
log.error("----------------提交行锁错误:{}-------------------",e.getMessage());
}
}
//设置提交状态恢复原来的值
try {
conn.setAutoCommit(connAutoCommit);
}catch (SQLException e){
if(!jobInfoThreadToStop){
log.error("------------------设置为自动提交出错:{}------------------",e.getMessage());
}
}
//关闭连接
try {
conn.close();
}catch (SQLException e){
if(!jobInfoThreadToStop){
log.error("----------------关闭数据库连接:{}-------------------",e.getMessage());
}
}
}
//关闭执行器
if(null !=preparedStatement){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!jobInfoThreadToStop){
log.error("---------------关闭执行器出错:{}----------------------",e.getMessage());
}
}
}
}
//计算总的处理时间 将时间保持在一秒处理一次
long cost = System.currentTimeMillis()-start;
//不足一秒的等待 等够1秒为止
if(cost < 1000){
// 预读期:成功-每秒扫描一次;失败-跳过这段时间
try {
//这个睡眠是空轮转时,延长睡眠时间,奖励CUP的使用频率
TimeUnit.MILLISECONDS.sleep((preReadSuc?1000:PRE_READ_MS) - System.currentTimeMillis()%1000);
} catch (InterruptedException e) {
if (!jobInfoThreadToStop) {
log.error("----------------------{}-------------------------",e.getMessage());
}
}
}
}
});
//设置为守护线程
jobInfoThread.setDaemon(true);
//设置名字
jobInfoThread.setName("myth-job,admin JobScheduleHelper#jobInfoThread");
jobInfoThread.start();
}
/**
* 开始操作任务排期表:
* 1. 扫描五秒要执行的数据
* 2.加载时,创建任务日志,传入调度时间
* 3.加载到任务调度轮盘
* 4.删除数据
*/
private void scheduleThreadStart(){
scheduleThread = new Thread(() ->{
try {
TimeUnit.MILLISECONDS.sleep(4000 - System.currentTimeMillis()%1000 );
} catch (InterruptedException e) {
if (!scheduleThreadToStop) {
log.error(e.getMessage(), e);
}
}
log.info("---------------------init myth-job admin scheduleThread success------------------------");
/**
* 判断是否是空轮转
* 空轮转的话是需要休眠的
*/
boolean preReadSuc;
while (!scheduleThreadToStop){
//开始去扫描任务节点
long start = System.currentTimeMillis();
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
preReadSuc = true;
try{
conn = dataSource.getConnection();
//记录当前的自动提交状态
connAutoCommit = conn.getAutoCommit();
//修改为不自动提交
conn.setAutoCommit(false);
//先添加扫描任务节点的行锁
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'job_task_schedule_lock' FOR UPDATE ");
preparedStatement.execute();
//行锁已经加上 后续处理
long nowTime = System.currentTimeMillis();
//查询所有符合条件的任务节点
List<JobTaskSchedule> jobTaskSchedules = jobTaskScheduleService.findJobTaskScheduleByTriggerNextTimeLessThanEqual(nowTime + SCHEDULE_READ_MS);
if(CollectionUtil.isNotEmpty(jobTaskSchedules)){
//循环遍历添加任务
jobTaskSchedules.forEach(mythJobTaskSchedule ->{
//如果是重跑就有logId
Integer logId = mythJobTaskSchedule.getLogId();
if(logId == null){
logId = saveLog(mythJobTaskSchedule);
}
mythJobTaskSchedule.setLogId(logId);
Long triggerTime = mythJobTaskSchedule.getTriggerTime();
if ("JAVA".equals(mythJobTaskSchedule.getJobType())) {
//构建调度执行器
JavaBeanJobTask javaBeanJobTask = new JavaBeanJobTask(mythJobTaskSchedule);
WorkRoulette.addJob(javaBeanJobTask,triggerTime);
}else if("SCRIPT".equals(mythJobTaskSchedule.getJobType())){
ScriptExecutorJobTask scriptExecutorJobTask = new ScriptExecutorJobTask(mythJobTaskSchedule);
WorkRoulette.addJob(scriptExecutorJobTask,triggerTime);
}
jobTaskScheduleService.delete(mythJobTaskSchedule.getId());
});
}else{
preReadSuc = false;
}
}catch (Exception e){
if(!scheduleThreadToStop){
log.error("------------------扫描排期表出现异常:{}-----------------",e.getMessage());
}
}finally {
//commit
if (conn!=null) {
try {
conn.commit();
}catch (SQLException e){
if(!scheduleThreadToStop){
log.error("----------------提交行锁错误:{}-------------------",e.getMessage());
}
}
//设置提交状态恢复原来的值
try {
conn.setAutoCommit(connAutoCommit);
}catch (SQLException e){
if(!scheduleThreadToStop){
log.error("------------------设置为自动提交出错:{}------------------",e.getMessage());
}
}
//关闭连接
try {
conn.close();
}catch (SQLException e){
if(!scheduleThreadToStop){
log.error("----------------关闭数据库连接:{}-------------------",e.getMessage());
}
}
}
//关闭执行器
if(null !=preparedStatement){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!scheduleThreadToStop){
log.error("---------------关闭执行器出错:{}----------------------",e.getMessage());
}
}
}
//计算总的处理时间 将时间保持在一秒处理一次
long cost = System.currentTimeMillis()-start;
//不足一秒的等待 等够1秒为止
if(cost < 1000){
// 预读期:成功-每秒扫描一次;失败-跳过这段时间
try {
//这个睡眠是空轮转时,延长睡眠时间,奖励CUP的使用频率
TimeUnit.MILLISECONDS.sleep((preReadSuc?1000:SCHEDULE_READ_MS) - System.currentTimeMillis()%1000);
} catch (InterruptedException e) {
if (!scheduleThreadToStop) {
log.error("----------------------{}-------------------------",e.getMessage());
}
}
}
}
}
});
scheduleThread.setName("myth-job,admin JobScheduleHelper#scheduleThread");
scheduleThread.setDaemon(true);
scheduleThread.start();
}
public void doStop(){
this.jobInfoThreadToStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (jobInfoThread.getState() != Thread.State.TERMINATED) {
jobInfoThread.interrupt();
try {
jobInfoThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【任务节点节点扫描线程被注销】-----------------------");
this.scheduleThreadToStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (scheduleThread.getState() != Thread.State.TERMINATED) {
scheduleThread.interrupt();
try {
scheduleThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【排期表扫描线程被注销】-----------------------");
}
@Autowired
public void setDataSource(DataSource dataSource) {
this.dataSource = dataSource;
}
private Integer saveLog(JobTaskSchedule mythJobTaskSchedule){
JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
jobTaskRunLog.setRunId(mythJobTaskSchedule.getRunId());
jobTaskRunLog.setIsVirtual(mythJobTaskSchedule.getIsVirtual());
jobTaskRunLog.setFlowId(mythJobTaskSchedule.getFlowId());
jobTaskRunLog.setFlowName(mythJobTaskSchedule.getFlowName());
jobTaskRunLog.setNodeId(mythJobTaskSchedule.getNodeId());
jobTaskRunLog.setNodeName(mythJobTaskSchedule.getNodeName());
jobTaskRunLog.setRunParams(mythJobTaskSchedule.getRunParam());
jobTaskRunLog.setFailedRemainingCount(mythJobTaskSchedule.getFailedRetryCount());
jobTaskRunLog.setJobType(mythJobTaskSchedule.getJobType());
JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
mythJobTaskRunLogService.saveJobTaskRunLog(jobTaskRunLog);
return jobTaskRunLog.getLogId();
}
/**
* 判断失败节点的重试次数是不是为0
* @param errorJobLog 上级节点的全部失败节点
* @return
*/
private boolean parentNodeErrorCount(List<JobTaskRunLog> errorJobLog){
if(CollectionUtil.isEmpty(errorJobLog)){
return true;
}
for (JobTaskRunLog jobTaskRunLog : errorJobLog) {
//失败重试次数大于0 而且错误原因不是上级节点执行失败
if(jobTaskRunLog.getFailedRemainingCount()>0 && !("6".equals(jobTaskRunLog.getRunCode()))){
log.debug("----------------【{}节点没有重试完毕】----------------",jobTaskRunLog);
return false;
}
}
return true;
}
private void runJobTask(JobTask jobTask,List<JobTaskSchedule> jobTaskSchedules){
//到这里 父类节点一定是全部都执行成功了!
JobTaskSchedule jobTaskSchedule = new JobTaskSchedule();
BeanUtils.copyProperties(jobTask,jobTaskSchedule);
jobTaskSchedules.add(jobTaskSchedule);
}
}
package com.byit.thread;
import com.byit.dto.executor.JobRunResultDto;
import com.byit.enums.EmailEnum;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.util.SpringUtil;
......@@ -32,9 +33,6 @@ public class LogCallbackThread implements Runnable {
if (jobTaskRunLogById.getRunCount()>1) {
String runMsg = jobTaskRunLogById.getRunMsg()+"|"+jobRunResultDto.getReturnResult().getMsg();
//这里需要追加文件 TODO
//这里需要追加文件 TODO
jobTaskRunLog.setRunMsg(runMsg);
}else{
jobTaskRunLog.setRunMsg(jobRunResultDto.getReturnResult().getMsg());
......@@ -45,7 +43,7 @@ public class LogCallbackThread implements Runnable {
jobTaskRunLog.setEndTime(jobRunResultDto.getEndTime());
jobTaskRunLog.setRunCode(jobRunResultDto.getReturnResult().getCode());
jobTaskRunLog.setAlertEnd("0");
jobTaskRunLog.setAlertEnd(EmailEnum.IS_ALARM_NO.getCode());
mythJobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
}
......
package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.enums.RunRecordingEnum;
import com.byit.model.JobTaskRunLog;
import com.byit.model.RunRecording;
import com.byit.service.mapservice.JobTaskRunLogAndJobTaskService;
import com.byit.service.JobTaskRunLogService;
import com.byit.service.RunRecordingService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* @program: byit-myth-job->LogScanHelper
* @description: 对日志表的扫描
* @author: huangfu
* @date: 2019/12/26 10:34
**/
@Component
@Slf4j
@Deprecated
public class LogScanHelper {
private DataSource dataSource;
private final JobTaskRunLogService jobTaskRunLogService;
private final RunRecordingService runRecordingService;
private final JobTaskRunLogAndJobTaskService jobTaskRunLogAndJobTaskService;
@Autowired
public LogScanHelper(JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService, JobTaskRunLogAndJobTaskService jobTaskRunLogAndJobTaskService) {
this.jobTaskRunLogService = jobTaskRunLogService;
this.runRecordingService = runRecordingService;
this.jobTaskRunLogAndJobTaskService = jobTaskRunLogAndJobTaskService;
}
@Autowired
public void setDataSource(DataSource dataSource) {
this.dataSource = dataSource;
}
/**
* 扫描虚节点的线程是否停止
*/
private volatile boolean virtualNodeScanIsStop = false;
/**
* 扫描未完成告警的节点的线程是否停止
*/
private volatile boolean notAlarmedNodeScanIsStop = false;
/**
* 扫描失败的节点的线程是否停止
*/
private volatile boolean errorNodeScanIsStop = false;
/**
* 扫描失败的节点的线程是否停止
*/
private Thread errorNodeScanThread = null;
/**
* 扫描虚节点的线程定义
*/
private Thread virtualNodeScanThread = null;
/**
* 扫描未完成告警的节点的线程
*/
private Thread notAlarmedNodeScanThread = null;
public void start(){
virtualNodeScanMethod();
//notAlarmedNodeScanMethod();
errorNodeScan();
}
/**
* 虚节点扫描
*/
private void virtualNodeScanMethod(){
//扫描虚节点线程
virtualNodeScanThread = new Thread(() ->{
dateAligned(5000);
log.info("---------------------【com.byit.thread.LogScanHelper#virtualNodeScanMethod】init success------------------------");
while (!virtualNodeScanIsStop){
//是否需要睡眠
boolean isSleep = false;
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
try{
conn = dataSource.getConnection();
connAutoCommit = conn.getAutoCommit();
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'log_virtual_node_lock' FOR UPDATE ");
preparedStatement.execute();
//查询日志表没有完结的虚拟节点
List<JobTaskRunLog> notEndVirtualNodes = jobTaskRunLogService.findNotEndVirtualNode( );
if(CollectionUtil.isNotEmpty(notEndVirtualNodes)){
notEndVirtualNodes.forEach(notEndVirtualNode ->{
log.debug("------------【开始查询虚拟节点的执行情况】----------------");
//根据运行标识和工作流id查询运行日志
RunRecording runRecordingByFlowIdAndRunId = runRecordingService.findRunRecordingByFlowIdAndRunId(notEndVirtualNode.getMapFlowId( ), notEndVirtualNode.getRunId());
//判断当前的工作流是否已经完结
if(runRecordingByFlowIdAndRunId!=null && "4".equals(runRecordingByFlowIdAndRunId.getFlowStatus())){
log.debug("------------【查询到有已经完成的虚拟节点修改日志】----------------");
notEndVirtualNode.setRunCode(runRecordingByFlowIdAndRunId.getFlowRunResult());
notEndVirtualNode.setEndTime(new Date());
//修改日志信息 改为成功或者失败
jobTaskRunLogService.updateJobTaskRunLog(notEndVirtualNode);
}else{
dateAligned(1000);
}
});
}else{
isSleep = true;
}
}catch (Exception e){
e.printStackTrace();
}finally {
//释放资源
if(conn != null){
try {
conn.commit();
} catch (SQLException e) {
if(!virtualNodeScanIsStop){
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if(conn != null){
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if(!virtualNodeScanIsStop){
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if(preparedStatement != null){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!notAlarmedNodeScanIsStop){
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
conn.close();
} catch (SQLException e) {
if(!virtualNodeScanIsStop){
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
if(isSleep){
dateAligned(20000);
}
}
});
virtualNodeScanThread.setName("myth-job#【LogScanHelper】#virtualNodeScanThread");
virtualNodeScanThread.setDaemon(true);
virtualNodeScanThread.start();
}
/**
* @deprecated 废除原因:不需要这条线程去扫描结束节点,只需要扫描对应工作流的节点数目是否全部匹配即可
* 这一操作需要放置到扫描运行实例的线程里面
* @deprecatedDate 2020年3月6日10:36:19
* 未告警的节点扫描
*/
private void notAlarmedNodeScanMethod(){
notAlarmedNodeScanThread = new Thread(() ->{
dateAligned(5000);
log.info("---------------------【com.byit.thread.LogScanHelper#notAlarmedNodeScanMethod】init success------------------------");
while (!notAlarmedNodeScanIsStop){
boolean isSleep = false;
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
try {
conn = dataSource.getConnection( );
connAutoCommit = conn.getAutoCommit( );
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'log_node_callback_lock' FOR UPDATE ");
preparedStatement.execute();
//查询的是失败的或者是已经结束的节点(完成的)
List<JobTaskRunLog> jobTaskRunLogEndOrFailureNode = jobTaskRunLogService.findJobTaskRunLogEndOrFailureNode( );
if(CollectionUtil.isNotEmpty(jobTaskRunLogEndOrFailureNode)){
log.debug("-------------【查询到有完结而且未告警的节点】-------------");
//遍历结束的节点 修改执行记录表
jobTaskRunLogEndOrFailureNode.forEach(endNode ->{
String runId = endNode.getRunId( );
Integer flowId = endNode.getFlowId( );
//如果运行结果为null 那么就是调度都没成功 那么就取调度的值
String code = endNode.getRunCode();
runRecordingService.updateRunRecordingByFlowIdAndRunId(RunRecording.builder().runId(runId).flowId(flowId).flowRunResult(code).flowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode()).build());
endNode.setAlertEnd("1");
jobTaskRunLogService.updateJobTaskRunLog(endNode);
});
}else{
isSleep = true;
}
}catch (Exception e){
if(!notAlarmedNodeScanIsStop){
e.printStackTrace();
}
}finally {
if(conn != null){
try {
conn.commit();
} catch (SQLException e) {
if(!notAlarmedNodeScanIsStop){
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if(conn != null){
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if(!notAlarmedNodeScanIsStop){
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if(preparedStatement != null){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!notAlarmedNodeScanIsStop){
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
conn.close();
} catch (SQLException e) {
if(!notAlarmedNodeScanIsStop){
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
if(isSleep){
dateAligned(20000);
}
}
});
notAlarmedNodeScanThread.setName("myth-job#【LogScanHelper】#notAlarmedNodeScanThread");
notAlarmedNodeScanThread.setDaemon(true);
notAlarmedNodeScanThread.start();
}
/**
* 失败的节点扫描 执行失败重试
* 扫描到失败节点,过滤引上级节点失败的情况
*/
public void errorNodeScan(){
errorNodeScanThread = new Thread(()->{
dateAligned(5000);
log.info("---------------------【com.byit.thread.LogScanHelper#errorNodeScan】init success------------------------");
while (!errorNodeScanIsStop) {
//是否需要睡眠
boolean isSleep = false;
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
try {
conn = dataSource.getConnection();
connAutoCommit = conn.getAutoCommit();
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'log_error_lock' FOR UPDATE ");
preparedStatement.execute();
//查询又重试次数的失败节点
List<JobTaskRunLog> errorNodes = jobTaskRunLogService.findErrorNode();
if(CollectionUtil.isNotEmpty(errorNodes)){
errorNodes.forEach(errorNode ->{
log.debug("-----------------【操作失败节点{}】-----------------",errorNode);
jobTaskRunLogAndJobTaskService.updateLogAndSaveJobTask(errorNode);
});
}else{
isSleep = true;
}
} catch (Exception e) {
e.printStackTrace();
} finally {
//释放资源
if (conn != null) {
try {
conn.commit();
} catch (SQLException e) {
if (!virtualNodeScanIsStop) {
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if (conn != null) {
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if (!virtualNodeScanIsStop) {
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if (preparedStatement != null) {
try {
preparedStatement.close();
} catch (SQLException e) {
if (!notAlarmedNodeScanIsStop) {
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
conn.close();
} catch (SQLException e) {
if (!virtualNodeScanIsStop) {
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
if (isSleep) {
dateAligned(20000);
}
}
});
errorNodeScanThread.setName("myth-job#【LogScanHelper】#errorNodeScan");
errorNodeScanThread.setDaemon(true);
errorNodeScanThread.start();
}
public void doStop(){
this.virtualNodeScanIsStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (virtualNodeScanThread.getState() != Thread.State.TERMINATED) {
virtualNodeScanThread.interrupt();
try {
virtualNodeScanThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【虚节点扫描线程被注销】-----------------------");
this.notAlarmedNodeScanIsStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (notAlarmedNodeScanThread.getState() != Thread.State.TERMINATED) {
notAlarmedNodeScanThread.interrupt();
try {
notAlarmedNodeScanThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【日志扫描线程被注销】-----------------------");
this.errorNodeScanIsStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (errorNodeScanThread.getState() != Thread.State.TERMINATED) {
errorNodeScanThread.interrupt();
try {
errorNodeScanThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【失败重试节点扫描线程被注销】-----------------------");
}
/**
* 对齐时钟。整秒运行
*/
private void dateAligned(long waitTime){
try {
TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis()%1000);
} catch (InterruptedException e) {
log.warn("----------------【线程被中断】-----------------------");
}
}
}
package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import com.byit.enums.RunRecordingEnum;
import com.byit.filesystem.FileSystem;
import com.byit.enums.JobResultEnum;
import com.byit.job.exceptions.BusinessException;
import com.byit.model.EmailAlarm;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.RunRecording;
import com.byit.service.EmailAlarmService;
import com.byit.service.JobTaskRunLogService;
import com.byit.service.RunRecordingService;
import com.byit.util.TimeFormatUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.csource.common.MyException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* 运行记录扫描线程
*
* @author huangfu
*/
@Deprecated
@Component
@Slf4j
public class RunRecordingScanHelper {
public static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
/**
* 完成时告警
*/
public static final String WHEN_DONE = "1";
/**
* 失败时告警
*/
public static final String FAILURE_DONE = "2";
/**
* 成功时告警
*/
public static final String SUCCESS_DONE = "3";
/**
* 文件服务器ip
*/
@Value("${file.system.ip}")
private String fileSystemIp;
/**
* 文件服务器端口号
*/
@Value("${file.system.port}")
private String fileSystemPort;
private static final String FILE_SYSTEM_PRE="http://";
private DataSource dataSource;
private final EmailAlarmService emailAlarmService;
private final JobTaskRunLogService jobTaskRunLogService;
private final RunRecordingService runRecordingService;
private final FileSystem fileSystem;
/**
* 查询完结且没有告警的节点线程是否停止
*/
private volatile boolean runRecordingThreadStop = false;
/**
* 查询完结且没有告警的节点线程
*/
private Thread runRecordingThread;
/**
* 判断工作流是否完结是否停止
*/
private volatile boolean judgeFlowIsEndStop = false;
/**
* 判断工作流是否完结的线程
*/
private Thread judgeFlowIsEndThread;
@Autowired
public RunRecordingScanHelper(EmailAlarmService emailAlarmService, JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService, FileSystem fileSystem) {
this.emailAlarmService = emailAlarmService;
this.jobTaskRunLogService = jobTaskRunLogService;
this.runRecordingService = runRecordingService;
this.fileSystem = fileSystem;
}
public void start() {
scanRunRecThread();
judgeFlowIsEndThread();
}
/**
* 判断工作流是否完结
*/
public void judgeFlowIsEndThread(){
judgeFlowIsEndThread = new Thread(() ->{
dateAligned(5000);
log.info("----------------------【com.byit.thread.RunRecordingScanThread#judgeFlowIsEnd】start-------------------");
while (!judgeFlowIsEndStop){
boolean isSleep = false;
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
try {
conn = dataSource.getConnection();
connAutoCommit = conn.getAutoCommit();
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'judge_flow_end_lock' FOR UPDATE ");
preparedStatement.execute();
//进行操作
judgeFlowIsEnd();
isSleep = true;
}catch (Exception e){
if (!judgeFlowIsEndStop) {
e.printStackTrace();
}
}finally {
if (conn != null) {
try {
conn.commit();
} catch (SQLException e) {
if (!judgeFlowIsEndStop) {
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if (conn != null) {
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if (!judgeFlowIsEndStop) {
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if (preparedStatement != null) {
try {
preparedStatement.close();
} catch (SQLException e) {
if (!judgeFlowIsEndStop) {
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
if(conn!=null){
conn.close();
}
} catch (SQLException e) {
if (!judgeFlowIsEndStop) {
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
if(isSleep){
dateAligned(20000);
}
}
});
judgeFlowIsEndThread.setDaemon(true);
judgeFlowIsEndThread.setName("myth-job#【judgeFlowIsEndThread】# judgeFlowIsEndThread");
judgeFlowIsEndThread.start();
}
/**
* 扫描未完成的工作流
*/
private void judgeFlowIsEnd(){
log.debug("---------------开始扫描运行中的运行实例----------------");
//查询未完结的工作流实例
List<RunRecording> runningRunRecordings = runRecordingService.findRunningRunRecording();
log.debug("---------扫描到运行实例{}个----------",runningRunRecordings.size());
runningRunRecordings.forEach(runRecording -> {
String runId = runRecording.getRunId();
Integer flowId = runRecording.getFlowId();
List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBsByFlowIdAndRunId = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
//判断节点数量与设定数量是否一致
if(jobTaskRunLogWithBLOBsByFlowIdAndRunId==null || jobTaskRunLogWithBLOBsByFlowIdAndRunId.size() != runRecording.getFlowNodeCount()){
log.debug("------------发现工作流{}不符合条件----------------",jobTaskRunLogWithBLOBsByFlowIdAndRunId);
return;
}
//判断是否全部完结 如果全部完结则判断工作流是否成功
if(logIsAllEnd(jobTaskRunLogWithBLOBsByFlowIdAndRunId)){
String runCode = allNodeIsSuccess(jobTaskRunLogWithBLOBsByFlowIdAndRunId);
//设置运行结果
runRecording.setFlowRunResult(runCode);
log.debug("----------------扫描到有完结的运行实例{},{}-------------",runCode,RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
runRecording.setFlowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
runRecordingService.updateRunRecordingById(runRecording);
}
});
}
private String allNodeIsSuccess(List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBs){
for (JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOB : jobTaskRunLogWithBLOBs) {
if (StringUtils.isNotEmpty(jobTaskRunLogWithBLOB.getRunCode())) {
if(!"1".equals(jobTaskRunLogWithBLOB.getRunCode()) || !"3".equals(jobTaskRunLogWithBLOB.getRunCode())){
return jobTaskRunLogWithBLOB.getRunCode();
}
}else{
throw new BusinessException(JobResultEnum.RUN_MSG_FAIL);
}
}
return "1";
}
/**
* 判断节点是否全部完结
* @param jobTaskRunLogWithBLOBs
* @return
*/
private boolean logIsAllEnd(List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBs){
for (JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOB : jobTaskRunLogWithBLOBs) {
if ("0".equals(jobTaskRunLogWithBLOB.getRunCount())) {
//存在运行中的直接返回false
return false;
}
}
return true;
}
/**
* 查询完结且没有告警的节点线程构建
*/
private void scanRunRecThread(){
runRecordingThread = new Thread(() -> {
dateAligned(5000);
log.info("--------------------【com.byit.thread.RunRecordingScanThread#scanRunRecThread】init success----------------------");
while (!runRecordingThreadStop) {
boolean isSleep = false;
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
try {
conn = dataSource.getConnection();
connAutoCommit = conn.getAutoCommit();
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'run_recording_lock' FOR UPDATE ");
preparedStatement.execute();
//进行操作
scanRunRec();
isSleep = true;
} catch (Exception e) {
if (!runRecordingThreadStop) {
e.printStackTrace();
}
} finally {
if (conn != null) {
try {
conn.commit();
} catch (SQLException e) {
if (!runRecordingThreadStop) {
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if (conn != null) {
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if (!runRecordingThreadStop) {
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if (preparedStatement != null) {
try {
preparedStatement.close();
} catch (SQLException e) {
if (!runRecordingThreadStop) {
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
if(conn!=null){
conn.close();
}
} catch (SQLException e) {
if (!runRecordingThreadStop) {
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
if(isSleep){
dateAligned(20000);
}
}
});
runRecordingThread.setDaemon(true);
runRecordingThread.setName("myth-job#【RunRecordingScanThread】# start");
runRecordingThread.start();
}
/**
* 查询完结 但是没有告警的节点
*/
private void scanRunRec() {
//查询完结且未告警的工作流信息
List<RunRecording> runRecordingByEndAndNotIsAlarm = runRecordingService.findRunRecordingByEndAndNotIsAlarm();
if (CollectionUtil.isNotEmpty(runRecordingByEndAndNotIsAlarm)) {
runRecordingByEndAndNotIsAlarm.forEach(runRecording -> {
//如果设置为完成时告警
switch (runRecording.getAlarmlAction()) {
//设置为完成时告警
case WHEN_DONE:
if (RunRecordingEnum.FLOW_STATUS_IS_END.getCode().equals(runRecording.getFlowStatus())) {
saveEmailAlarms(runRecording);
}
break;
//失败时告警
case FAILURE_DONE:
if (RunRecordingEnum.RUN_FLOW_FAILURE.getCode().equals(runRecording.getFlowRunResult())
|| RunRecordingEnum.RUN_FLOW_RE_FAILURE.getCode().equals(runRecording.getFlowRunResult())) {
saveEmailAlarms(runRecording);
}
break;
//成功时告警
case SUCCESS_DONE:
if (RunRecordingEnum.RUN_FLOW_SUCCESS.getCode().equals(runRecording.getFlowRunResult())
|| RunRecordingEnum.RUN_FLOW_RE_SUCCESS.getCode().equals(runRecording.getFlowRunResult())) {
saveEmailAlarms(runRecording);
}
break;
default:
break;
}
});
}
}
public void doStop() {
this.runRecordingThreadStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
if (runRecordingThread.getState() != Thread.State.TERMINATED) {
runRecordingThread.interrupt();
try {
runRecordingThread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.warn("---------------【运行记录扫描日志线程被注销】-----------------------");
this.judgeFlowIsEndStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
if (judgeFlowIsEndThread.getState() != Thread.State.TERMINATED) {
judgeFlowIsEndThread.interrupt();
try {
judgeFlowIsEndThread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.warn("---------------【扫描工作流结果线程被注销】-----------------------");
}
/**
* 对齐时钟。整秒运行
*/
private void dateAligned(long waitTime) {
try {
TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis() % 1000);
} catch (InterruptedException e) {
log.warn("----------------【线程被中断】-----------------------");
}
}
private void saveEmailAlarms(RunRecording runRecording) {
//这一步是根据flowId和RunId查询对应的节点信息
List<JobTaskRunLogWithBLOBs> jobTaskRunLogByFlowIdAndRunId = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(runRecording.getFlowId(), runRecording.getRunId());
String flowName = runRecording.getFlowName();
String senContentHtml = runMsgHtml(jobTaskRunLogByFlowIdAndRunId, flowName);
EmailAlarm emailAlarm = EmailAlarm.builder()
.flowId(runRecording.getFlowId())
.flowName(flowName)
.runId(runRecording.getRunId())
.versionName(runRecording.getFlowVersionName())
.alarmEmail(runRecording.getAlarmEmail())
.alarmContent(senContentHtml)
.alarmResult("0")
.flowRes(runRecording.getFlowRunResult())
.alarmTitle(flowName)
.build();
//保存邮箱
emailAlarmService.saveEmailAlarm(emailAlarm);
//修改为已告警
runRecording.setIsAlarm("0");
runRecordingService.updateRunRecordingById(runRecording);
}
private String runMsgHtml(List<JobTaskRunLogWithBLOBs> jobTaskRunLogs, String title) {
StringBuilder stringBuilder = new StringBuilder();
stringBuilder.append("<table border='1' width='80%' align='center' cellspacing='0' cellpadding='6'>")
.append(String.format("<h2 style='text-align:center;color:red'>%s</h2>", title))
.append("<thead align='center' style='background: blue;color: #fff'>")
.append("<th width = '10%'>节点名称</th>")
.append("<th width = '10%'>开始时间</th>")
.append("<th width = '10%'>结束时间</th>")
.append("<th width = '10%'>耗费时间</th>")
.append("<th width = '10%'>运行结果</th>")
.append("<th width = '50%'>运行日志</th>")
.append("</thead>")
.append("<tbody>");
if (CollectionUtil.isNotEmpty(jobTaskRunLogs)) {
jobTaskRunLogs.forEach(jobTaskRunLog -> {
long timeConsuming = jobTaskRunLog.getEndTime().getTime() - jobTaskRunLog.getStartTime().getTime();
/*
* String iframeHtml = "<iframe src='%s'></iframe>";
* String logFilePath = FILE_SYSTEM_PRE+fileSystemIp+":"+fileSystemPort+"/"+jobTaskRunLog.getLogRemotelyPath();
* String logFileUrl = String.format(iframeHtml,logFilePath);
*/
String logStr = null;
try {
if(null != jobTaskRunLog.getLogRemotelyPath()){
logStr = new String(fileSystem.downloaderFile(jobTaskRunLog.getLogRemotelyPath()), StandardCharsets.UTF_8);
}
} catch (IOException | MyException e) {
e.printStackTrace();
}
stringBuilder.append("<tr align='center'>")
.append(String.format("<td>%s</td>", jobTaskRunLog.getNodeName()))
.append(String.format("<td>%s</td>", DateUtil.format(jobTaskRunLog.getStartTime(), DATE_FORMAT)))
.append(String.format("<td>%s</td>", DateUtil.format(jobTaskRunLog.getEndTime(), DATE_FORMAT)))
.append(String.format("<td>%s</td>", TimeFormatUtil.timeFormat(timeConsuming)))
.append(String.format("<td>%s</td>", "1".equals(jobTaskRunLog.getRunCode()) ? "成功"
: "3".equals(jobTaskRunLog.getRunCode()) ? "补批成功"
: "4".equals(jobTaskRunLog.getRunCode()) ? "补批失败" : "失败"))
.append("<td>")
.append("<div style='display:inline-block;width:100%;word-break:break-all;height: auto;overflow: auto;text-align: left;'>")
.append(String.format("%s", logStr==null?jobTaskRunLog.getRunMsg():logStr))
.append("</div>")
.append("</td></tr>");
});
}
stringBuilder.append("</tbody>")
.append("</table>");
return stringBuilder.toString();
}
@Autowired
public void setDataSource(DataSource dataSource) {
this.dataSource = dataSource;
}
}
\ No newline at end of file
......@@ -129,7 +129,13 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
String preHourDate = preHourLocalDateTime.format(DATE_FORMATTER);
String preHourHour = String.valueOf(preHourLocalDateTime.getHour());
List<FlowStatusSnapshoot> preHourFlowStatusSnapshootList = flowStatusSnapshootMapper.findByDateAndHourAndFlowId(preHourDate, preHourHour, flow.getFlowId());
if (null != preHourFlowStatusSnapshootList || preHourFlowStatusSnapshootList.size() > 0){
flowStatusSnapshootList.addAll(preHourFlowStatusSnapshootList);
}else {
//如果上个小时也没有,就设置为未运行
flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount());
flowStatusSnapshootList.add(flowStatusSnapshoot);
}
}
}
}else {
......@@ -138,7 +144,7 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
flowStatusSnapshootList.add(flowStatusSnapshoot);
}
});
if (flowList != null && flowList.size() > 0){
if (null != flowStatusSnapshootList && flowStatusSnapshootList.size() > 0){
flowStatusSnapshootMapper.saveList(flowStatusSnapshootList);
}
//获取过期的时间戳
......
......@@ -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;
}
......
......@@ -34,6 +34,12 @@
schedule_follow, is_update, scan_mark
</sql>
<select id="findAllFlow" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from flow
</select>
<select id="findHalfAnHourFlow" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
......
......@@ -31,6 +31,16 @@
repeat_count, version_mark, workspace_id, author, principal, version_name, is_inner
</sql>
<select id="findAllByFlowId" resultMap="BaseResultMap">
select <include refid="Base_Column_List" />
from flow_version where flow_id=#{flowId,jdbcType=INTEGER} and remove_mark = '1'
</select>
<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
......
......@@ -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},
......
......@@ -56,6 +56,20 @@
run_source
</sql>
<select id="findAllByFlowId" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from node_version
where remove_mark = '1' and flow_version_id = #{flowId,jdbcType=INTEGER}
</select>
<select id="findAll" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from node_version
where remove_mark = '1'
</select>
<select id="getById" parameterType="java.lang.Integer" resultMap="ResultMapWithBLOBs">
<!-- generated @mbg.generated date: 2019-12-31 -->
select
......
......@@ -32,6 +32,27 @@
flow_id, start_time, end_time, is_alarm,is_inner,fail_fast, flow_node_count, schedule_type, operator, re_run_id
</sql>
<select id="findAll" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
</select>
<select id="findAllError" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
where flow_run_result = '2' or flow_run_result = '4'
or flow_run_result = '5'
</select>
<select id="findAllRunIng" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
where flow_status = '2' or flow_status = '3'
</select>
<!--查询已完结 没有告警的-->
<select id="findRunRecordingByEndAndNotIsAlarm" resultMap="BaseResultMap">
select
......@@ -194,6 +215,18 @@
)
</select>
<select id="findNewStatus" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
where flow_id = #{flowId}
and trigger_time = (
select max(trigger_time)
from run_recording
where flow_id = #{flowId}
)
</select>
<delete id="deleteById" parameterType="java.lang.Integer">
<!-- generated @mbg.generated date: 2019-12-25 -->
delete from run_recording
......
......@@ -13,9 +13,14 @@
<artifactId>myth-core-common</artifactId>
<dependencies>
<dependency>
<!--<dependency>
<groupId>net.oschina.zcx7878</groupId>
<artifactId>fastdfs-client-java</artifactId>
</dependency>-->
<dependency>
<groupId>com.github.tobato</groupId>
<artifactId>fastdfs-client</artifactId>
<version>1.26.1-RELEASE</version>
</dependency>
<!-- slf4j -->
<dependency>
......
package com.byit.filesystem;
import cn.hutool.core.collection.CollectionUtil;
import org.csource.common.MyException;
import org.csource.common.NameValuePair;
import org.csource.fastdfs.ClientGlobal;
import org.csource.fastdfs.StorageClient1;
import org.csource.fastdfs.TrackerClient;
import org.csource.fastdfs.TrackerServer;
import com.github.tobato.fastdfs.FdfsClientConfig;
import com.github.tobato.fastdfs.domain.MateData;
import com.github.tobato.fastdfs.domain.StorePath;
import com.github.tobato.fastdfs.proto.storage.DownloadByteArray;
import com.github.tobato.fastdfs.service.FastFileStorageClient;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.annotation.Import;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.io.ByteArrayInputStream;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
......@@ -21,74 +21,62 @@ import java.util.Set;
*/
@Component
@ConditionalOnExpression("'${myth-job.filestystem}'.equals('FASTDFS')")
@Import(FdfsClientConfig.class)
public class FastDfsFileSystem implements FileSystem {
public FastDfsFileSystem() {
try {
ClientGlobal.init("fdfs_client.conf");
} catch (IOException | MyException e) {
e.printStackTrace();
}
private final FastFileStorageClient fastFileStorageClient;
public FastDfsFileSystem(FastFileStorageClient fastFileStorageClient) {
this.fastFileStorageClient = fastFileStorageClient;
}
@Override
public String uploadFile(byte[] fileBuffer, String fileExtName, Map<String, String> mateDaTa) throws IOException, MyException {
TrackerClient trackerClient = new TrackerClient();
TrackerServer trackerServer = trackerClient.getConnection();
StorageClient1 storageClient1 = new StorageClient1(trackerServer, null);
NameValuePair[] mathList = new NameValuePair[mateDaTa.size()];
if(CollectionUtil.isNotEmpty(mateDaTa)){
Set<Map.Entry<String, String>> mateSet = mateDaTa.entrySet();
int i = 0;
for (Map.Entry<String, String> mate : mateSet) {
mathList[i] = new NameValuePair(mate.getKey(),mate.getValue());
i++;
}
}
String fileId = storageClient1.upload_file1(fileBuffer, fileExtName, mathList);
trackerServer.close();
return fileId;
public String uploadFile(byte[] fileBuffer, String fileExtName, Map<String, String> mateDaTas) {
ByteArrayInputStream byteInput = new ByteArrayInputStream(fileBuffer);
Set<MateData> mateDataSet = new HashSet<>(2);
mateDaTas.forEach((key,value) ->{
MateData mateData = new MateData();
mateData.setName(key);
mateData.setValue(value);
mateDataSet.add(mateData);
});
StorePath storePath = fastFileStorageClient.uploadFile(byteInput, fileBuffer.length, fileExtName, mateDataSet);
return storePath.getFullPath();
}
@Override
public byte[] downloaderFile(String remPath) throws IOException, MyException {
TrackerClient trackerClient = new TrackerClient();
TrackerServer trackerServer = trackerClient.getConnection();
StorageClient1 storageClient1 = new StorageClient1(trackerServer, null);
byte[] bytes = storageClient1.download_file1(remPath);
trackerServer.close();
return bytes;
public byte[] downloaderFile(String remPath) {
DownloadByteArray callback = new DownloadByteArray();
return fastFileStorageClient.downloadFile(getPathGroup(remPath), getPath(remPath), callback);
}
@Override
public void fileRemove(String remPath) throws IOException, MyException {
TrackerClient trackerClient = new TrackerClient();
TrackerServer trackerServer = trackerClient.getConnection();
StorageClient1 storageClient1 = new StorageClient1(trackerServer, null);
storageClient1.delete_file1(remPath);
trackerServer.close();
public void fileRemove(String remPath) {
fastFileStorageClient.deleteFile(remPath);
}
@Override
public Map<String, String> getFileMate(String filePath) throws IOException, MyException {
TrackerClient trackerClient = new TrackerClient();
TrackerServer trackerServer = trackerClient.getConnection();
StorageClient1 storageClient1 = new StorageClient1(trackerServer, null);
NameValuePair[] metadata1 = storageClient1.get_metadata1(filePath);
Map<String,String> map = new HashMap<String,String>(5);
if(metadata1 != null){
for (int i = 0; i <metadata1.length ; i++) {
map.put(metadata1[i].getName(),metadata1[i].getValue());
public Map<String, String> getFileMate(String filePath) {
Set<MateData> metadataSet = fastFileStorageClient.getMetadata(getPathGroup(filePath), getPath(filePath));
HashMap<String, String> map = new HashMap<>(2);
metadataSet.forEach(mateData -> {
map.put(mateData.getName(),mateData.getValue());
});
return map;
}
}else{
map.put("filename","filename"+filePath.substring(filePath.lastIndexOf("."),filePath.length()));
public static String getPathGroup(String fullPath){
int i = fullPath.indexOf("/");
return fullPath.substring(0, i);
}
return map;
public static String getPath(String fullPath){
int i = fullPath.indexOf("/");
return fullPath.substring(i+1);
}
public static void main(String[] args) {
String filePath = "sadsadsadsadsad.py";
System.out.println(filePath.substring(filePath.lastIndexOf("."), filePath.length()));
System.out.println(getPath("ddmp/M00/00/09/CgB4Al6mSVSAHi-rAAAABhCiqmE437.log"));
}
}
package com.byit.filesystem;
import org.csource.common.MyException;
import java.io.IOException;
import java.util.Map;
......@@ -17,26 +16,23 @@ public interface FileSystem {
* @param mateDaTA 文件源信息
* @return 上传路径
* @throws IOException
* @throws MyException
*/
String uploadFile(byte[] fileBuffer, String fileExtName, Map<String, String> mateDaTA) throws IOException, MyException;
String uploadFile(byte[] fileBuffer, String fileExtName, Map<String, String> mateDaTA) throws IOException;
/**
* 文件下载接口
* @param remPath 远程地址
* @return 文件字节数组
* @throws IOException
* @throws MyException
*/
byte[] downloaderFile(String remPath) throws IOException, MyException;
byte[] downloaderFile(String remPath) throws IOException;
/**
* 文件删除
* @param remPath 远程地址
* @throws IOException
* @throws MyException
*/
void fileRemove(String remPath) throws IOException, MyException;
void fileRemove(String remPath) throws IOException;
/**
......@@ -44,9 +40,8 @@ public interface FileSystem {
* @param filePath
* @return
* @throws IOException
* @throws MyException
*/
Map<String,String> getFileMate(String filePath) throws IOException, MyException;
Map<String,String> getFileMate(String filePath) throws IOException;
}
package com.byit.dto.web;
import com.byit.enums.IEnum;
import com.byit.enums.TransferResultEnum;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
......@@ -32,8 +33,8 @@ public class ResponseResult<T> implements Serializable {
*/
public static<T> ResponseResult<T> ok(T t){
ResponseResult<T> tResponseResult = new ResponseResult<>();
tResponseResult.setCode("0000000");
tResponseResult.setMsg("成功");
tResponseResult.setCode(TransferResultEnum.SUCCESS.getCode());
tResponseResult.setMsg(TransferResultEnum.SUCCESS.getMsg());
tResponseResult.setResult(t);
return tResponseResult;
}
......@@ -44,8 +45,8 @@ public class ResponseResult<T> implements Serializable {
*/
public static<T> ResponseResult<T> ok(){
ResponseResult<T> tResponseResult = new ResponseResult<>();
tResponseResult.setCode("0000000");
tResponseResult.setMsg("成功");
tResponseResult.setCode(TransferResultEnum.SUCCESS.getCode());
tResponseResult.setMsg(TransferResultEnum.SUCCESS.getMsg());
return tResponseResult;
}
......@@ -71,7 +72,7 @@ public class ResponseResult<T> implements Serializable {
public static<T> ResponseResult<T> error(String msg){
ResponseResult<T> tResponseResult = new ResponseResult<>();
tResponseResult.setMsg(msg);
tResponseResult.setCode("00000500");
tResponseResult.setCode(TransferResultEnum.ERROR.getCode());
return tResponseResult;
}
......@@ -94,7 +95,7 @@ public class ResponseResult<T> implements Serializable {
*/
public boolean isSuccess(){
//成功
if ("0000000".equals(this.code)){
if (TransferResultEnum.SUCCESS.getCode().equals(this.code)){
return true;
}
//失败
......
......@@ -18,17 +18,16 @@ import java.io.Serializable;
@NoArgsConstructor
public class ReturnResult<T> implements Serializable {
public static final long serialVersionUID = 1573630876693L;
public static final ReturnResult<String> SUCCESS = new ReturnResult<String>(null);
public static final ReturnResult FAIL = new ReturnResult(JobResultEnum.FAIL.getRes(), JobResultEnum.FAIL.getMsg());
public static final ReturnResult FAIL_TIMEOUT = new ReturnResult(JobResultEnum.FAIL_TIMEOUT.getRes(),JobResultEnum.FAIL_TIMEOUT.getMsg());
public static final ReturnResult<String> SUCCESS = new ReturnResult<>(null);
public static final ReturnResult<String> FAIL = new ReturnResult<>(JobResultEnum.FAIL.getCode(), JobResultEnum.FAIL.getMsg());
private String code;
private String msg;
private T content;
/**
* 默认就是错误
* @param code
* @param msg
* @param code 错误码
* @param msg 执行信息
*/
public ReturnResult(String code, String msg) {
this.code = code;
......@@ -37,10 +36,10 @@ public class ReturnResult<T> implements Serializable {
/**
* 默认就是成功
* @param content
* @param content 结果集
*/
private ReturnResult(T content) {
this.code = JobResultEnum.SUCCESS.getRes();
this.code = JobResultEnum.SUCCESS.getCode();
this.msg = JobResultEnum.SUCCESS.getMsg();
this.content = content;
}
......
......@@ -5,24 +5,23 @@ package com.byit.enums;
* @author huangfu
*/
public enum JobResultEnum implements IEnum {
SUCCESS("100200","任务执行成功","1"),
FAIL("100500","任务执行失败","2"),
FAIL_TIMEOUT("100502","超时错误","2"),
DISPATCH_SUCCESS("200200","调度成功","1"),
DISPATCH_FAIL("200500","调度失败","2"),
SUCCESS("100200","任务执行成功"),
FAIL("100500","任务执行失败"),
FAIL_TIMEOUT("100502","超时错误"),
DISPATCH_SUCCESS("200200","调度成功"),
DISPATCH_FAIL("200500","调度失败"),
KILL_SUCCESS("400000","执行数据被kill"),
RUN_MSG_FAIL("300000","执行结果异常","2");
RUN_MSG_FAIL("300000","执行结果异常");
private String code;
private String msg;
private String res;
JobResultEnum() {
}
JobResultEnum(String code, String msg,String res) {
JobResultEnum(String code, String msg) {
this.code = code;
this.msg = msg;
this.res = res;
}
......@@ -35,7 +34,4 @@ public enum JobResultEnum implements IEnum {
public String getMsg() {
return this.msg;
}
public String getRes() {
return this.res;
}
}
\ No newline at end of file
package com.byit.enums;
/**
* 调用结果枚举 标准
* @author huangfu
*/
public enum TransferResultEnum implements IEnum {
SUCCESS("000000","调用成功"),
ERROR("500000","发生错误")
;
private String code;
private String msg;
TransferResultEnum(String code, String msg) {
this.code = code;
this.msg = msg;
}
@Override
public String getCode() {
return code;
}
@Override
public String getMsg() {
return msg;
}
}
package com.byit.enums.task;
/**
* @author huangfu
*/
public enum RunResultEnum {
TRIGGER_SUCCESS("1","调度成功"),
TRIGGER_ERROR("2","调度失败"),
RUN_SUCCESS("1","执行成功"),
RUN_ERROR("2","执行失败"),
KILL_SUCCESS("5","节点被kill")
;
private String code;
private String msg;
RunResultEnum(String code, String msg) {
this.code = code;
this.msg = msg;
}
public String getCode() {
return code;
}
public String getMsg() {
return msg;
}
}
package com.byit.enums.task;
/**
* 运行类型枚举
* @author huangfu
*/
public enum RunTypeEnum {
PLUGIN_RUN("2","插件执行!"),
EXECUTIVE_MACHINE_RUN("1","执行机执行!")
;
private String code;
private String msg;
RunTypeEnum(String code, String msg) {
this.code = code;
this.msg = msg;
}
public String getCode() {
return code;
}
public String getMsg() {
return msg;
}
}
......@@ -2,6 +2,7 @@ package com.byit.web.advice;
import com.byit.dto.web.ResponseResult;
import com.byit.enums.IEnum;
import com.byit.enums.TransferResultEnum;
import com.byit.exception.DataValidationException;
import com.byit.job.exceptions.IException;
import org.springframework.web.bind.annotation.ExceptionHandler;
......@@ -26,7 +27,7 @@ public class ExceptionHandle {
if(e instanceof IException){
return error(e);
}else if(e instanceof DataValidationException){
return ResponseResult.error("500",e.getMessage());
return ResponseResult.error(TransferResultEnum.ERROR.getCode(),e.getMessage());
}
return ResponseResult.error(e.getMessage());
}
......
......@@ -14,5 +14,5 @@ public interface ScriptExecutorService {
* @param scriptDto 参数
* @return 调用结果
*/
DispatchResponseDto runPythonScript(ScriptDto scriptDto);
DispatchResponseDto runScript(ScriptDto scriptDto);
}
......@@ -33,7 +33,7 @@ public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpReq
200,
60L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(1000),
new LinkedBlockingQueue<>(1000),
r ->new Thread(r, "Netty RunJobServerHandler serverThread-" + r.hashCode()));
private static final String PENG = "PENG";
......@@ -42,23 +42,23 @@ public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpReq
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
log.debug("----------------------有请求过来了------------------------");
DispatchResponseDto dispatchResponseDto = new DispatchResponseDto();
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_FAIL.getRes());
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_FAIL.getCode());
dispatchResponseDto.setMsg(JobResultEnum.DISPATCH_FAIL.getMsg());
if(req != null){
//解析 调度中心 的数据对象
AdminSenPluginDto adminSenPluginDto = analysisParam(req.content( ));
if(null == adminSenPluginDto){
throw new Exception("核心参数 为 null");
throw new Exception("核心参数为空!");
}
//检测是否有心跳参数,有心跳参数则为测试参数,且为PENG的话,服务端回复 PONG
String heartbeat = adminSenPluginDto.getHeartbeat();
if(null == heartbeat){
JOB_TRIGGER_POOL.execute(new RunJobThread(adminSenPluginDto));
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_SUCCESS.getRes());
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_SUCCESS.getCode());
dispatchResponseDto.setMsg(JobResultEnum.DISPATCH_SUCCESS.getMsg());
}else if(PENG.equals(heartbeat)){
log.debug("----------调度平台心跳检测-------------");
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_SUCCESS.getRes());
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_SUCCESS.getCode());
dispatchResponseDto.setMsg(JobResultEnum.DISPATCH_SUCCESS.getMsg());
dispatchResponseDto.setContent(PONG);
}
......@@ -78,8 +78,8 @@ public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpReq
/**
* 格式化参数
* @param byteBuf
* @return
* @param byteBuf 将缓冲流更改为字符串
* @return 调用的必要信息承载类
*/
private AdminSenPluginDto analysisParam(ByteBuf byteBuf){
return JSON.parseObject(byteBuf.toString(CharsetUtil.UTF_8),AdminSenPluginDto.class);
......@@ -87,12 +87,11 @@ public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpReq
/**
* 异常捕获
* @param ctx
* @param cause
* @throws Exception
* @param ctx 上线文对象
* @param cause 异常信息
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
......
......@@ -45,16 +45,16 @@ public class RunJobThread implements Runnable {
log.info("---------服务器端:{},花费时间:{}-------------", jobRunResultDto,jobRunResultDto.getStartTime().getTime()-jobRunResultDto.getEndTime().getTime());
}
private ReturnResult runJob(String jobHandlerName, String param){
private ReturnResult<String> runJob(String jobHandlerName, String param){
Class<? extends IJobHandler> jobClass = JobUtils.jobCache.get(jobHandlerName);
try {
IJobHandler iJobHandler = jobClass.newInstance();
return iJobHandler.execute(param);
} catch (Exception e) {
e.printStackTrace( );
ReturnResult returnResult = new ReturnResult();
ReturnResult<String> returnResult = new ReturnResult<>();
returnResult.setMsg(e.getMessage());
returnResult.setCode(JobResultEnum.FAIL.getRes());
returnResult.setCode(JobResultEnum.FAIL.getCode());
return returnResult;
}
......
......@@ -101,6 +101,10 @@ public class JobUtils {
*/
public static final String REQUEST_LOADNODESTATISTICDATA = "/api/flow/loadNodeStatisticData";
/**
* 获取当前的工作流运行状态
*/
public static final String REQUEST_LOADCURRENTSTATUS = "/api/flow/loadCurrentStatus";
/**
* 补批工作流
*/
public static final String REQUEST_REPAIRFLOW = "/api/flow/repairFlow";
......@@ -139,6 +143,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";
......@@ -424,6 +432,21 @@ public class JobUtils {
}
/**
* 获取当前的工作流运行状态
* @param workspaceName 工作空间名称
* @param flowNames 工作流名称,多个以","分割
* @return
*/
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.info("--------------------获取当前的工作流运行状态接口调用成功,结果为:{}------------------------",response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 补批工作流
* @param workspaceName 工作空间名称
* @param flowName 工作流名称
......@@ -597,6 +620,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
......
......@@ -8,7 +8,6 @@ import com.byit.dto.executor.ScriptParamAndPlaceholderDto;
import com.byit.enums.PlaceholderEnum;
import com.byit.executor.api.ScriptExecutorService;
import com.byit.process.MythJobProcess;
import com.byit.process.ProcessFailureException;
import com.byit.filesystem.FileSystem;
import com.byit.enums.JobResultEnum;
import com.byit.job.utils.PlaceholderUtils;
......@@ -17,12 +16,10 @@ import com.byit.pool.RunThreadPool;
import com.byit.rpc.remoting.provider.annotation.RpcService;
import com.byit.utils.ServiceInfoUtil;
import lombok.extern.slf4j.Slf4j;
import org.csource.common.MyException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.io.*;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
......@@ -37,30 +34,49 @@ import java.util.*;
@RpcService
@Slf4j
public class ScriptExecutorServiceImpl implements ScriptExecutorService {
@Value("${myth-job.log.root.path}")
private String rootLogPath;
/**
* 带点的日志后缀
*/
public static final String LOG_RETOUCH_SUFFIX = ".log";
/**
* 日志后缀
*/
public static final String LOG_SUFFIX = "log";
/**
* 脚本生成路径
*/
@Value("${myth-job.script.root.path}")
private String rootScriptPath;
@Resource
private StringRedisTemplate stringRedisTemplate;
private final ServiceInfoUtil serviceInfoUtil;
/**
* 节点执行成功
*/
private static final Integer NODE_RUN_SUCCESS_CODE = 0;
/**
* 节点被杀死
*/
private static final Integer NODE_KILL_CODE = -100;
private static final String KILL_MESSAGE = "该任务节点已被强制kill!";
private static final String FILE_NAME = "filename";
private final FileSystem fileSystem;
private final StringRedisTemplate stringRedisTemplate;
public ScriptExecutorServiceImpl(ServiceInfoUtil serviceInfoUtil, FileSystem fileSystem) {
this.serviceInfoUtil = serviceInfoUtil;
public ScriptExecutorServiceImpl(StringRedisTemplate stringRedisTemplate, FileSystem fileSystem) {
this.stringRedisTemplate = stringRedisTemplate;
this.fileSystem = fileSystem;
}
/**
* 执行脚本任务
* @param scriptDto 参数
* @return 返回调用的结果 执行结果会通过异步Http的方式返回给调度中心
*/
@Override
public DispatchResponseDto runPythonScript(ScriptDto scriptDto) {
log.info("----------------runPythonScript start{} ------------------------",scriptDto);
public DispatchResponseDto runScript(ScriptDto scriptDto) {
log.info("----------------runScript 开始调用脚本执行机 start{} ------------------------",scriptDto);
DispatchResponseDto dispatchResponseDto;
try {
RunThreadPool.SCRIPT_RUN_THREAD_POOL.execute(()->{
startScript(scriptDto);
});
RunThreadPool.SCRIPT_RUN_THREAD_POOL.execute(()-> startScript(scriptDto));
dispatchResponseDto = DispatchResponseDto.builder()
.code(JobResultEnum.DISPATCH_SUCCESS.getCode())
......@@ -73,26 +89,27 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
.build();
}
dispatchResponseDto.setUrl(ServiceInfoUtil.getIpAndPort());
log.info("----------------runPythonScript end---------------------");
log.info("----------------runScript 调用脚本执行机结束 end---------------------");
return dispatchResponseDto;
}
/**
* 这个在执行成功后会向调度中心发送一个http请求返回执行的结果信息
* 异步执行脚本信息
* @param scriptDto 执行脚本所必须的参数
*/
public void startScript(ScriptDto scriptDto){
log.info("--------------runPythonScript,脚本调用开始-----------");
//创建回复对象
JobRunResultDto jobRunResultDto = new JobRunResultDto();
String logPath = "";
try {
jobRunResultDto.setStartTime(new Date());
//设定运行标识
jobRunResultDto.setJobRunId(scriptDto.getRunId());
//获取回调通知URL
String callbackUrl = scriptDto.getCallbackUrl();
String remotePath = scriptDto.getRemotePath();
String command = scriptDto.getCommand();
String param = scriptDto.getParam();
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = null;
String param = scriptDto.getParam();
String command = scriptDto.getCommand();
String remotePath = scriptDto.getRemotePath();
if(null != param){
scriptParamAndPlaceholderDto = JSON.parseObject(param, ScriptParamAndPlaceholderDto.class);
}
......@@ -106,65 +123,65 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
if(scriptParamAndPlaceholderDto != null){
command = PlaceholderUtils.commandReplace(command,scriptParamAndPlaceholderDto.getParam());
}
assert command != null;
List<String> cmdList = Arrays.asList(command.split(" "));
MythJobProcess mythJobProcess = new MythJobProcess(cmdList, null, null, scriptDto.getLogId(), stringRedisTemplate);
//保存日志
String logData = mythJobProcess.call();
//成功 是0 失败是其他的 杀死是-100
int exitCode = mythJobProcess.getExitCode();
if(exitCode == -100){
logData = "该任务节点已被强制kill!";
if(exitCode == NODE_KILL_CODE){
logData = KILL_MESSAGE;
}
byte[] logDataByte = stringToByteArray(logData);
Map<String,String> fileMateData = new HashMap<String,String>(2);
fileMateData.put("filename",scriptDto.getRunId()+scriptDto.getRunId()+".log");
String logPath = "";
try {
Map<String,String> fileMateData = new HashMap<>(2);
fileMateData.put(FILE_NAME,scriptDto.getRunId()+scriptDto.getRunId()+ LOG_RETOUCH_SUFFIX);
//不为空 则追加
if(scriptDto.getLogRemotePath() != null){
byte[] sourceLogByte = fileSystem.downloaderFile(scriptDto.getLogRemotePath());
byte[] resultLogByte = mergeFile(sourceLogByte, logDataByte);
//上传日志文件
logPath = fileSystem.uploadFile(resultLogByte,"log",fileMateData);
logPath = fileSystem.uploadFile(resultLogByte, LOG_SUFFIX,fileMateData);
//删除原有的日志文件
fileSystem.fileRemove(scriptDto.getLogRemotePath());
}else{
//上传日志文件
logPath = fileSystem.uploadFile(logDataByte,"log",fileMateData);
logPath = fileSystem.uploadFile(logDataByte,LOG_SUFFIX,fileMateData);
}
if(exitCode == 0){
//判断任务的执行状态 0成功 -100kill 其他 失败
if(exitCode == NODE_RUN_SUCCESS_CODE){
jobRunResultDto.setReturnResult(ReturnResult.SUCCESS);
}else if(exitCode == -100){
ReturnResult returnResult = new ReturnResult("5","节点被kill!");
}else if(exitCode == NODE_KILL_CODE){
ReturnResult<String> returnResult = new ReturnResult<>();
returnResult.setCode(JobResultEnum.KILL_SUCCESS.getCode());
returnResult.setMsg(JobResultEnum.KILL_SUCCESS.getMsg());
jobRunResultDto.setReturnResult(returnResult);
}else{
jobRunResultDto.setReturnResult(ReturnResult.FAIL);
}
} catch (ProcessFailureException ignored){
jobRunResultDto.setReturnResult(ReturnResult.FAIL);
} catch (IOException | MyException e) {
jobRunResultDto.setReturnResult(ReturnResult.FAIL);
e.printStackTrace();
} catch (Exception e){
ReturnResult<String> returnResult = new ReturnResult<>();
returnResult.setCode(ReturnResult.FAIL.getCode());
returnResult.setMsg(e.getMessage());
jobRunResultDto.setReturnResult(returnResult);
}
log.info("-----------执行命令,{}-----------",command);
//设置结束时间
jobRunResultDto.setEndTime(new Date());
jobRunResultDto.setLogId(scriptDto.getLogId());
//设置远程日志文件的路径
jobRunResultDto.setLogRemotelyPath(logPath);
cn.hutool.http.HttpUtil.post(callbackUrl, JSON.toJSONString(jobRunResultDto));
cn.hutool.http.HttpUtil.post(scriptDto.getCallbackUrl(), JSON.toJSONString(jobRunResultDto));
log.info("--------------runPythonScript,脚本调用结束-----------");
}
/**
* 合并两个字节数组
* @param sourceByte
* @param targetByte
* @return
* @param sourceByte 字节数组1
* @param targetByte 字节数组2
* @return 合并后的字节数组
*/
private byte[] mergeFile(byte[] sourceByte, byte[] targetByte){
byte[] result = new byte[sourceByte.length+targetByte.length];
......@@ -186,13 +203,11 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
OutputStream out = null;
File file = null;
File file;
try {
//下载脚本文件
byte[] scriptByteArray = fileSystem.downloaderFile(remotePath);
/**
* 替换脚本占位符
*/
//替换脚本占位符
if(scriptParamAndPlaceholderDto != null){
scriptByteArray = PlaceholderUtils.resolvePlaceholders(scriptByteArray,scriptParamAndPlaceholderDto.getPlaceholder());
}
......@@ -205,7 +220,7 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
//写出脚本文件
out.write(scriptByteArray);
return file.getPath();
} catch (IOException | MyException e) {
} catch (IOException e) {
e.printStackTrace();
} finally {
if(out != null){
......@@ -222,8 +237,8 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
/**
* 将字符串转换我数组
* @param logData
* @return
* @param logData 日志字符串
* @return 转换的字节数组
*/
private byte[] stringToByteArray(String logData) {
return logData.getBytes(StandardCharsets.UTF_8);
......@@ -231,19 +246,10 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
/**
* 格式化当前时间
* @return
* @return 格式化后的时间字符串
*/
private String formatDate(){
LocalDate date = LocalDate.now();
return date.format(DateTimeFormatter.ofPattern("yyyyMMdd"));
}
public static void main(String[] args) throws IOException {
File file = new File("D:\\2020project\\byit-myth-job\\demo-client\\byit-demo-client\\src\\main\\java\\com\\byit\\job\\Mains.java");
System.out.println(file.getPath());
System.out.println(file.getAbsolutePath());
System.out.println(file.getCanonicalPath());
}
}
......@@ -35,3 +35,11 @@ authentication:
header-name: token
expire: 43200 # 外部token有效期为12小时
pub-key: client/pub.key # 解密
fdfs:
so-timeout: 1500
connect-timeout: 600
pool:
jmx-enabled: false
tracker-list:
- 10.0.120.2:22122
\ No newline at end of file
......@@ -35,3 +35,11 @@ authentication:
header-name: token
expire: 43200 # 外部token有效期为12小时
pub-key: client/pub.key # 解密
fdfs:
so-timeout: 1500
connect-timeout: 600
pool:
jmx-enabled: false
tracker-list:
- 10.0.120.2:22122
\ No newline at end of file
......@@ -21,8 +21,8 @@ myth-job:
spring:
redis:
database: 0
host: ${Redis_IP}
port: ${Redis_port}
host: 10.0.120.208
port: 6379
password:
timeout: 3000
pool:
......@@ -36,3 +36,13 @@ authentication:
header-name: token
expire: 43200 # 外部token有效期为12小时
pub-key: client/pub.key # 解密
fdfs:
so-timeout: 1500
connect-timeout: 600
pool:
jmx-enabled: false
tracker-list:
- 10.0.120.216:22122
- 10.0.120.217:22122
- 10.0.120.218:22122
\ No newline at end of file
......@@ -5,7 +5,7 @@ myth-rpc:
biz: byit-myth-job
env: test
remoting:
port: ${Remote_PORT}
port: 7776
logging:
config: classpath:logback.xml
......@@ -36,3 +36,13 @@ authentication:
header-name: token
expire: 43200 # 外部token有效期为12小时
pub-key: client/pub.key # 解密
fdfs:
so-timeout: 1500
connect-timeout: 600
pool:
jmx-enabled: false
tracker-list:
- 10.0.120.216:22122
- 10.0.120.217:22122
- 10.0.120.218:22122
\ No newline at end of file
#路由规则参考byit-myth-rpc中的LoadBalance枚举类中的类型 默认是轮询
gateway:
load:
balance: ROUND
......
......@@ -91,6 +91,30 @@
</resources>
</configuration>
</plugin>
<plugin>
<groupId>com.spotify</groupId>
<artifactId>docker-maven-plugin</artifactId>
<version>1.0.0</version>
<configuration>
<imageName>${docker.image.prefix}/${project.artifactId}</imageName>
<!-- optionally overwrite tags every time image is built with docker:build -->
<forceTags>true</forceTags>
<imageTags>
<imageTag>${docker.image.tag}</imageTag>
</imageTags>
<dockerDirectory>${project.basedir}</dockerDirectory>
<resources>
<resource>
<targetPath>/</targetPath>
<directory>${project.build.directory}</directory>
<include>${project.build.finalName}.jar</include>
</resource>
</resources>
<serverId>harbor</serverId>
<registryUrl>10.0.101.122</registryUrl>
</configuration>
</plugin>
</plugins>
</build>
......
......@@ -33,3 +33,4 @@ spring:
static-path-pattern: /static/**
resources:
static-locations: classpath:/static/
......@@ -49,20 +49,6 @@
</execution>
</executions>
</plugin>
<!-- Javadoc -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-javadoc-plugin</artifactId>
<version>3.0.1</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
</plugin>
<!-- GPG -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
......
......@@ -45,7 +45,7 @@ public class NettyClientHandler extends SimpleChannelInboundHandler<RpcResponse>
logger.debug(">>>>>>>>>>> myth-rpc netty client close an idle channel.");*/
nettyConnectClient.send(Beat.BEAT_PING); // beat N, close if fail(may throw error)
logger.info(">>>>>>>>>>> myth-rpc netty client send beat-ping."+ctx.channel().id().asShortText());
logger.info(">>>>>>>>>>> myth-rpc netty client send beat-ping.");
} else {
super.userEventTriggered(ctx, evt);
......
......@@ -71,7 +71,7 @@ public class NettyServerHandler extends SimpleChannelInboundHandler<RpcRequest>
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent){
ctx.channel().close(); // beat 3N, close if idle
logger.info(">>>>>>>>>>> myth-rpc provider netty server close an idle channel."+ctx.channel().id());
logger.info(">>>>>>>>>>> myth-rpc provider netty server close an idle channel.");
} else {
super.userEventTriggered(ctx, evt);
}
......
......@@ -15,7 +15,6 @@ public class NettyPluginClient extends PluginClient {
@Override
public void send(String address, PluginRpcRequestPacket pluginRpcRequestPacket) throws Exception {
System.out.println("--------------我发送信息了");
PluginConnectClient.send(pluginRpcRequestPacket,address,pluginConnectClientImpl,pluginClientInitialization);
}
}
......@@ -32,7 +32,7 @@ public class NettyClientHandler extends SimpleChannelInboundHandler<PluginRpcRes
//判断事件是否是心跳事件
if( evt instanceof IdleStateEvent){
pluginConnectClient.send(PluginBeat.PLUGIN_RPC_REQUEST_PACKET);
System.out.println("------客户端发送心跳请求-----"+ctx.channel().id().asShortText());
System.out.println("------客户端发送心跳请求-----");
}else{
super.userEventTriggered(ctx, evt);
}
......
......@@ -23,6 +23,9 @@ import java.util.concurrent.ThreadPoolExecutor;
*/
public class NettyPluginServerHandler extends SimpleChannelInboundHandler<PluginRpcRequestPacket> {
public static final String TOKEN_NAME = "token";
public static final String HEADER_TOKEN_NAME = TOKEN_NAME;
public static final String HEADER_CONTENT_TYPE = "contentType";
private PluginServerFactory pluginServerFactory;
private ThreadPoolExecutor threadPoolExecutor;
......@@ -32,15 +35,14 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
}
/**
* //TODO 这一块有问题 不能返回两次结果 考虑能否加一个结果回调,回调的方式由自己实现
* @param ctx
* @param msg
* @throws Exception
* 接收到消费方的请求 开始处理消费方的请求
* @param ctx 上下文对象
* @param msg 消息对象
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, PluginRpcRequestPacket msg) throws Exception {
protected void channelRead0(ChannelHandlerContext ctx, PluginRpcRequestPacket msg) {
if(PluginBeat.BEAT_ID.equals(msg.getRequestId())){
System.out.println("------接收到客户端的心跳连接-------"+ctx.channel().id().asShortText());
System.out.println("------接收到客户端的心跳连接-------");
return;
}
PluginRpcResponsePacket transferPluginRpcResponse = new PluginRpcResponsePacket();
......@@ -55,20 +57,20 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
ReturnResult<String> execute = iJobHandler.execute(msg.getParam());
long endTime = System.currentTimeMillis();
rpcResponsePacket.setResult(execute);
rpcResponsePacket.setCode("000000");
rpcResponsePacket.setMsg("SUCCESS");
rpcResponsePacket.setCode(JobResultEnum.SUCCESS.getCode());
rpcResponsePacket.setMsg(JobResultEnum.SUCCESS.getMsg());
rpcResponsePacket.setStatus(true);
rpcResponsePacket.setRunTime(endTime-startTime);
rpcResponsePacket.setExtension(msg.getExtension());
sendMsg(rpcResponsePacket,msg);
}catch (Throwable e){
rpcResponsePacket.setCode("500000");
rpcResponsePacket.setCode(JobResultEnum.FAIL.getCode());
rpcResponsePacket.setMsg(e.getMessage());
rpcResponsePacket.setStatus(false);
rpcResponsePacket.setExtension(msg.getExtension());
ReturnResult<String> execute = new ReturnResult<>();
execute.setMsg(e.getMessage());
execute.setCode(JobResultEnum.FAIL.getRes());
execute.setCode(JobResultEnum.FAIL.getCode());
rpcResponsePacket.setResult(execute);
sendMsg(rpcResponsePacket,msg);
throw new RuntimeException(e);
......@@ -91,8 +93,8 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
String responseStr = JSON.toJSONString(rpcResponsePacket);
HttpRequest post = HttpRequest.post(msg.getCallbackUrl());
post.header("token", "Token");
post.header("contentType","application/json");
post.header(HEADER_TOKEN_NAME, "Token");
post.header(HEADER_CONTENT_TYPE,"application/json");
post.body(responseStr).execute();
System.out.println("-------消息回复成功-----------");
}
......@@ -101,10 +103,9 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
* 异常处理
* @param ctx 上下文对象
* @param cause 异常对象
* @throws Exception 异常信息
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
......
......@@ -34,7 +34,7 @@ public class IndexController {
@ResponseBody
public DispatchResponseDto say1(String name){
DispatchResponseDto dispatchResponseDto = scriptExecutorService.runPythonScript(null);
DispatchResponseDto dispatchResponseDto = scriptExecutorService.runScript(null);
return dispatchResponseDto;
}
}
......@@ -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);
}
......
package com.byit.job;
import com.byit.filesystem.FastDfsFileSystem;
import org.csource.common.MyException;
import java.io.IOException;
......@@ -12,10 +11,10 @@ import java.io.IOException;
* @date: 2019/11/20 12:18
**/
public class Mains {
public static void main(String[] args) throws InterruptedException, IOException, MyException {
FastDfsFileSystem fastDfsFileSystem = new FastDfsFileSystem();
public static void main(String[] args) {
/*FastDfsFileSystem fastDfsFileSystem = new FastDfsFileSystem(fastFileStorageClient);
byte[] bytes = fastDfsFileSystem.downloaderFile("ddmp/M00/00/00/CgB4Al5wa36ADcUtAAAAiVc2FQ85978.py");
System.out.println(new String(bytes));
System.out.println(new String(bytes));*/
/*new JobRunServerLauncher("/plugin.xml");
JobUtils.jobCache.forEach((key,value) ->{
......
......@@ -13,7 +13,7 @@ import org.springframework.stereotype.Component;
* @author huangfu
*/
@Component
@TaskHandler(expand = "{sadasdadasdasdasd}",cron = "0 0/2 * * * ?",taskName = "sentEmailServer",autoPublish = true,publishUrl = "http://127.0.0.1:8081/myth-job-admin/api/node/autoAddJavaTask")
@TaskHandler(expand = "{sadasdadasdasdasd}",cron = "0 0/2 * * * ?",taskName = "sentEmailServer",autoPublish = false,publishUrl = "http://127.0.0.1:8081/myth-job-admin/api/node/autoAddJavaTask")
@Slf4j
public class SentEmailServer extends BaseJobHandler {
@Override
......
......@@ -27,6 +27,8 @@
<relativePath/>
</parent>
<properties>
<docker.image.prefix>10.0.101.122/myth</docker.image.prefix>
<docker.image.tag>v1.0</docker.image.tag>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
......@@ -69,12 +71,20 @@
<springfox-swagger-ui.version>2.9.2</springfox-swagger-ui.version>
<org-apache-commons.version>1.3</org-apache-commons.version>
<fastdfs-client-java-version>1.27.0.0</fastdfs-client-java-version>
<fastdfs-client-version>1.26.1-RELEASE</fastdfs-client-version>
</properties>
<dependencyManagement>
<dependencies>
<!-- fastdfs-client 依赖-->
<dependency>
<groupId>com.github.tobato</groupId>
<artifactId>fastdfs-client</artifactId>
<version>${fastdfs-client-version}</version>
</dependency>
<dependency>
<groupId>net.oschina.zcx7878</groupId>
<artifactId>fastdfs-client-java</artifactId>
......@@ -214,4 +224,32 @@
</snapshotRepository>
</distributionManagement>
<build>
<plugins>
<plugin>
<groupId>com.spotify</groupId>
<artifactId>docker-maven-plugin</artifactId>
<version>1.0.0</version>
<configuration>
<imageName>${docker.image.prefix}/${project.artifactId}</imageName>
<!-- optionally overwrite tags every time image is built with docker:build -->
<forceTags>true</forceTags>
<imageTags>
<imageTag>${docker.image.tag}</imageTag>
</imageTags>
<dockerDirectory>${project.basedir}</dockerDirectory>
<resources>
<resource>
<targetPath>/</targetPath>
<directory>${project.build.directory}</directory>
<include>${project.build.finalName}.jar</include>
</resource>
</resources>
<serverId>harbor</serverId>
<registryUrl>10.0.101.122</registryUrl>
</configuration>
</plugin>
</plugins>
</build>
</project>
\ No newline at end of file
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