Skip to content

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

【日志被重复添加】修改日志被重复添加的BUG

parent fef7d1cd
Showing with 76 additions and 15 deletions
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobFlowCurrent.java
......@@ -2,6 +2,7 @@ package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
......@@ -18,6 +19,8 @@ import java.util.concurrent.TimeUnit;
@NoArgsConstructor
@EqualsAndHashCode
@Builder
@DynamicInsert
@DynamicUpdate
@Entity
@Table(name = "job_flow_current")
@org.hibernate.annotations.Table(appliesTo = "job_flow_current",comment="任务流的当前版本的表，现实表")
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobFlowDependent.java
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.Column;
import javax.persistence.Entity;
......@@ -16,6 +18,8 @@ import java.io.Serializable;
@NoArgsConstructor
@EqualsAndHashCode
@Builder
@DynamicInsert
@DynamicUpdate
@Entity
@Table(name = "job_flow_dependent")
@org.hibernate.annotations.Table(appliesTo = "job_flow_dependent",comment="任务流的关联表，现实表")
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobFlowNodeCurrent.java
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
......@@ -17,6 +19,8 @@ import java.util.Date;
@NoArgsConstructor
@EqualsAndHashCode
@Builder
@DynamicInsert
@DynamicUpdate
@Entity
@Table(name = "job_flow_node_current")
@org.hibernate.annotations.Table(appliesTo = "job_flow_node_current",comment="这是任务流当前版本的节点表")
......@@ -161,7 +165,7 @@ public class MythJobFlowNodeCurrent implements Serializable {
/**
* 当前任务的类型 java python shell sql script
*/
@Column(columnDefinition = "varchar(36) COMMENT '当前任务的类型 java python shell sql script'")
@Column(nullable = false,columnDefinition = "varchar(36) COMMENT '当前任务的类型 java python shell sql script'")
private String jobType;
/**
* 设置任务的优先级 1最低 2最高
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobFlowNodeVersion.java
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
......@@ -18,6 +20,8 @@ import java.util.Date;
@EqualsAndHashCode
@Builder
@Entity
@DynamicInsert
@DynamicUpdate
@Table(name = "job_flow_node_version")
@org.hibernate.annotations.Table(appliesTo = "job_flow_node_version",comment="节点的版本表")
public class MythJobFlowNodeVersion implements Serializable {
......@@ -171,7 +175,7 @@ public class MythJobFlowNodeVersion implements Serializable {
/**
* 当前任务的类型 java python shell sql script
*/
@Column(columnDefinition = "varchar(36) COMMENT '当前任务的类型 java python shell sql script'")
@Column(nullable = false,columnDefinition = "varchar(36) COMMENT '当前任务的类型 java python shell sql script'")
private String jobType;
/**
* 设置任务的优先级 1最低 2最高
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobFlowRunRecording.java
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
......@@ -17,6 +19,8 @@ import java.io.Serializable;
@EqualsAndHashCode
@Builder
@Entity
@DynamicInsert
@DynamicUpdate
@Table(name = "job_flow_run_recording")
@org.hibernate.annotations.Table(appliesTo = "job_flow_run_recording",comment="任务流的执行记录表")
public class MythJobFlowRunRecording implements Serializable {
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobFlowVersion.java
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
......@@ -18,6 +20,8 @@ import java.util.Date;
@EqualsAndHashCode
@Builder
@Entity
@DynamicInsert
@DynamicUpdate
@Table(name = "job_flow_version")
@org.hibernate.annotations.Table(appliesTo = "job_flow_version",comment="这个实体是任务流的版本表")
public class MythJobFlowVersion implements Serializable {
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobTask.java
......@@ -2,6 +2,8 @@ package com.byit.model;
import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
......@@ -19,6 +21,8 @@ import java.util.Date;
@EqualsAndHashCode
@Builder
@Entity
@DynamicInsert
@DynamicUpdate
@Table(name = "job_task")
@org.hibernate.annotations.Table(appliesTo = "job_task",comment="任务表，即将执行的任务全在这个表;工作任务")
public class MythJobTask implements Serializable {
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobTaskRunLog.java
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
......@@ -19,6 +21,8 @@ import java.util.Date;
@Builder
@Entity
@Table(name = "job_task_run_log")
@DynamicInsert
@DynamicUpdate
@org.hibernate.annotations.Table(appliesTo = "job_task_run_log",comment="节点执行日志")
public class MythJobTaskRunLog implements Serializable {
/**
......@@ -81,7 +85,7 @@ public class MythJobTaskRunLog implements Serializable {
/**
* 调度结果 1成功 2失败
*/
@Column(columnDefinition = "char(1) COMMENT '调度结果 1成功 2失败'")
@Column(columnDefinition = "varchar(13) COMMENT '调度结果 1成功 2失败'")
private String triggerCode;
/**
* 调度信息
......@@ -94,9 +98,9 @@ public class MythJobTaskRunLog implements Serializable {
@Column(columnDefinition = "DATE COMMENT '执行时间'")
private Date runTime;
/**
* 运行结果 0未执行完成 1成功 2失败
* 运行结果 0未执行完成
*/
@Column(columnDefinition = "char(1) COMMENT '运行结果'")
@Column(columnDefinition = "varchar(13) COMMENT '运行结果'")
private String runCode;
/**
* 运行结果信息
......@@ -124,5 +128,7 @@ public class MythJobTaskRunLog implements Serializable {
*/
@Column(columnDefinition = "char(1) COMMENT '告警结果,1-告警成功 2-告警失败'")
private String alarmStatus;
@Column(nullable = false,columnDefinition = "varchar(13) COMMENT '任务类型 BEAN ACTUATOR'")
private String jobType;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobTaskSchedule.java
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
......@@ -18,6 +20,8 @@ import java.util.Date;
@EqualsAndHashCode
@Builder
@Entity
@DynamicInsert
@DynamicUpdate
@Table(name = "job_task_schedule")
@org.hibernate.annotations.Table(appliesTo = "job_task_schedule",comment="工作任务排期表，存储七秒内即将执行的任务节点，这个表也是对接工作流的")
public class MythJobTaskSchedule implements Serializable {
......@@ -171,7 +175,7 @@ public class MythJobTaskSchedule implements Serializable {
/**
* 当前任务的类型 java python shell sql script
*/
@Column(columnDefinition = "varchar(36) COMMENT '当前任务的类型 java python shell sql script'")
@Column(nullable = false,columnDefinition = "varchar(36) COMMENT '当前任务的类型 java python shell sql script'")
private String jobType;
/**
* 设置任务的优先级 1最低 2最高
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/MythJobTaskRunLogService.java
......@@ -10,6 +10,12 @@ import com.byit.model.MythJobTaskRunLog;
**/
public interface MythJobTaskRunLogService {
/**
* 查询一条数据 根据ID
* @param id
* @return
*/
MythJobTaskRunLog findMythJobTaskRunLogById(Integer id);
/**
* 既是保存接口又是修改接口
* @return
*/
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/MythJobTaskRunLogServiceImpl.java
......@@ -16,6 +16,12 @@ import org.springframework.stereotype.Service;
public class MythJobTaskRunLogServiceImpl implements MythJobTaskRunLogService {
@Autowired
private MythJobTaskRunLogRepository mythJobTaskRunLogRepository;
@Override
public MythJobTaskRunLog findMythJobTaskRunLogById(Integer id) {
return mythJobTaskRunLogRepository.findOne(id);
}
@Override
public MythJobTaskRunLog save(MythJobTaskRunLog mythJobTaskRunLog) {
return mythJobTaskRunLogRepository.save(mythJobTaskRunLog);
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/JobScheduleHelper.java
......@@ -242,9 +242,9 @@ public class JobScheduleHelper{
if(CollectionUtil.isNotEmpty(mythJobTaskSchedules)){
//循环遍历添加任务
mythJobTaskSchedules.forEach(mythJobTaskSchedule ->{
if ("BEAN".equals(mythJobTaskSchedule.getJobType())) {
Integer logId = saveLog(mythJobTaskSchedule);
mythJobTaskSchedule.setLogId(logId);
if ("BEAN".equals(mythJobTaskSchedule.getJobType())) {
JavaBeanJobTask javaBeanJobTask = new JavaBeanJobTask(mythJobTaskSchedule);
mythJobTaskScheduleService.delete(mythJobTaskSchedule.getNodeId());
WorkRoulette.addJob(javaBeanJobTask,mythJobTaskSchedule.getTriggerNextTime());
......@@ -336,8 +336,10 @@ public class JobScheduleHelper{
mythJobTaskRunLog.setFailedRemainingCount(mythJobTaskSchedule.getFailedRetryCount());
mythJobTaskRunLog.setAlarmEmail(mythJobTaskSchedule.getAlarmEmail());
mythJobTaskRunLog.setMailAction(mythJobTaskSchedule.getMailAction());
mythJobTaskRunLog.setJobType(mythJobTaskSchedule.getJobType());
MythJobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(MythJobTaskRunLogServiceImpl.class);
MythJobTaskRunLog mythJobTaskRunLogSave = mythJobTaskRunLogService.save(mythJobTaskRunLog);
return mythJobTaskRunLogSave.getLogId();
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/util/SourceObj2TargetObjUtil.java
......@@ -33,7 +33,8 @@ public class SourceObj2TargetObjUtil {
mythJobTask.setSourcePrincipal(pluginBeanJobInfo.getAuthor());
mythJobTask.setNodeName(pluginBeanJobInfo.getJobHandelName());
mythJobTask.setTriggerNextTime(System.currentTimeMillis()+20000);
mythJobTask.setDependencyNodes("1");
mythJobTask.setDependencyNodes("0");
mythJobTask.setJobType("BEAN");
return mythJobTask;
}
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/enums/JobResultEnum.java
......@@ -5,9 +5,11 @@ package com.byit.job.enums;
* @author huangfu
*/
public enum JobResultEnum implements IEnum {
SUCCESS("200","任务执行成功"),
FAIL("500","任务执行失败"),
FAIL_TIMEOUT("502","超时错误");
SUCCESS("100200","任务执行成功"),
FAIL("100500","任务执行失败"),
FAIL_TIMEOUT("100502","超时错误"),
DISPATCH_SUCCESS("200200","调度成功"),
DISPATCH_FAIL("200500","调度失败");
private String code;
private String msg;
......
byit-myth-executor/myth-exector-plugin/src/main/java/com/byit/rpc/RunJobServerHandler.java
......@@ -2,6 +2,8 @@ package com.byit.rpc;
import com.alibaba.fastjson.JSON;
import com.byit.job.dto.AdminSenPluginDto;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.enums.JobResultEnum;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
......@@ -38,7 +40,9 @@ public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpReq
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
log.debug("----------------------有请求过来了------------------------");
String heartbeatResponseBody = "FAILURE";
DispatchResponseDto dispatchResponseDto = new DispatchResponseDto();
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_FAIL.getCode());
dispatchResponseDto.setMsg(JobResultEnum.DISPATCH_FAIL.getMsg());
if(req != null){
//解析 调度中心 的数据对象
AdminSenPluginDto adminSenPluginDto = analysisParam(req.content( ));
......@@ -49,13 +53,16 @@ public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpReq
String heartbeat = adminSenPluginDto.getHeartbeat();
if(null == heartbeat){
JOB_TRIGGER_POOL.execute(new RunJobThread(adminSenPluginDto));
heartbeatResponseBody = "SUCCESS";
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_SUCCESS.getCode());
dispatchResponseDto.setMsg(JobResultEnum.DISPATCH_SUCCESS.getMsg());
}else if(PENG.equals(heartbeat)){
log.debug("----------调度平台心跳检测-------------");
heartbeatResponseBody = PONG;
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_SUCCESS.getCode());
dispatchResponseDto.setMsg(JobResultEnum.DISPATCH_SUCCESS.getMsg());
dispatchResponseDto.setContent(PONG);
}
//----------------------------------消息发送-----------------------------------
ByteBuf byteBuf = Unpooled.copiedBuffer(heartbeatResponseBody, CharsetUtil.UTF_8);
ByteBuf byteBuf = Unpooled.copiedBuffer(JSON.toJSONString(dispatchResponseDto), CharsetUtil.UTF_8);
//HTTP响应
FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,byteBuf);
//设置头信息
......
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