Skip to content

B
byit-myth-job
Commit 1cb7eeb1 by huangfusuper
Options

test环境修改 调度代码打磨

parent a16f9910
Showing with 292 additions and 229 deletions
byit-myth-admin/src/main/java/com/byit/service/TestServiceImpl.java
......@@ -13,6 +13,6 @@ public class TestServiceImpl {
public DispatchResponseDto test(){
return scriptExecutorService.runPythonScript(null);
return scriptExecutorService.runScript(null);
}
}
byit-myth-admin/src/main/resources/config/application-pro.yml
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:
......
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
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:
......@@ -32,7 +32,7 @@ mybatis:
myth-rpc:
registry:
address: http://${Eureka_IP}/myth-register
address: http://10.0.120.208:8080/myth-register
env: test
biz: byit-myth-job
logging:
......@@ -62,4 +62,10 @@ fdfs:
tracker-list:
- 10.0.120.216:22122
- 10.0.120.217:22122
- 10.0.120.218:22122
\ No newline at end of file
- 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
byit-myth-core/myth-admin-core/src/main/java/com/byit/enums/EmailEnum.java
......@@ -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;
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/RunScriptServiceImpl.java
......@@ -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);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaTaskJobTask.java
......@@ -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,14 +57,13 @@ 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.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
log.setRunCode(NodeRunStatusPropertyEnum.RUN_FAILURE.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;
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java
......@@ -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,75 +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());
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脚本调度日志结束------------");
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/JavaTaskCallbackThread.java
package com.byit.thread;
import com.byit.dto.executor.JobRunResultDto;
import com.byit.dto.web.ReturnResult;
import com.byit.enums.RunRecordingEnum;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.packet.response.PluginRpcResponsePacket;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/LogCallbackThread.java
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);
}
......
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/web/ReturnResult.java
......@@ -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;
}
......
byit-myth-core/myth-dto-core/src/main/java/com/byit/enums/JobResultEnum.java
......@@ -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
byit-myth-core/myth-dto-core/src/main/java/com/byit/enums/task/RunResultEnum.java 0 → 100644
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;
}
}
byit-myth-core/myth-dto-core/src/main/java/com/byit/enums/task/RunTypeEnum.java 0 → 100644
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;
}
}
byit-myth-executor/myth-executor-api/src/main/java/com/byit/executor/api/ScriptExecutorService.java
......@@ -14,5 +14,5 @@ public interface ScriptExecutorService {
* @param scriptDto 参数
* @return 调用结果
*/
DispatchResponseDto runPythonScript(ScriptDto scriptDto);
DispatchResponseDto runScript(ScriptDto scriptDto);
}
byit-myth-executor/myth-executor-plugin/src/main/java/com/byit/rpc/RunJobServerHandler.java
......@@ -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();
}
......
byit-myth-executor/myth-executor-plugin/src/main/java/com/byit/thread/RunJobThread.java
......@@ -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;
}
......
byit-myth-executor/myth-executor-server/src/main/java/com/byit/service/ScriptExecutorServiceImpl.java
......@@ -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;
......@@ -21,7 +20,6 @@ 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;
......@@ -36,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())
......@@ -72,98 +89,99 @@ 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();
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;
if(null != param){
scriptParamAndPlaceholderDto = JSON.parseObject(param, ScriptParamAndPlaceholderDto.class);
}
if (null != command && command.contains(PlaceholderEnum.BIZ_SCRIPT_FILE.getName())){
String scriptPath = byteArrayToFile(remotePath,scriptParamAndPlaceholderDto);
log.info("-------------脚本所在路径为：{}-------------",scriptPath);
//初始化命令信息
command = PlaceholderUtils.initCommand(command,scriptPath);
}
//开始执行脚本
if(scriptParamAndPlaceholderDto != null){
command = PlaceholderUtils.commandReplace(command,scriptParamAndPlaceholderDto.getParam());
}
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！";
}
byte[] logDataByte = stringToByteArray(logData);
Map<String,String> fileMateData = new HashMap<String,String>(2);
fileMateData.put("filename",scriptDto.getRunId()+scriptDto.getRunId()+".log");
String logPath = "";
try {
jobRunResultDto.setStartTime(new Date());
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = null;
String param = scriptDto.getParam();
String command = scriptDto.getCommand();
String remotePath = scriptDto.getRemotePath();
if(null != param){
scriptParamAndPlaceholderDto = JSON.parseObject(param, ScriptParamAndPlaceholderDto.class);
}
if (null != command && command.contains(PlaceholderEnum.BIZ_SCRIPT_FILE.getName())){
String scriptPath = byteArrayToFile(remotePath,scriptParamAndPlaceholderDto);
log.info("-------------脚本所在路径为：{}-------------",scriptPath);
//初始化命令信息
command = PlaceholderUtils.initCommand(command,scriptPath);
}
//开始执行脚本
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 == NODE_KILL_CODE){
logData = KILL_MESSAGE;
}
byte[] logDataByte = stringToByteArray(logData);
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 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];
......@@ -185,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());
}
......@@ -221,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);
......@@ -230,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());
}
}
byit-myth-executor/myth-executor-server/src/main/resources/application-pro.yml
......@@ -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:
......
byit-myth-executor/myth-executor-server/src/main/resources/application-test.yml
# myth-rpc
myth-rpc:
registry:
address: http://${Eureka_IP}/myth-register
address: http://10.0.120.208:8080/myth-register
biz: byit-myth-job
env: test
remoting:
port: ${Remote_PORT}
port: 7776
logging:
config: classpath:logback.xml
......@@ -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:
......
byit-myth-gateway/src/main/resources/application-test.yml
#路由规则参考byit-myth-rpc中的LoadBalance枚举类中的类型 默认是轮询
#·�ɹ���ο�byit-myth-rpc�е�LoadBalanceö�����е����� Ĭ������ѯ
gateway:
load:
balance: ROUND
......@@ -12,11 +12,11 @@ logging:
springframework: INFO
myth-rpc:
registry:
address: http://${Eureka_IP}/myth-register
address: http://10.0.120.208:8080/myth-register
biz: byit-myth-job
env: test
remoting:
port: ${Remote_PORT}
port: 7776
spring:
datasource:
......@@ -25,9 +25,9 @@ spring:
idle-timeout: 60000
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
username: ${CM_USER}
password: root
url: jdbc:mysql://10.0.120.30:3307/myth-registry?useUnicode=true&useSSL=true&characterEncoding=utf-8&mysqlEncoding=utf8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=false&autoReconnect=true&failOverReadOnly=false
username: root
zuul:
route:
refreshCron: 0/10 * * * * ?
......
byit-myth-register/myth-register-server/src/main/resources/application-test.yml
......@@ -15,9 +15,9 @@ 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
username: ${CM_USER}
password: ${CM_PWD}
url: jdbc:mysql://10.0.120.30:3307/myth-registry?Unicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false
username: root
password: root
tomcat:
max-active: 30
max-wait: 10000
......
byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/server/netty/handler/NettyPluginServerHandler.java
......@@ -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,13 +35,12 @@ 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());
return;
......@@ -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();
}
......
demo-client/byit-demo-client/src/main/java/com/byit/demo/client/controller/IndexController.java
......@@ -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;
}
}
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