Skip to content

B
byit-myth-job
Commit e640ec86 by guominglei
Options

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

parents edb3b0a1 e9acfd09
Showing with 292 additions and 126 deletions
byit-myth-admin/src/main/java/com/byit/service/ApiFlowService.java
......@@ -36,7 +36,7 @@ public interface ApiFlowService {
* @param flow
* @return
*/
void validateFlow(Integer workspaceId, PluginFlow flow);
void validateFlow(Integer workspaceId, PluginFlow flow, boolean isInner);
/**
* 重跑任务
......
byit-myth-admin/src/main/java/com/byit/service/impl/ApiFlowServiceImpl.java
......@@ -158,10 +158,10 @@ public class ApiFlowServiceImpl implements ApiFlowService {
flow.setTriggerNextTime(StringUtils.isEmpty(pluginFlow.getConfig().getFlowCron()) ? null : new CronExpression(pluginFlow.getConfig().getFlowCron()).getNextValidTimeAfter(new Date()).getTime());
//设置超时时间，未设置默认30分钟
flow.setFlowTimeout(null == pluginFlow.getConfig().getFlowTimeout() ? 1000 * 60 * 120 : pluginFlow.getConfig().getFlowTimeout());
//判断是否是重发
if (pluginFlow.isRePublish()){
Flow oldFlow = flowMapper.getByWorkSpaceAndName(workspaceId, pluginFlow.getName());
//判断是否是重发
if ((pluginFlow.isRePublish() && !isInnner) || (isInnner && null != oldFlow)){
String version = oldFlow.getVersionName();
Integer versionTag = Integer.valueOf(version.split("\\.")[1]);
flow.setFlowId(oldFlow.getFlowId());
......@@ -261,8 +261,11 @@ public class ApiFlowServiceImpl implements ApiFlowService {
node.setNodeName(pluginNode.getName());
if (PluginNodeTypeEnum.FLOW.getCode().equals(pluginNode.getType())){
PluginFlow pluginFlow = (PluginFlow) pluginNode;
//TODO 这段逻辑我不确定需要和明雷对接
pluginFlow.getConfig().setFlowCron(flow.getFlowCron());
//如果是内嵌工作流先保存工作流信息
Flow innerFlow = saveFlow((PluginFlow) pluginNode, flow.getWorkspaceId(), true);
Flow innerFlow = saveFlow(pluginFlow, flow.getWorkspaceId(), true);
node.setIsVirtual(NodePropertyEnum.IS_VIRTUAL.getCode());
node.setMapFlowId(innerFlow.getFlowId());
}else {
......@@ -339,7 +342,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
PluginFlow flow = pluginPackage.getFlow();
//校验工作流参数配置
validateFlow(workspace.getWorkspaceId(), flow);
validateFlow(workspace.getWorkspaceId(), flow, false);
//校验工作流名称是否重复
HashSet<String> flowNameSet = new HashSet<>();
//用来工作流的内嵌关系
......@@ -383,7 +386,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
ValidationUtil.isTrueValidation(!(node instanceof PluginFlow), "工作流的type设置错误，只有当节点为内嵌工作流是type才允许设置为flow");
//校验父工作流和内嵌工作流的执行类型是否一致
ValidationUtil.isTrueValidation(!(flow.getConfig().getExecType().equals(((PluginFlow) node).getConfig().getExecType())), "内嵌工作流的调度配置必须和父工作流配置保持一致！");
validateFlow(workspace.getWorkspaceId(), (PluginFlow) node);
validateFlow(workspace.getWorkspaceId(), (PluginFlow) node, true);
if(FlowPropertyEnum.NO_SCHEDULE.getCode().equals(flow.getConfig().getScheduleFollow())){
ValidationUtil.dataNotBank( ((PluginFlow) node).getConfig().getFlowCron(), "工作流设置为不跟随调度时节点必须设置调度时间！");
}
......@@ -419,7 +422,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
}
@Override
public void validateFlow(Integer workspaceId, PluginFlow pluginFlow) {
public void validateFlow(Integer workspaceId, PluginFlow pluginFlow, boolean isInner) {
ValidationUtil.dataNotBank(pluginFlow.getName(), "工作流名称不允许为空！");
ValidationUtil.dataNotNull(pluginFlow.getNodeList(), "工作流下属节点不允许为空！");
ValidationUtil.dataNotNull(pluginFlow.getConfig(), "工作流配置不允许为空！");
......@@ -434,12 +437,14 @@ public class ApiFlowServiceImpl implements ApiFlowService {
Flow flow = flowMapper.getByWorkSpaceAndName(workspaceId, pluginFlow.getName());
//判断是否是重发
//是重发
if (!isInner){
if (pluginFlow.isRePublish()){
ValidationUtil.dataNotNull(flow, "工作流【"+ pluginFlow.getName() + "】不存在！");
}else {//不是重发
ValidationUtil.isTrueValidation(null != flow, "工作流【"+ pluginFlow.getName() + "】已存在！");
}
}
}
/**
* 重跑任务
......
byit-myth-admin/src/main/java/com/byit/service/impl/ApiNodeServiceImpl.java
......@@ -145,7 +145,7 @@ public class ApiNodeServiceImpl implements ApiNodeService {
String killResult = HttpUtil.post(killUrl, requestMap);
log.debug("请求结果{}", killResult);
ResponseResult responseResult = JSON.parseObject(killResult, ResponseResult.class);
ValidationUtil.isTrueValidation(!"SUCCESS".equals(responseResult.getResult()), "杀死job失败，错误信息为：" + responseResult.getResult());
ValidationUtil.isTrueValidation(!"SUCCESS".equals(responseResult.getResult()), "杀死job失败，错误信息为：" + responseResult.getMsg());
return true;
}
......
byit-myth-admin/src/main/resources/config/application-test.yml
spring:
datasource:
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://${CM_IP}/myth-job?Unicode=true&characterEncoding=UTF-8&useSSL=true
url: jdbc:mysql://${CM_IP}/${CM_DB}?Unicode=true&characterEncoding=UTF-8&useSSL=true
username: ${CM_USER}
password: ${CM_PWD}
......
byit-myth-admin/src/main/resources/logback-spring.xml
......@@ -58,7 +58,7 @@
</encoder>
</appender>
<logger name="com.byit.thread.helper" level="info" additivity="false">
<logger name="com.byit.thread.helper" level="debug" additivity="false">
<appender-ref ref="console"/>
</logger>
<!--<logger name="com.byit.factory.DaemonScanThreadRunHelperRedisLock" level="debug" additivity="false">
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/factory/DaemonScanThreadRunHelperRedisLock.java
......@@ -47,16 +47,15 @@ public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunH
dateAligned(CYCLE_INTERVAL,threadName);
//调用业务操作
sleepTime = examplesValue.start();
RedissLockUtil.unlock(lockName);
}
}catch (Exception e) {
RedissLockUtil.unlock(lockName);
if(!THREAD_GROUP_STOP){
log.error("---------{}---------",e.getMessage());
}
}finally {
RedissLockUtil.unlock(lockName);
log.debug("------------{},解锁成功，锁名称为{}----------",threadName,lockName);
}
sleepTime = sleepTime==null?examplesValue.UNIVERSAL_WAIT_TIME:sleepTime;
sleepTime = sleepTime==null || sleepTime<=0?examplesValue.UNIVERSAL_WAIT_TIME:sleepTime;
dateAligned(sleepTime,threadName);
}
});
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/flowservice/impl/NodeVerificationImpl.java
......@@ -55,8 +55,9 @@ public class NodeVerificationImpl implements NodeVerification {
public boolean superiorNodeStatus(JobTask thisJobTask) {
//获取该节点的运行标识
String runId = thisJobTask.getRunId();
String reRunId;
if(ScheduleTypeEnum.REPEAT.getCode().equals(thisJobTask.getScheduleType())) {
runId = thisJobTask.getReRunId();
reRunId = thisJobTask.getReRunId();
}
//查询该节点的依赖节点
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskMapper.java
......@@ -29,6 +29,13 @@ public interface JobTaskMapper {
List<JobTask> findAllByRunId(@Param("runId")String runId,@Param("flowId")Integer flowId);
/**
*
* @param runId
* @return
*/
List<JobTask> findByRunId(@Param("runId") String runId);
/**
* 根据id查询
* @param id
* @return
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobTaskService.java
......@@ -27,6 +27,8 @@ public interface JobTaskService {
*/
List<JobTask> findJobTaskByRunId(String runId,Integer flowId);
List<JobTask> findByRunId(String runId);
/**
* 添加单个任务节点
* @param jobTask
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/FlowServiceImpl.java
......@@ -467,7 +467,7 @@ public class FlowServiceImpl implements FlowService {
String killResult = HttpUtil.post(killUrl, requestMap);
log.info("请求结果{}", killResult);
ResponseResult responseResult = JSON.parseObject(killResult, ResponseResult.class);
ValidationUtil.isTrueValidation(!"SUCCESS".equals(responseResult.getResult()), "杀死job失败，错误信息为：" + responseResult.getResult());
ValidationUtil.isTrueValidation(!"SUCCESS".equals(responseResult.getResult()), "杀死job失败，错误信息为：" + responseResult.getMsg());
return true;
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/FlowStatusSnapshootServiceImpl.java
......@@ -32,7 +32,7 @@ public class FlowStatusSnapshootServiceImpl implements FlowStatusSnapshootServic
public void removeAndSaveSnapshoot(List<FlowStatusSnapshoot> allFlow) {
if(CollectionUtil.isEmpty(allFlow)) {
log.info("-----------------无需快照插入-------------------");
log.debug("-----------------无需快照插入-------------------");
return;
}
//获取当天的日期
......@@ -40,6 +40,9 @@ public class FlowStatusSnapshootServiceImpl implements FlowStatusSnapshootServic
String thisHourStr = DateUtil.dateFormat(new Date(), "HH");
String hourBefStr = String.valueOf(Integer.parseInt(thisHourStr)+1);
if (hourBefStr.length() ==1 ) {
hourBefStr = "0"+hourBefStr;
}
//20201212 12:12:12
Date date = DateUtil.strFormatDate(thisDateStr + " " + hourBefStr + ":00:00", "yyyy-MM-dd HH:mm:ss");
long between = cn.hutool.core.date.DateUtil.between(new Date(), date, DateUnit.MINUTE);
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobTaskServiceImpl.java
......@@ -43,6 +43,11 @@ public class JobTaskServiceImpl implements JobTaskService {
return jobTaskMapper.findAllByRunId(runId,flowId);
}
@Override
public List<JobTask> findByRunId(String runId) {
return jobTaskMapper.findByRunId(runId);
}
/**
* 添加一个任务
* @param jobTask 任务实体
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/TaskAndLogServer.java
......@@ -3,13 +3,16 @@ package com.byit.service.mapservice;
import com.byit.model.JobTask;
import org.springframework.transaction.annotation.Transactional;
import java.net.UnknownHostException;
/**
* @author 任务表和日志表的映射业务类，保证是一个事务+原子性
*/
public interface TaskAndLogServer {
/**
* 添加失败日志，删除任务表的任务
* @param jobTask
* @param jobTask 任务节点
* @param isInner 虚节点
*/
void addRunLogAndRemoveTask(JobTask jobTask);
void addRunLogAndRemoveTask(JobTask jobTask,boolean isInner) throws UnknownHostException;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/impl/RunRecordingAndEmailServiceImpl.java
......@@ -71,6 +71,7 @@ public class RunRecordingAndEmailServiceImpl implements RunRecordingAndEmailServ
public void saveEmailAndRunRecording(RunRecording runRecording) {
//这一步是根据flowId和RunId查询对应的节点信息
List<JobTaskRunLogWithBLOBs> jobTaskRunLogByFlowIdAndRunId = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(runRecording.getFlowId(), runRecording.getRunId());
if (runRecording.getFlowNodeCount() == jobTaskRunLogByFlowIdAndRunId.size()) {
String flowName = runRecording.getFlowName();
//String senContentHtml = runMsgHtml(jobTaskRunLogByFlowIdAndRunId, flowName)
Integer workspaceId = runRecording.getWorkspaceId();
......@@ -101,6 +102,8 @@ public class RunRecordingAndEmailServiceImpl implements RunRecordingAndEmailServ
runRecordingService.updateRunRecordingById(runRecording);
}
}
/**
* 根据模板引擎构建html代码
* @param title
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/impl/RunRecordingAndJobTaskServiceImpl.java
......@@ -107,14 +107,14 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask
.flowStatus("2")
.flowTimeout(virFlow.getFlowTimeout())
.dispatchIp(InetAddress.getLocalHost().getHostAddress())
.alarmEmail(virFlow.getAlarmEmail())
.alarmlAction(virFlow.getAlarmlAction())
.alarmEmail(mainRecording.getAlarmEmail())
.alarmlAction(mainRecording.getAlarmlAction())
.priority(virFlow.getPriority())
.triggerTime(equals?jobTask.getTriggerTime():virFlow.getTriggerNextTime())
.principal(virFlow.getPrincipal())
.startTime(new Date())
.flowNodeCount(virFlow.getFlowNodeCount())
.isAlarm(EmailEnum.IS_ALARM_NO.getCode())
.isAlarm(mainRecording.getIsAlarm())
.isInner(FlowPropertyEnum.IS_INNER.getCode())
.failFast(RunRecordingEnum.FAIL_FAST_NO.getCode())
.workspaceId(virFlow.getWorkspaceId())
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/impl/RunRecordingAndLogServiceImpl.java
package com.byit.service.mapservice.impl;
import com.byit.enums.EmailEnum;
import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.RunRecordingEnum;
import com.byit.enums.task.RunResultEnum;
import com.byit.model.JobTaskRunLog;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.NodeVersion;
import com.byit.model.RunRecording;
import com.byit.service.JobTaskRunLogService;
import com.byit.service.JobTaskService;
import com.byit.service.NodeVersionService;
import com.byit.service.RunRecordingService;
import com.byit.model.*;
import com.byit.service.*;
import com.byit.service.mapservice.RunRecordingAndLogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;
......@@ -23,36 +22,42 @@ import java.util.stream.Collectors;
* @author huangfu
*/
@Service
@Slf4j
public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService {
public static final String TRIGGER_MSG = "任务执行超时，被强制执行快速失败！";
private final JobTaskRunLogService jobTaskRunLogService;
private final RunRecordingService runRecordingService;
private final NodeVersionService nodeVersionService;
private final JobTaskService jobTaskService;
private final FlowService flowService;
public RunRecordingAndLogServiceImpl(JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService,
NodeVersionService nodeVersionService, JobTaskService jobTaskService) {
NodeVersionService nodeVersionService, JobTaskService jobTaskService, FlowService flowService) {
this.jobTaskRunLogService = jobTaskRunLogService;
this.runRecordingService = runRecordingService;
this.nodeVersionService = nodeVersionService;
this.jobTaskService = jobTaskService;
this.flowService = flowService;
}
@Override
public void logAndRunRecordingFailFast(RunRecording runRecording) {
String runId = runRecording.getRunId();
Integer flowId = runRecording.getFlowId();
//Integer flowId = runRecording.getFlowId();
//查询该实例对应的所由日志节点
List<JobTaskRunLogWithBLOBs> jobTaskRunLog = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
List<Integer> logNodeId = jobTaskRunLog.stream().map(JobTaskRunLog::getNodeId).collect(Collectors.toList());
List<NodeVersion> allByFlowId = nodeVersionService.findAllByFlowId(flowId);
//List<JobTaskRunLogWithBLOBs> jobTaskRunLog = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
//List<Integer> logNodeId = jobTaskRunLog.stream().map(JobTaskRunLog::getNodeId).collect(Collectors.toList());
//List<NodeVersion> allByFlowId = nodeVersionService.findAllByFlowId(flowId);
//筛选没在日志里面的节点
List<NodeVersion> notLogNode = allByFlowId.stream().filter(nodeVersion -> !(logNodeId.contains(nodeVersion.getNodeId()))).collect(Collectors.toList());
//List<NodeVersion> notLogNode = allByFlowId.stream().filter(nodeVersion -> !(logNodeId.contains(nodeVersion.getNodeId()))).collect(Collectors.toList());
//将这些节点置为失败并将实例也置为失败
//保存错误日志节点
notLogNode.forEach(node ->{
//查询task所有的对应节点
List<JobTask> byRunId = jobTaskService.findByRunId(runId);
byRunId.forEach(task ->{
JobTaskRunLogWithBLOBs log = new JobTaskRunLogWithBLOBs();
BeanUtils.copyProperties(node,log);
BeanUtils.copyProperties(task,log);
log.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
log.setTriggerMsg(TRIGGER_MSG);
log.setRunCode(RunResultEnum.RUN_ERROR.getCode());
......@@ -74,7 +79,67 @@ public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService
runRecording.setFlowRunResult(RunRecordingEnum.RUN_FLOW_FAILURE.getCode());
runRecording.setFailFast(RunRecordingEnum.FAIL_FAST_YES.getCode());
runRecordingService.updateRunRecordingById(runRecording);
//筛选虚节点
List<JobTask> virtualTasks = byRunId.stream().filter(task -> "0".equals(task.getIsVirtual())).collect(Collectors.toList());
virtualTasks.forEach(virtualTask -> {
Integer mapFlowId = virtualTask.getMapFlowId();
List<NodeVersion> allByFlowId = nodeVersionService.findAllByFlowId(mapFlowId);
allByFlowId.forEach(node ->{
JobTaskRunLogWithBLOBs log = new JobTaskRunLogWithBLOBs();
BeanUtils.copyProperties(node,log);
log.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
log.setTriggerMsg(TRIGGER_MSG);
log.setRunCode(RunResultEnum.RUN_ERROR.getCode());
log.setTriggerMsg(TRIGGER_MSG);
Date thisTime = new Date();
log.setStartTime(thisTime);
log.setEndTime(thisTime);
log.setRunId(runId);
log.setFlowName(runRecording.getFlowName());
log.setTriggerTime(thisTime);
jobTaskRunLogService.saveJobTaskRunLog(log);
});
//执行实例的快速失败
Flow virFlow = flowService.findFlowById(mapFlowId);
log.info("-----------【虚节点保存运行记录】--------------");
//保存进运行记录表
RunRecording virtualRunRecording = null;
try {
virtualRunRecording = RunRecording.builder()
.runId(virtualTask.getRunId())
.flowId(mapFlowId)
.flowName(virFlow.getFlowName())
.flowVersionName(virFlow.getVersionName())
.flowStatus("4")
.flowRunResult("2")
.flowTimeout(virFlow.getFlowTimeout())
.dispatchIp(InetAddress.getLocalHost().getHostAddress())
.alarmEmail(virFlow.getAlarmEmail())
.alarmlAction(virFlow.getAlarmlAction())
.priority(virFlow.getPriority())
.triggerTime(runRecording.getTriggerTime())
.principal(virFlow.getPrincipal())
.startTime(new Date())
.flowNodeCount(virFlow.getFlowNodeCount())
.isAlarm(EmailEnum.IS_ALARM_NO.getCode())
.isInner(FlowPropertyEnum.IS_INNER.getCode())
.failFast(RunRecordingEnum.FAIL_FAST_YES.getCode())
.workspaceId(virFlow.getWorkspaceId())
.repeatTime(runRecording.getRepeatTime())
.operator(virtualTask.getOperator())
.scheduleType(virtualTask.getScheduleType())
.build();
} catch (UnknownHostException e) {
e.printStackTrace();
}
runRecordingService.saveRunRecording(virtualRunRecording);
});
//删除task里面的数据
jobTaskService.removeByRunId(runId);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/impl/TaskAndLogServerImpl.java
package com.byit.service.mapservice.impl;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.enums.EmailEnum;
import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.RunRecordingEnum;
import com.byit.job.utils.CronExpression;
import com.byit.job.utils.PlaceholderUtils;
import com.byit.model.JobTask;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.service.JobTaskRunLogService;
import com.byit.service.JobTaskService;
import com.byit.model.*;
import com.byit.service.*;
import com.byit.service.mapservice.TaskAndLogServer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
......@@ -12,7 +15,12 @@ import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.ParseException;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;
/**
* @author huangfu
......@@ -23,14 +31,25 @@ import java.util.Date;
public class TaskAndLogServerImpl implements TaskAndLogServer {
private final JobTaskService jobTaskService;
private final JobTaskRunLogService jobTaskRunLogService;
private final FlowService flowService;
private final RunRecordingService runRecordingService;
private final NodeService nodeService;
/**
* 节点依赖查询操作
*/
private final NodeDependencyService nodeDependencyService;
public TaskAndLogServerImpl(JobTaskService jobTaskService, JobTaskRunLogService jobTaskRunLogService) {
public TaskAndLogServerImpl(JobTaskService jobTaskService, JobTaskRunLogService jobTaskRunLogService, FlowService flowService, RunRecordingService runRecordingService, NodeService nodeService, NodeDependencyService nodeDependencyService) {
this.jobTaskService = jobTaskService;
this.jobTaskRunLogService = jobTaskRunLogService;
this.flowService = flowService;
this.runRecordingService = runRecordingService;
this.nodeService = nodeService;
this.nodeDependencyService = nodeDependencyService;
}
@Override
public void addRunLogAndRemoveTask(JobTask jobTask) {
public void addRunLogAndRemoveTask(JobTask jobTask, boolean isInner) throws UnknownHostException {
Date thisDate = new Date();
//删除任务节点
jobTaskService.removeMythJobTaskById(jobTask.getId());
......@@ -75,5 +94,68 @@ public class TaskAndLogServerImpl implements TaskAndLogServer {
//添加日志节点
jobTaskRunLogService.saveJobTaskRunLog(jobTaskRunLog);
if (isInner) {
RunRecording mainRecording = runRecordingService.findRunRecordingByFlowIdAndRunId(jobTask.getFlowId(), jobTask.getRunId());
log.error("------该节点是虚节点，先将该节点对应的实例保存--------");
//查询该节点对应的工作流
Integer mapFlowId = jobTask.getMapFlowId();
Flow virFlow = flowService.findFlowById(mapFlowId);
//保存进运行记录表 快速失败
RunRecording runRecording = RunRecording.builder()
.runId(jobTask.getRunId())
.flowId(jobTask.getMapFlowId())
.flowName(virFlow.getFlowName())
.flowVersionName(jobTask.getVersionName())
.flowStatus("4")
.flowRunResult("2")
.flowTimeout(virFlow.getFlowTimeout())
.dispatchIp(InetAddress.getLocalHost().getHostAddress())
.alarmEmail(mainRecording.getAlarmEmail())
.alarmlAction(mainRecording.getAlarmlAction())
.priority(virFlow.getPriority())
.triggerTime(jobTask.getTriggerTime())
.principal(virFlow.getPrincipal())
.startTime(new Date())
.flowNodeCount(virFlow.getFlowNodeCount())
.isAlarm(mainRecording.getIsAlarm())
.isInner(FlowPropertyEnum.IS_INNER.getCode())
.failFast(RunRecordingEnum.FAIL_FAST_YES.getCode())
.workspaceId(virFlow.getWorkspaceId())
.build();
runRecordingService.saveRunRecording(runRecording);
log.error("在将该虚节点对应的节点拉取过来");
//获取所有的节点，开始将所有节点保存到任务表
List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(mapFlowId);
List<JobTask> jobTasks = nodeByFlowIdAndVersionName.stream()
.map(node -> {
JobTask task = new JobTask();
List<Integer> dependIdByNodeId = nodeDependencyService.findDependIdByNodeId(jobTask.getNodeId());
if(CollectionUtil.isNotEmpty(dependIdByNodeId)){
String parentIds = StringUtils.join(dependIdByNodeId, ",");
task.setNodeDepend(parentIds);
}
BeanUtils.copyProperties(node, task);
task.setTriggerTime(jobTask.getTriggerTime());
task.setRunId(jobTask.getRunId());
task.setTriggerStatus("1");
task.setFlowName(runRecording.getFlowName());
task.setOperator(jobTask.getOperator());
task.setScheduleType(jobTask.getScheduleType());
return task;
}).collect(Collectors.toList());
jobTaskService.saveJobTasks(jobTasks);
log.info("-----------saveRunRecordingAndTask end【虚节点保存服务】--------------");
if(StringUtils.isNotBlank(virFlow.getFlowCron())){
try {
virFlow.setTriggerNextTime(new CronExpression(virFlow.getFlowCron()).getNextValidTimeAfter(new Date()).getTime());
} catch (ParseException e) {
e.printStackTrace();
}
flowService.updateByIdSelective(virFlow);
}
}
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/JobSnapshotThreadRunHelper.java
......@@ -32,7 +32,7 @@ public class JobSnapshotThreadRunHelper extends BaseThreadRunHelper {
@Value("${myth-job.snapshoot-date}")
private Long snapshootDate;
private final FlowStatusSnapshootMapper flowStatusSnapshootMapper;
private static final String LOCK_NAME = "JobSnapshotThreadRunHelper";
private static final String LOCK_NAME = "job_snapshot_thread_run_helper";
private static final Long SELLP_TIME = 300000L;
private final DataSource dataSource;
private final FlowService flowService;
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/JobTaskThreadRunHelper.java
......@@ -21,6 +21,7 @@ import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
......@@ -105,7 +106,11 @@ public class JobTaskThreadRunHelper extends BaseThreadRunHelper {
}
}catch (SuperiorNodeRunException se) {
//删除这个数据 并且添加到日志
taskAndLogServer.addRunLogAndRemoveTask(jobTask);
try {
taskAndLogServer.addRunLogAndRemoveTask(jobTask,true);
} catch (UnknownHostException e) {
e.printStackTrace();
}
} catch (Exception e) {
e.printStackTrace();
}
......@@ -125,7 +130,11 @@ public class JobTaskThreadRunHelper extends BaseThreadRunHelper {
}
}catch (SuperiorNodeRunException se) {
//删除这个数据 并且添加到日志
taskAndLogServer.addRunLogAndRemoveTask(jobTask);
try {
taskAndLogServer.addRunLogAndRemoveTask(jobTask,false);
} catch (UnknownHostException e) {
e.printStackTrace();
}
}
}
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/MakeUpFlowThreadRunHelper.java
......@@ -59,17 +59,17 @@ public class MakeUpFlowThreadRunHelper extends BaseThreadRunHelper {
Map<Integer, WaitingRecord> nextRunFlow = allByTriggerTime.stream()
.collect(Collectors.toMap(WaitingRecord::getFlowId, Function.identity(), BinaryOperator.minBy(Comparator.comparingInt(WaitingRecord::getWaitOrder))));
log.info("-------筛选后的数据为：{}----------", JSON.toJSONString(nextRunFlow));
log.debug("-------筛选后的数据为：{}----------", JSON.toJSONString(nextRunFlow));
for (WaitingRecord value : nextRunFlow.values()) {
boolean runRecordingIsRunning = runRecordingService.findRunRecordingIsRunning(value.getFlowId());
log.info("-------{}的运行状态为{}",value,runRecordingIsRunning);
log.info("==================================================的运行状态为{}======================================" ,runRecordingIsRunning);
log.debug("-------{}的运行状态为{}",value,runRecordingIsRunning);
log.debug("==================================================的运行状态为{}======================================" ,runRecordingIsRunning);
if(!runRecordingIsRunning){
runRecordingAndJobTaskService.updateRunRecordingAndTask(value);
}
}
log.info("============================com.byit.thread.helper.MakeUpFlowThreadRunHelper.start 执行完成=====================================");
log.debug("============================com.byit.thread.helper.MakeUpFlowThreadRunHelper.start 执行完成=====================================");
return UNIVERSAL_WAIT_TIME;
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/TaskThreadRunHelper.java
......@@ -23,6 +23,7 @@ import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
......@@ -166,7 +167,11 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper {
//如果有失败的节点 就把该节点置为失败
if(CollectionUtil.isNotEmpty(errorJobLog)){
//删除这个数据 并且添加到日志
taskAndLogServer.addRunLogAndRemoveTask(thisJobTask);
try {
taskAndLogServer.addRunLogAndRemoveTask(thisJobTask,false);
} catch (UnknownHostException e) {
e.printStackTrace();
}
}else{
//执行代码
runJobTask(thisJobTask,jobTaskSchedules);
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/TimeoutExampleThreadRunHelper.java
......@@ -10,6 +10,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
......@@ -38,7 +39,7 @@ public class TimeoutExampleThreadRunHelper extends BaseThreadRunHelper {
List<RunRecording> allRunIng = runRecordingService.findAllRunIng(new FlowConditionDto());
List<RunRecording> timeoutRunRecordings = allRunIng.stream().filter(runRecording -> {
long startTime = runRecording.getStartTime().getTime();
long startTime = runRecording.getTriggerTime();
long thisTime = System.currentTimeMillis();
Long flowTimeout = runRecording.getFlowTimeout();
long time = thisTime - startTime;
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskMapper.xml
......@@ -72,6 +72,15 @@
from job_task
where run_id=#{runId,jdbcType=VARCHAR} and flow_Id = #{flowId,jdbcType = INTEGER}
</select>
<select id="findByRunId" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
,
<include refid="Blob_Column_List" />
from job_task
where run_id=#{runId,jdbcType=VARCHAR}
</select>
<!--根据id查询-->
<select id="findJobTaskById" parameterType="java.lang.Integer" resultMap="ResultMapWithBLOBs">
select
......@@ -467,7 +476,7 @@
</foreach>
</delete>
<delete id="deleteByRunId" parameterType="java.lang.Integer">
<delete id="deleteByRunId">
delete from job_task
where run_id = #{runId,jdbcType=VARCHAR}
</delete>
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml
......@@ -126,14 +126,14 @@
where run_id = #{runId,jdbcType=VARCHAR} and flow_id = #{flowId,jdbcType=INTEGER} and schedule_type != 4
</select>
<select id="findRunRecordingById" parameterType="java.lang.Integer" resultMap="BaseResultMap">
<select id="findRunRecordingById" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
where recording_id = #{recordingId,jdbcType=INTEGER} and schedule_type != 4
</select>
<select id="findOnScheduleByFlowId" parameterType="java.lang.Integer" resultMap="BaseResultMap">
<select id="findOnScheduleByFlowId" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
......@@ -141,7 +141,7 @@
and flow_status not in ('1','4') and schedule_type != 4
</select>
<select id="findUnStartByFlowId" parameterType="java.lang.Integer" resultMap="BaseResultMap">
<select id="findUnStartByFlowId" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
......@@ -291,13 +291,13 @@
) flow
</select>
<delete id="deleteById" parameterType="java.lang.Integer">
<delete id="deleteById">
<!-- generated @mbg.generated date: 2019-12-25 -->
delete from run_recording
where recording_id = #{recordingId,jdbcType=INTEGER}
</delete>
<delete id="deleteByRunId" parameterType="java.lang.Integer">
<delete id="deleteByRunId">
delete from run_recording
where run_id = #{runId,jdbcType=INTEGER}
</delete>
......
byit-myth-executor/myth-executor-server/src/main/java/com/byit/node/impl/CommandAndScriptProcessingMachine.java
......@@ -62,7 +62,7 @@ public class CommandAndScriptProcessingMachine implements ProcessingMachine {
}
if (paramAndPlaceholderDto != null) {
//执行命令参数的替换
command = PlaceholderUtils.commandReplace(command,paramAndPlaceholderDto.getParam());
//command = PlaceholderUtils.commandReplace(command,paramAndPlaceholderDto.getParam());
log.debug("-----命令参数替换完成，命令为：[{}]------------",command);
scriptDto.setCommand(command);
}
......
byit-myth-executor/myth-executor-server/src/main/java/com/byit/node/impl/RunScriptProcessingMachine.java
......@@ -6,9 +6,11 @@ import com.byit.node.ProcessingNodeChain;
import com.byit.node.machine.ProcessingMachine;
import com.byit.process.MythJobProcess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
......@@ -36,11 +38,17 @@ public class RunScriptProcessingMachine implements ProcessingMachine {
try {
boolean isRunNode = scriptDto.getRunId().contains("REAL:EXEC:");
List<String> cmdList = Arrays.asList(scriptDto.getCommand().split(" "));
List<String> newCmdList = new ArrayList<>();
cmdList.forEach(cmd ->{
if (StringUtils.isNotBlank(cmd)) {
newCmdList.add(cmd);
}
});
MythJobProcess mythJobProcess;
if(isRunNode) {
mythJobProcess = new MythJobProcess(cmdList, null, null, scriptDto.getLogId(), stringRedisTemplate);
mythJobProcess = new MythJobProcess(newCmdList, null, null, scriptDto.getLogId(), stringRedisTemplate);
}else {
mythJobProcess = new MythJobProcess(cmdList, null, null, scriptDto.getLogId(), null);
mythJobProcess = new MythJobProcess(newCmdList, null, null, scriptDto.getLogId(), null);
}
//保存日志
......
byit-myth-executor/myth-executor-server/src/main/java/com/byit/process/MythJobProcess.java
......@@ -222,16 +222,16 @@ public class MythJobProcess implements Callable<String>{
try {
if (this.isExecuteAsUser) {
final String cmd = String.format("%s %s %s -9 %d", this.executeAsUserBinary, this.effectiveUser, KILL_COMMAND, this.processId);
System.out.println("执行命令" + cmd);
log.debug("执行命令:{}", cmd);
Runtime.getRuntime().exec(cmd);
} else {
final String cmd = String.format("%s -9 %d", KILL_COMMAND, this.processId);
System.out.println("执行命令" + cmd);
log.debug("执行命令:{}", cmd);
Runtime.getRuntime().exec(cmd);
}
return this.completeLatch.await(time, unit);
} catch (final IOException e) {
log.error("尝试杀死失败.", e);
log.error("尝试杀死失败.{}", e.getMessage());
}
}
//如果是杀死将退出值改为-100
......@@ -246,7 +246,7 @@ public class MythJobProcess implements Callable<String>{
* 强制杀死这个过程
*/
public void hardKill() {
log.info("强制杀死这个过程");
log.debug("强制杀死这个过程");
//判断是否已经开始运行
checkStarted();
//判断是否正在执行
......@@ -255,15 +255,15 @@ public class MythJobProcess implements Callable<String>{
try {
if (this.isExecuteAsUser) {
final String cmd = String.format("%s %s %s -9 %d", this.executeAsUserBinary, this.effectiveUser, KILL_COMMAND, this.processId);
System.out.println("执行命令" + cmd);
log.debug("执行命令:{}", cmd);
Runtime.getRuntime().exec(cmd);
} else {
final String cmd = String.format("%s -9 %d", KILL_COMMAND, this.processId);
System.out.println("执行命令" + cmd);
log.debug("执行命令:{}" ,cmd);
Runtime.getRuntime().exec(cmd);
}
} catch (final IOException e) {
log.error("Kill attempt failed.", e);
log.error("Kill attempt failed.:{}", e.getMessage());
}
}
this.process.destroy();
......
byit-myth-gateway/src/main/resources/application-test.yml
......@@ -25,7 +25,7 @@ spring:
maximum-pool-size: 5
minimum-idle: 1
password: ${CM_PWD}
url: jdbc:mysql://${CM_IP}/myth-registry?useUnicode=true&useSSL=true&characterEncoding=utf-8&mysqlEncoding=utf8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=false&autoReconnect=true&failOverReadOnly=false
url: jdbc:mysql://${CM_IP}/${CM_DB}?useUnicode=true&useSSL=true&characterEncoding=utf-8&mysqlEncoding=utf8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=false&autoReconnect=true&failOverReadOnly=false
username: ${CM_USER}
zuul:
route:
......
byit-myth-register/myth-register-server/src/main/java/com/byit/registry/admin/controller/ApiController.java
......@@ -65,7 +65,7 @@ public class ApiController {
try {
registryParamVO = JacksonUtil.readValue(data, RegistryParamVO.class);
} catch (Exception e) {
System.out.print(e);
System.err.println(e.getMessage());
}
// parse param
......@@ -197,7 +197,7 @@ public class ApiController {
//注册数据就是从这里加载的
registryParamVO = JacksonUtil.readValue(data, RegistryParamVO.class);
} catch (Exception e) {
System.out.println(e);
System.err.println(e.getMessage());
}
// parse param
......
byit-myth-register/myth-register-server/src/main/resources/application-test.yml
......@@ -15,7 +15,7 @@ spring:
datasource:
driver-class-name: com.mysql.jdbc.Driver
type: org.apache.tomcat.jdbc.pool.DataSource
url: jdbc:mysql://${CM_IP}/myth-registry?Unicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false
url: jdbc:mysql://${CM_IP}/${CM_DB}?Unicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false
username: ${CM_USER}
password: ${CM_PWD}
tomcat:
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/invoker/route/LoadBalance.java
......@@ -41,11 +41,3 @@ public enum LoadBalance {
return defaultRouter;
}
}
\ No newline at end of file
class test{
public static void main(String[] args) {
for (LoadBalance value : LoadBalance.values( )) {
System.out.println(value );
}
}
}
\ No newline at end of file
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/net/impl/netty/client/NettyClientHandler.java
......@@ -27,10 +27,10 @@ public class NettyClientHandler extends SimpleChannelInboundHandler<RpcResponse>
@Override
protected void channelRead0(ChannelHandlerContext ctx, RpcResponse rpcResponse) throws Exception {
logger.info("------------------客户端接收到响应：{}--------------",rpcResponse);
logger.debug("------------------客户端接收到响应：{}--------------",rpcResponse);
// notify response
rpcInvokerFactory.notifyInvokerFuture(rpcResponse.getRequestId(), rpcResponse);
logger.info("------------------客户端解锁---------------");
logger.debug("------------------客户端解锁---------------");
}
@Override
......
byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/param/defaultparam/DefaultResultCallback.java deleted 100644 → 0
package com.byit.param.defaultparam;
import com.alibaba.fastjson.JSON;
import com.byit.packet.response.PluginRpcResponsePacket;
import com.byit.param.ResultCallback;
import io.netty.channel.ChannelHandlerContext;
/**
* 默认的回调方式
* @author huangfu
*/
public class DefaultResultCallback implements ResultCallback {
@Override
public void resultCallback(ChannelHandlerContext ctx, PluginRpcResponsePacket pluginRpcResponsePacket) {
System.out.println(pluginRpcResponsePacket);
System.out.println(JSON.toJSONString(pluginRpcResponsePacket));
}
}
byit-plugin-core/myth-plugin-rpc-client/src/main/java/com/byit/client/handler/NettyClientHandler.java
......@@ -32,7 +32,6 @@ public class NettyClientHandler extends SimpleChannelInboundHandler<PluginRpcRes
//判断事件是否是心跳事件
if( evt instanceof IdleStateEvent){
pluginConnectClient.send(PluginBeat.PLUGIN_RPC_REQUEST_PACKET);
System.out.println("------客户端发送心跳请求-----");
}else{
super.userEventTriggered(ctx, evt);
}
......
byit-plugin-core/myth-plugin-rpc-client/src/main/java/com/byit/client/handler/NettyPluginConnectionClient.java
......@@ -6,7 +6,6 @@ import com.byit.handler.PacketEncodeHandler;
import com.byit.init.PluginClientInitialization;
import com.byit.packet.request.PluginRpcRequestPacket;
import com.byit.param.PluginBeat;
import com.byit.rpc.remoting.net.params.Beat;
import com.byit.utils.IpUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
......@@ -55,9 +54,8 @@ public class NettyPluginConnectionClient extends PluginConnectClient {
this.channel = bootstrap.connect(ip, port).sync().channel();
// valid
if (!isValidate()) {
System.out.println("------关闭链接");
System.err.println("------关闭链接--------");
close();
return;
}
}
......@@ -65,7 +63,7 @@ public class NettyPluginConnectionClient extends PluginConnectClient {
@Override
public void close() {
if(this.channel !=null && isValidate()){
System.out.println("------关闭链接"+channel.id().asShortText());
System.err.println(String.format("------关闭链接%s------", channel.id().asShortText()));
this.channel.close();
}
......@@ -83,7 +81,7 @@ public class NettyPluginConnectionClient extends PluginConnectClient {
}
@Override
public void send(PluginRpcRequestPacket pluginRpcRequestPacket) throws Exception {
public void send(PluginRpcRequestPacket pluginRpcRequestPacket) {
this.channel.writeAndFlush(pluginRpcRequestPacket);
}
}
byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/factory/RpcSpringPluginServerFactory.java
package com.byit.factory;
import com.byit.callback.RemainingOperationsCallBack;
import com.byit.registry.PluginServiceRegistry;
import com.byit.server.PluginServer;
import com.byit.task.annotations.TaskHandler;
import com.byit.task.handler.interfaces.IJobHandler;
import org.apache.commons.lang3.StringUtils;
......@@ -12,7 +9,6 @@ import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
......@@ -64,7 +60,6 @@ public class RpcSpringPluginServerFactory extends PluginServerFactory implements
}
if(target != null){
TaskHandler annotation = target.getClass().getAnnotation(TaskHandler.class);
System.out.println("注解："+annotation);
String taskName = annotation.taskName();
super.addService(taskName,value);
String expand = annotation.expand();
......@@ -73,7 +68,7 @@ public class RpcSpringPluginServerFactory extends PluginServerFactory implements
}
}
}else{
System.err.println("警告！bean"+key+"不是【com.byit.task.handler.interfaces.IJobHandler】类型！忽略该bean！");
System.out.println(String.format("警告！bean %s 不是【com.byit.task.handler.interfaces.IJobHandler】类型！忽略该bean！",key));
}
});
}
......@@ -93,7 +88,7 @@ public class RpcSpringPluginServerFactory extends PluginServerFactory implements
}
private static Object getCglibProxyTargetObject(Object proxy) throws Exception {
System.out.println(proxy+"-------该对象为cglib代理对象-------");
System.out.println(String.format("%s:-------该对象为cglib代理对象-------", proxy));
Field h = proxy.getClass().getDeclaredField("CGLIB$CALLBACK_0");
h.setAccessible(true);
Object dynamicAdvisedInterceptor = h.get(proxy);
......@@ -108,7 +103,7 @@ public class RpcSpringPluginServerFactory extends PluginServerFactory implements
private static Object getJdkDynamicProxyTargetObject(Object proxy) throws Exception {
System.out.println(proxy+"-------该对象为jdk代理对象-------");
System.out.println(String.format("-------%s,该对象为jdk代理对象-------", proxy));
Field h = proxy.getClass().getSuperclass().getDeclaredField("h");
h.setAccessible(true);
AopProxy aopProxy = (AopProxy) h.get(proxy);
......
byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/server/netty/MainNettyPluginServer.java
......@@ -53,7 +53,7 @@ public class MainNettyPluginServer extends PluginServer {
ChannelFuture closeFuture = channelFuture.channel().closeFuture().sync();
closeFuture.addListener(future ->{
if (future.isSuccess()) {
System.out.println("-----------------");
System.err.println("--------服务关闭---------");
}
});
}catch (Exception e){
......@@ -78,7 +78,7 @@ public class MainNettyPluginServer extends PluginServer {
@Override
public void stop() {
System.out.println("-------------");
System.err.println("--------服务关闭---------");
super.onStop();
}
}
byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/server/netty/handler/NettyPluginServerHandler.java
......@@ -42,7 +42,6 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
@Override
protected void channelRead0(ChannelHandlerContext ctx, PluginRpcRequestPacket msg) {
if(PluginBeat.BEAT_ID.equals(msg.getRequestId())){
System.out.println("------接收到客户端的心跳连接-------");
return;
}
PluginRpcResponsePacket transferPluginRpcResponse = new PluginRpcResponsePacket();
......@@ -114,7 +113,7 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent){
ctx.channel().close();
System.out.println("-------心跳超时，关闭链接-------");
System.err.println("-------心跳超时，关闭链接-------");
} else {
super.userEventTriggered(ctx, evt);
}
......
demo-client/byit-plugin-client/src/main/java/com/byit/controller/TestController.java
package com.byit.controller;
import com.alibaba.fastjson.JSON;
import com.byit.packet.request.PluginRpcRequestPacket;
import com.byit.packet.response.PluginRpcResponsePacket;
import com.byit.param.defaultparam.DefaultResultCallback;
import com.byit.service.TestService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
......
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