Skip to content

B
byit-myth-job
Commit 286ec10e by huangfusuper
Options

【代码重构】修改JPA为MyBatsi

parent d8b4ddb2
Showing with 373 additions and 1861 deletions
byit-myth-admin/src/main/java/com/byit/controller/JobController.java
......@@ -3,8 +3,8 @@ package com.byit.controller;
import com.byit.conf.MythJobAutoConfigure;
import com.byit.job.dto.JobRunResultDto;
import com.byit.job.dto.PluginBeanJobInfo;
import com.byit.model.MythJobTask;
import com.byit.service.MythJobTaskService;
import com.byit.model.JobTask;
import com.byit.service.JobTaskService;
import com.byit.thread.LogCallbackThread;
import com.byit.util.SourceObj2TargetObjUtil;
import org.springframework.beans.factory.annotation.Autowired;
......@@ -22,18 +22,18 @@ import java.util.List;
@RequestMapping("job")
public class JobController {
private final MythJobTaskService mythJobTaskService;
private final JobTaskService jobTaskService;
@Autowired
public JobController(MythJobTaskService mythJobTaskService) {
this.mythJobTaskService = mythJobTaskService;
public JobController(JobTaskService jobTaskService) {
this.jobTaskService = jobTaskService;
}
@PostMapping(value = "addJob")
public String addJob(@RequestBody PluginBeanJobInfo pluginBeanJobInfo){
MythJobTask mythJobTask = SourceObj2TargetObjUtil.pluginBeanJobInfo2MythJobTask(pluginBeanJobInfo);
mythJobTaskService.addMythJobTask(mythJobTask);
JobTask jobTask = SourceObj2TargetObjUtil.pluginBeanJobInfo2JobTask(pluginBeanJobInfo);
jobTaskService.addMythJobTask(jobTask);
return "SUCCESS";
}
......@@ -43,7 +43,7 @@ public class JobController {
}
@GetMapping(value = "getJobTask")
public List<MythJobTask> getJobInfo(){
return mythJobTaskService.findMythJobTaskByTriggerNextTimeLessThanEqual(11);
public List<JobTask> getJobInfo(){
return jobTaskService.findJobTaskByTriggerNextTimeLessThanEqual(11);
}
}
byit-myth-admin/src/main/resources/application.yml
......@@ -4,10 +4,10 @@ spring:
url: jdbc:mysql://10.0.10.118:3306/myth-job?Unicode=true&characterEncoding=UTF-8&useSSL=true
username: root
password: 123456
jpa:
hibernate:
ddl-auto: none
show-sql: false
# jpa:
# hibernate:
# ddl-auto: none
# show-sql: false
mybatis:
mapper-locations: /mapper/*.xml
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/conf/MythJobAutoConfigure.java
......@@ -14,7 +14,7 @@ import java.util.concurrent.TimeUnit;
* @date: 2019/12/9 14:51
**/
@Configuration
@MapperScan("com.byit.dao")
@MapperScan("com.byit.mapper")
public class MythJobAutoConfigure {
/**
* 日志的回调线程池，主要是将执行结果写到库里面
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskMapper.java
package com.byit.mapper;
import com.byit.model.JobTask;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
public interface JobTaskMapper {
int deleteById(Integer nodeId);
import java.util.List;
/**
* 任务表的操作
* @author AutomaticGenerated
*/
@Repository
public interface JobTaskMapper {
/**
* 查询七秒内即将执行的数据
* @param maxNextTime
* @return
*/
List<JobTask> findJobTaskByTriggerNextTimeLessThanEqual(Long maxNextTime);
/**
* 根据id返回
* @param id 节点的id
* @return
*/
JobTask getById(Integer id);
/**
* 插入
* @param record
* @return
*/
int insert(JobTask record);
/**
* 动态插入
* @param record
* @return
*/
int insertSelective(JobTask record);
JobTask getById(Integer nodeId);
/**
* 动态修改根据id
* @param record
* @return
*/
int updateByIdSelective(JobTask record);
int updateByPrimaryKeyWithBLOBs(JobTask record);
/**
* 根据ID删除数据
* @param id
* @return
*/
int deleteById(Integer id);
/**
* 批量删除
* @param mythJobTasks
* @return
*/
int deleteByIds(@Param("mythJobTasks") List<JobTask> mythJobTasks);
int updateById(JobTask record);
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskRunLogMapper.java
package com.byit.mapper;
import com.byit.model.JobTaskRunLog;
import com.byit.model.JobTaskRunLogWithBLOBs;
import org.springframework.stereotype.Repository;
/**
* 对执行后日志的操作
* @author huangfu
*/
@Repository
public interface JobTaskRunLogMapper {
int deleteById(Integer logId);
int insert(JobTaskRunLogWithBLOBs record);
int insert(JobTaskRunLog record);
int insertSelective(JobTaskRunLogWithBLOBs record);
/**
* 动态插入
* @param record
* @return
*/
int insertSelective(JobTaskRunLog record);
JobTaskRunLogWithBLOBs getById(Integer logId);
JobTaskRunLog getById(Integer logId);
int updateByIdSelective(JobTaskRunLogWithBLOBs record);
int updateByIdSelective(JobTaskRunLog record);
int updateByPrimaryKeyWithBLOBs(JobTaskRunLogWithBLOBs record);
int updateByPrimaryKeyWithBLOBs(JobTaskRunLog record);
int updateById(JobTaskRunLog record);
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskScheduleMapper.java
package com.byit.mapper;
import com.byit.model.JobTaskSchedule;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
public interface JobTaskScheduleMapper {
int deleteById(Integer nodeId);
import java.util.List;
/**
* @author huangfu
*/
@Repository
public interface JobTaskScheduleMapper {
/**
* 查询五秒内将要执行的数据
* @param maxNextTime
* @return
*/
List<JobTaskSchedule> findJobTaskScheduleByTriggerNextTimeLessThanEqual(long maxNextTime);
/**
* 插入数据
* @param record
* @return
*/
int insert(JobTaskSchedule record);
/**
* 根据条件插入一条数据
* @param record
* @return
*/
int insertSelective(JobTaskSchedule record);
JobTaskSchedule getById(Integer nodeId);
/**
* 批量插入
* @param jobTaskSchedules
* @return
*/
int insertDataList(@Param("jobTaskSchedules") List<JobTaskSchedule> jobTaskSchedules);
int deleteById(Integer id);
int updateByIdSelective(JobTaskSchedule record);
int updateByPrimaryKeyWithBLOBs(JobTaskSchedule record);
int updateById(JobTaskSchedule record);
JobTaskSchedule getById(Integer id);
int updateByIdSelective(JobTaskSchedule record);
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/JobTask.java deleted 100644 → 0
package com.byit.model;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import java.io.Serializable;
import java.util.Date;
import lombok.Data;
/**
*
*/
@ApiModel
@Data
public class JobTask implements Serializable {
/**
* 当前版本节点主键
*/
@ApiModelProperty("当前版本节点主键")
private Integer nodeId;
/**
* 当前工作流版本的报警邮箱
*/
@ApiModelProperty("当前工作流版本的报警邮箱")
private String alarmEmail;
/**
* 阻塞策略
*/
@ApiModelProperty("阻塞策略")
private String blockStrategy;
/**
* 调度中心端请求插件时的令牌
*/
@ApiModelProperty("调度中心端请求插件时的令牌")
private String callbackToken;
/**
* 该节点的依赖节点
*/
@ApiModelProperty("该节点的依赖节点")
private String dependencyNodes;
/**
* 当前节点的失败重试次数
*/
@ApiModelProperty("当前节点的失败重试次数")
private Integer failedRetryCount;
/**
* 工作流ID
*/
@ApiModelProperty("工作流ID")
private Integer flowId;
/**
* 是否跟随任务流 1跟随 2不跟随
*/
@ApiModelProperty("是否跟随任务流 1跟随 2不跟随")
private String followTaskFlow;
/**
* 插件端请求调度中心的令牌
*/
@ApiModelProperty("插件端请求调度中心的令牌")
private String gatewayToken;
/**
* 当前任务的类型 java python shell sql script
*/
@ApiModelProperty("当前任务的类型 java python shell sql script")
private String jobType;
/**
* 本地节点（插件方） 的节点的名字
*/
@ApiModelProperty("本地节点（插件方） 的节点的名字")
private String localNodeHandlerName;
/**
* 当前工作流版本的告警的时机
*/
@ApiModelProperty("当前工作流版本的告警的时机")
private String mailAction;
/**
* 这个在设置节点执行时间跟随任务流的时候，他是没用的，但是设置不跟随的时候，节点的执行按照他自己的时间执行
*/
@ApiModelProperty("这个在设置节点执行时间跟随任务流的时候，他是没用的，但是设置不跟随的时候，节点的执行按照他自己的时间执行")
private String nodeCron;
/**
* 节点的说明
*/
@ApiModelProperty("节点的说明")
private String nodeDesc;
/**
* 当前节点的名称
*/
@ApiModelProperty("当前节点的名称")
private String nodeName;
/**
* nodeOfFlowId
*/
@ApiModelProperty("nodeOfFlowId")
private Integer nodeOfFlowId;
/**
* 节点的超时时间 -1不超时
*/
@ApiModelProperty("节点的超时时间 -1不超时")
private Long nodeTimeout;
/**
* 节点的类型 node flow
*/
@ApiModelProperty("节点的类型 node flow")
private String nodeType;
/**
* 插件端的url集合
*/
@ApiModelProperty("插件端的url集合")
private String pluginUrls;
/**
* 设置任务的优先级，1最低 2最高
*/
@ApiModelProperty("设置任务的优先级，1最低 2最高")
private String priority;
/**
* 节点的剩余次数
*/
@ApiModelProperty("节点的剩余次数")
private Integer remainingCount;
/**
* 当前节点总共重复次数
*/
@ApiModelProperty("当前节点总共重复次数")
private Integer repeatCount;
/**
* 重试的间隔
*/
@ApiModelProperty("重试的间隔")
private Long retryInterval;
/**
* 路由策略
*/
@ApiModelProperty("路由策略")
private String routingStrategy;
/**
* 运行标识
*/
@ApiModelProperty("运行标识")
private String runId;
/**
* 节点的参数
*/
@ApiModelProperty("节点的参数")
private String runParam;
/**
* 源码备注
*/
@ApiModelProperty("源码备注")
private String runSourceDesc;
/**
* 脚本的文件服务器路径集
*/
@ApiModelProperty("脚本的文件服务器路径集")
private String scriptUrls;
/**
* 源码负责人
*/
@ApiModelProperty("源码负责人")
private String sourcePrincipal;
/**
* 源码的修改时间
*/
@ApiModelProperty("源码的修改时间")
private Date sourceUpdateTime;
/**
* 当前版本的节点的下次执行时间
*/
@ApiModelProperty("当前版本的节点的下次执行时间")
private Long triggerNextTime;
/**
* 调度状态：0-暂停，1-运行
*/
@ApiModelProperty("调度状态：0-暂停，1-运行")
private String triggerStatus;
/**
* 源码
*/
@ApiModelProperty("源码")
private String runSource;
/**
*/
private static final long serialVersionUID = 1L;
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/JobTaskRunLog.java
......@@ -127,6 +127,18 @@ public class JobTaskRunLog implements Serializable {
private Date triggerTime;
/**
* 运行结果信息
*/
@ApiModelProperty("运行结果信息")
private String runMsg;
/**
* 调度信息
*/
@ApiModelProperty("调度信息")
private String triggerMsg;
/**
*/
private static final long serialVersionUID = 1L;
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/JobTaskRunLogWithBLOBs.java deleted 100644 → 0
package com.byit.model;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import java.io.Serializable;
import lombok.Data;
/**
*
*/
@ApiModel
@Data
public class JobTaskRunLogWithBLOBs extends JobTaskRunLog implements Serializable {
/**
* 运行结果信息
*/
@ApiModelProperty("运行结果信息")
private String runMsg;
/**
* 调度信息
*/
@ApiModelProperty("调度信息")
private String triggerMsg;
/**
*/
private static final long serialVersionUID = 1L;
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/JobTaskSchedule.java
......@@ -13,6 +13,12 @@ import lombok.Data;
@Data
public class JobTaskSchedule implements Serializable {
/**
* 排期表的主键
*/
@ApiModelProperty("排期表的主键")
private Integer id;
/**
* 当前版本节点主键
*/
@ApiModelProperty("当前版本节点主键")
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobFlowCurrent.java deleted 100644 → 0
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
import java.util.concurrent.TimeUnit;
/**
* @program: byit-myth-job->MythJobCurrent
* @description: 这个实体是任务流的当前表
* @author: huangfu
* @date: 2019/12/19 16:42
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
@EqualsAndHashCode
@Builder
@DynamicInsert
@DynamicUpdate
@Entity
@Table(name = "job_flow_current")
@org.hibernate.annotations.Table(appliesTo = "job_flow_current",comment="任务流的当前版本的表，现实表")
public class MythJobFlowCurrent implements Serializable {
/**
* 当前版本工作流主键
*/
@Id
@GeneratedValue(strategy=GenerationType.IDENTITY)
@Column(nullable = false,columnDefinition = "int(13) COMMENT '当前版本工作流主键'")
private Integer flowId;
/**
* 当前版本工作流名字
*/
@Column(nullable = false,columnDefinition = "varchar(256) COMMENT '当前版本工作流名字'")
private String flowName;
/**
* 当前版本工作流节点数目
*/
@Column(nullable = false,columnDefinition = "int(13) COMMENT '前版本工作流节点数目'")
private Integer flowNodeCount;
/**
* 当前工作流的介绍
*/
@Column(columnDefinition = "varchar(512) COMMENT '当前工作流的介绍'")
private String flowDesc;
/**
* 当前版本的工作流的超时时间
*/
@Column(columnDefinition = "bigint(20) COMMENT '档期啊版本的工作流的超时时间'")
private Long flowTimeout ;
/**
* 当前版本的工作流的下次执行时间
*/
@Column(columnDefinition = "bigint(20) COMMENT '当前版本的工作流的下次执行时间'")
private Long triggerNextTime;
/**
* 当前工作流版本总共重复次数
*/
@Column(columnDefinition = "int(12) COMMENT '当前工作流版本额重复次数'")
private Integer repeatCount ;
/**
* 当前工作流版本的执行剩余次数
*/
@Column(columnDefinition = "int(12) COMMENT '当前工作流版本的执行剩余次数'")
private Integer remainingCount;
/**
* 当前工作流版本的报警邮箱
*/
@Column(columnDefinition = "varchar(256) COMMENT '当前工作流版本的报警邮箱'")
private String alarmEmail;
/**
* 当前工作流版本的告警的时机
*/
@Column(columnDefinition = "char(1) COMMENT '当前工作流版本的告警的时机'")
private String mailAction;
/**
* 设置任务的优先级 1最低 2最高
*/
@Column(columnDefinition = "char(1) COMMENT '设置任务的优先级，1最低 2最高'")
private String priority;
/**
* 工作空间的id
*/
@Column(columnDefinition = "int(13) COMMENT '工作空间的id'")
private Integer workspaceId;
/**
* 节点是否跟随任务流，1跟随 2不跟随
*/
@Column(columnDefinition = "char(1) COMMENT '节点是否跟随任务流，1跟随 2不跟随'")
private String nodeDateIsFollowFlow;
/**
* 任务流的cron表达式
*/
@Column(columnDefinition = "varchar(256) COMMENT '任务流的cron表达式'")
private String flowCron;
/**
* 执行类型 周期执行1 手动执行2
*/
@Column(columnDefinition = "char(1) COMMENT '执行类型 周期执行1 手动执行2'")
private String execType;
/**
* 是否有下游节点 0 否，1 是
*/
@Column(columnDefinition = "char(1) default '0' COMMENT '是否有下游节点 0 否，1 是'")
private String isHaveDepend;
/**
* 是否是顶级工作流 0 是， 1 是
*/
@Column(columnDefinition = "char(1) default '1' COMMENT '是否是顶级工作流 0 否， 1 是'")
private String isTop;
/**
* 是否是顶级工作流 0 是， 1 是
*/
@Column(columnDefinition = "char(1) default '0' COMMENT '是否是内嵌工作流 0 否， 1 是' ")
private String isInner;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobFlowDependent.java deleted 100644 → 0
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.Table;
import java.io.Serializable;
/**
* @author huangfu
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
@EqualsAndHashCode
@Builder
@DynamicInsert
@DynamicUpdate
@Entity
@Table(name = "job_flow_dependent")
@org.hibernate.annotations.Table(appliesTo = "job_flow_dependent",comment="任务流的关联表，现实表")
public class MythJobFlowDependent implements Serializable {
@Id
@Column(nullable = false,columnDefinition = "int(13) COMMENT '工作流Id'")
private Integer flowId;
@Id
@Column(nullable = false,columnDefinition = "int(13) COMMENT '上游依赖的工作流Id'")
private Integer dependFlowId;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobFlowNodeCurrent.java deleted 100644 → 0
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
import java.util.Date;
/**
* @program: byit-myth-job->MythJobFlowNode
* @description: 这是任务流当前版本的节点表
* @author: huangfu
* @date: 2019/12/19 18:47
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
@EqualsAndHashCode
@Builder
@DynamicInsert
@DynamicUpdate
@Entity
@Table(name = "job_flow_node_current")
@org.hibernate.annotations.Table(appliesTo = "job_flow_node_current",comment="这是任务流当前版本的节点表")
public class MythJobFlowNodeCurrent implements Serializable {
/**
* 当前版本节点主键
*/
@Id
@GeneratedValue(strategy= GenerationType.IDENTITY)
@Column(nullable = false,columnDefinition = "INT(13) COMMENT '当前版本节点主键'")
private Integer nodeId;
/**
* 当前节点的名称
*/
@Column(nullable = false,columnDefinition = "VARCHAR(255) COMMENT '当前节点的名称'")
private String nodeName;
/**
* 本地节点（插件方） 的节点的名字
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '本地节点（插件方） 的节点的名字'")
private String localNodeHandlerName;
/**
* 这个在设置节点执行时间跟随任务流的时候，他是没用的，但是设置不跟随的时候，节点的执行按照他自己的时间执行
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '这个在设置节点执行时间跟随任务流的时候，他是没用的，但是设置不跟随的时候，节点的执行按照他自己的时间执行'")
private String nodeCron;
/**
* 节点的说明
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '节点的说明'")
private String nodeDesc;
/**
* 插件端的url集合
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '插件端的url集合'")
private String pluginUrls;
/**
* 路由策略
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '路由策略'")
private String routingStrategy;
/**
* 阻塞策略
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '阻塞策略'")
private String blockStrategy;
/**
* 调度中心端请求插件时的令牌
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '调度中心端请求插件时的令牌'")
private String callbackToken;
/**
* 插件端请求调度中心的令牌
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '插件端请求调度中心的令牌'")
private String gatewayToken;
/**
* 节点的参数
*/
@Column(columnDefinition = "VARCHAR(256) COMMENT '节点的参数'")
private String runParam;
/**
* 节点的类型 node flow
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '节点的类型 node flow' ")
private String nodeType;
/**
* 工作流ID
*/
@Column(columnDefinition = "INT(13) COMMENT '工作流ID' ")
private Integer flowId;
/**
* 该节点的依赖节点
*/
@Column(columnDefinition = "VARCHAR(256) COMMENT '该节点的依赖节点' ")
private String dependencyNodes;
/**
* 节点的超时时间 -1不超时
*/
@Column(columnDefinition = "BIGINT COMMENT '节点的超时时间 -1不超时' ")
private Long nodeTimeout;
/**
* 脚本的文件服务器路径集
*/
@Column(columnDefinition = "VARCHAR(512) COMMENT '脚本的文件服务器路径集' ")
private String scriptUrls;
/**
* 源码
*/
@Column(columnDefinition = "LONGTEXT COMMENT '源码' ")
private String runSource;
/**
* 源码备注
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '源码备注' " )
private String runSourceDesc;
/**
* 源码的修改时间
*/
@Column(columnDefinition = "DATE COMMENT '源码的修改时间' " )
private Date sourceUpdateTime;
/**
* 源码负责人
*/
@Column(columnDefinition = "varchar(32) COMMENT '源码负责人'")
private String sourcePrincipal;
/**
* 节点的任务流id
*/
@Column(columnDefinition = "int(13) COMMENT 'nodeOfFlowId'")
private Integer nodeOfFlowId;
/**
* 当前节点总共重复次数
*/
@Column(columnDefinition = "int(12) COMMENT '当前节点总共重复次数'")
private Integer repeatCount ;
/**
* 节点的剩余次数
*/
@Column(columnDefinition = "int(12) COMMENT '节点的剩余次数'")
private Integer remainingCount;
/**
* 当前节点的失败重试次数
*/
@Column(columnDefinition = "int(12) COMMENT '当前节点的失败重试次数'")
private Integer failedRetryCount;
/**
* 重试的间隔
*/
@Column(columnDefinition = "bigint(12) COMMENT '重试的间隔'")
private Long retryInterval;
/**
* 是否跟随任务流
*/
@Column(columnDefinition = "char(1) COMMENT '是否跟随任务流 1跟随 2不跟随'")
private String followTaskFlow;
/**
* 当前版本的节点的下次执行时间
*/
@Column(columnDefinition = "bigint(20) COMMENT '当前版本的节点的下次执行时间'")
private Long triggerNextTime;
/**
* 当前任务的类型 java python shell sql script
*/
@Column(nullable = false,columnDefinition = "varchar(36) COMMENT '当前任务的类型 java python shell sql script'")
private String jobType;
/**
* 设置任务的优先级 1最低 2最高
*/
@Column(columnDefinition = "char(1) COMMENT '设置任务的优先级，1最低 2最高'")
private String priority;
/**
* 当前工作流版本的报警邮箱
*/
@Column(columnDefinition = "varchar(256) COMMENT '当前工作流版本的报警邮箱'")
private String alarmEmail;
/**
* 当前工作流版本的告警的时机
*/
@Column(columnDefinition = "char(1) COMMENT '当前工作流版本的告警的时机'")
private String mailAction;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobFlowNodeVersion.java deleted 100644 → 0
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
import java.util.Date;
/**
* @program: byit-myth-job->MythJobFlowNodeVersion
* @description: 节点的版本表
* @author: huangfu
* @date: 2019/12/19 19:53
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
@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 {
/**
* 当前版本节点主键
*/
@Id
@GeneratedValue(strategy= GenerationType.IDENTITY)
@Column(nullable = false,columnDefinition = "INT(13) COMMENT '当前版本节点主键'")
private Integer nodeId;
/**
* 当前节点的名称
*/
@Column(nullable = false,columnDefinition = "VARCHAR(255) COMMENT '当前节点的名称'")
private String nodeName;
/**
* 本地节点（插件方） 的节点的名字
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '本地节点（插件方） 的节点的名字'")
private String localNodeHandlerName;
/**
* 这个在设置节点执行时间跟随任务流的时候，他是没用的，但是设置不跟随的时候，节点的执行按照他自己的时间执行
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '这个在设置节点执行时间跟随任务流的时候，他是没用的，但是设置不跟随的时候，节点的执行按照他自己的时间执行'")
private String nodeCron;
/**
* 节点的说明
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '节点的说明'")
private String nodeDesc;
/**
* 插件端的url集合
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '插件端的url集合'")
private String pluginUrls;
/**
* 路由策略
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '路由策略'")
private String routingStrategy;
/**
* 阻塞策略
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '阻塞策略'")
private String blockStrategy;
/**
* 调度中心端请求插件时的令牌
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '调度中心端请求插件时的令牌'")
private String callbackToken;
/**
* 插件端请求调度中心的令牌
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '插件端请求调度中心的令牌'")
private String gatewayToken;
/**
* 节点的参数
*/
@Column(columnDefinition = "VARCHAR(256) COMMENT '节点的参数'")
private String runParam;
/**
* 节点的类型 node flow
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '节点的类型 node flow' ")
private String nodeType;
/**
* 工作流ID
*/
@Column(columnDefinition = "INT(13) COMMENT '工作流ID' ")
private Integer flowId;
/**
* 该节点的依赖节点
*/
@Column(columnDefinition = "VARCHAR(256) COMMENT '该节点的依赖节点' ")
private String dependencyNodes;
/**
* 节点的超时时间 -1不超时
*/
@Column(columnDefinition = "BIGINT COMMENT '节点的超时时间 -1不超时' ")
private Long nodeTimeout;
/**
* 脚本的文件服务器路径集
*/
@Column(columnDefinition = "VARCHAR(512) COMMENT '脚本的文件服务器路径集' ")
private String scriptUrls;
/**
* 源码
*/
@Column(columnDefinition = "LONGTEXT COMMENT '源码' ")
private String runSource;
/**
* 源码备注
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '源码备注' " )
private String runSourceDesc;
/**
* 源码的修改时间
*/
@Column(columnDefinition = "DATE COMMENT '源码的修改时间' " )
private Date sourceUpdateTime;
/**
* 源码负责人
*/
@Column(columnDefinition = "varchar(32) COMMENT '源码负责人'")
private String sourcePrincipal;
/**
* 节点的任务流id
*/
@Column(columnDefinition = "int(13) COMMENT 'nodeOfFlowId'")
private Integer nodeOfFlowId;
/**
* 当前节点总共重复次数
*/
@Column(columnDefinition = "int(12) COMMENT '当前节点总共重复次数'")
private Integer repeatCount ;
/**
* 当前节点的失败重试次数
*/
@Column(columnDefinition = "int(12) COMMENT '当前节点的失败重试次数'")
private Integer failedRetryCount;
/**
* 重试的间隔
*/
@Column(columnDefinition = "bigint(12) COMMENT '重试的间隔'")
private Long retryInterval;
/**
* 是否跟随任务流
*/
@Column(columnDefinition = "char(1) COMMENT '是否跟随任务流 1跟随 2不跟随'")
private String followTaskFlow;
/**
* 当前版本的节点的下次执行时间
*/
@Column(columnDefinition = "bigint(20) COMMENT '当前版本的节点的下次执行时间'")
private Long triggerNextTime;
/**
* 当前节点的版本标识
*/
@Column(columnDefinition = "char(1) COMMENT '当前节点的版本标识'")
private String versionMark;
/**
* 删除标识
*/
@Column(columnDefinition = "char(1) COMMENT '删除标识 1正常 2 删除'")
private String removeMark;
/**
* 所属工作流的id
*/
@Column(columnDefinition = "int(13) COMMENT '所属工作流的id'")
private Integer flowVersionId;
/**
* 当前任务的类型 java python shell sql script
*/
@Column(nullable = false,columnDefinition = "varchar(36) COMMENT '当前任务的类型 java python shell sql script'")
private String jobType;
/**
* 设置任务的优先级 1最低 2最高
*/
@Column(columnDefinition = "char(1) COMMENT '设置任务的优先级，1最低 2最高'")
private String priority;
/**
* 当前工作流版本的报警邮箱
*/
@Column(columnDefinition = "varchar(256) COMMENT '当前工作流版本的报警邮箱'")
private String alarmEmail;
/**
* 当前工作流版本的告警的时机
*/
@Column(columnDefinition = "char(1) COMMENT '当前工作流版本的告警的时机'")
private String mailAction;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobFlowRunRecording.java deleted 100644 → 0
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
/**
* @program: byit-myth-job->MythJobFlowRunRecording
* @description: 任务流的执行记录表
* @author: huangfu
* @date: 2019/12/19 20:08
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
@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 {
/**
* 当前版本工作流主键
*/
@Id
@GeneratedValue(strategy= GenerationType.IDENTITY)
@Column(nullable = false,columnDefinition = "int(13) COMMENT '记录主键'")
private Integer recordingId;
/**
* 工作流的名字
*/
@Column(nullable = false,columnDefinition = "varchar(255) COMMENT '工作流的名字'")
private String flowName;
/**
* 节点数量
*/
@Column(nullable = false,columnDefinition = "int(13) COMMENT '节点数量'")
private Integer nodeCount;
/**
* 工作流的版本id
*/
@Column(nullable = false,columnDefinition = "int(13) COMMENT '工作流的版本id'")
private Integer flowVersionId;
/**
* 工作流的超时时间
*/
@Column(nullable = false,columnDefinition = "bigint(20) COMMENT '工作流的超时时间'")
private Long flowTimeout;
/**
* 初始值为节点总数，每次一个节点运行成功就将总数-1
*/
@Column(nullable = false,columnDefinition = "int(13) COMMENT '初始值为节点总数，每次一个节点运行完就将总数-1'")
private Integer remainingNode;
/**
* 1 未开始 2运行 3暂停 4结束
*/
@Column(nullable = false,columnDefinition = "char(1) COMMENT '1 未开始 2运行中 3暂停 4成功 5失败'")
private String flowStatus;
/**
* 本次任务的执行时间
*/
@Column(nullable = false,columnDefinition = "bigint(20) COMMENT '本次任务的执行时间'")
private Long triggerTime;
/**
* 执行结果
*/
@Column(columnDefinition = "char(1) COMMENT '执行结果'")
private String flowRunResult;
/**
* 被哪一个调度器加载的
*/
@Column(columnDefinition = "varchar(36) COMMENT '被哪一个调度器加载的'")
private String dispatch_ip;
/**
* 运行标识
*/
@Column(columnDefinition = "varchar(36) COMMENT '运行标识'")
private String RunId;
/**
* 设置任务的优先级 1最低 2最高
*/
@Column(columnDefinition = "char(1) COMMENT '设置任务的优先级，1最低 2最高'")
private String priority;
/**
* 当前工作流版本的报警邮箱
*/
@Column(columnDefinition = "varchar(256) COMMENT '当前工作流版本的报警邮箱'")
private String alarmEmail;
/**
* 当前工作流版本的告警的时机
*/
@Column(columnDefinition = "char(1) COMMENT '当前工作流版本的告警的时机'")
private String mailAction;
/**
* 调度状态：0-暂停，1-运行
*/
@Column(columnDefinition = "char(1) COMMENT '调度状态：0-未开始，1-调度成功， 2-调度失败'")
private String triggerStatus;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobFlowVersion.java deleted 100644 → 0
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
import java.util.Date;
/**
* @program: byit-myth-job->MythJobFlowVersion
* @description: 这个实体是任务流的版本表
* @author: huangfu
* @date: 2019/12/19 18:30
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
@EqualsAndHashCode
@Builder
@Entity
@DynamicInsert
@DynamicUpdate
@Table(name = "job_flow_version")
@org.hibernate.annotations.Table(appliesTo = "job_flow_version",comment="这个实体是任务流的版本表")
public class MythJobFlowVersion implements Serializable {
/**
* 版本工作流主键
*/
@Id
@GeneratedValue(strategy= GenerationType.IDENTITY)
@Column(nullable = false,columnDefinition = "int(13) COMMENT '版本工作流主键'")
private Integer versionId;
/**
* 工作流的id
*/
@Column(nullable = false,columnDefinition = "int(13) COMMENT '工作流主键'")
private Integer flowId;
/**
* 版本工作流名字
*/
@Column(nullable = false,columnDefinition = "varchar(256) COMMENT '版本工作流名字'")
private String flowName;
/**
* 版本工作流节点数目
*/
@Column(nullable = false,columnDefinition = "int(13) COMMENT '前版本工作流节点数目'")
private Integer flowNodeCount;
/**
* 工作流的介绍
*/
@Column(columnDefinition = "varchar(512) COMMENT '工作流的介绍'")
private String flowDesc;
/**
* 版本的工作流的超时时间
*/
@Column(columnDefinition = "bigint(20) COMMENT '档期啊版本的工作流的超时时间'")
private Long flowTimeout ;
/**
* 工作流版本额重复次数
*/
@Column(columnDefinition = "int(12) COMMENT '工作流版本额重复次数'")
private Integer repeatCount ;
/**
* 工作流版本的报警邮箱
*/
@Column(columnDefinition = "varchar(256) COMMENT '工作流版本的报警邮箱'")
private String alarmEmail;
/**
* 工作流版本的告警的时机
*/
@Column(columnDefinition = "char(1) COMMENT '工作流版本的告警的时机'")
private String mailAction;
/**
* 设置任务的优先级 1最低 2最高
*/
@Column(columnDefinition = "char(1) COMMENT '设置任务的优先级，1最低 2最高'")
private String priority;
/**
* 工作空间的id
*/
@Column(columnDefinition = "int(13) COMMENT '工作空间的id'")
private Integer workspaceId;
/**
* 节点是否跟随任务流，1跟随 2不跟随
*/
@Column(columnDefinition = "char(1) COMMENT '节点是否跟随任务流，1跟随 2不跟随'")
private String nodeDateIsFollowFlow;
/**
* 任务流的cron表达式
*/
@Column(columnDefinition = "varchar(256) COMMENT '任务流的cron表达式'")
private String flowCron;
/**
* 执行类型 周期执行1 手动执行2
*/
@Column(columnDefinition = "char(1) COMMENT '执行类型 周期执行1 手动执行2'")
private String execType;
/**
* 删除标志 1正常 2删除
*/
@Column(columnDefinition = "char(1) COMMENT '删除标志 1正常 2删除'")
private String removeMark;
/**
* 当前版本的标志
*/
@Column(columnDefinition = "char(16) COMMENT '当前版本的标志 this'")
private String versionMark;
/**
* 添加时间
*/
@Column(columnDefinition = "DATE COMMENT '添加时间'")
private Date addTime;
/**
* 修改时间
*/
@Column(columnDefinition = "DATE COMMENT '修改时间'")
private Date updateTime;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobTask.java deleted 100644 → 0
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;
import java.util.Date;
/**
* @program: byit-myth-job->MythTaskNode
* @description: 任务表，即将执行的任务全在这个表
* @author: huangfu
* @date: 2019/12/20 9:58
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
@EqualsAndHashCode
@Builder
@Entity
@DynamicInsert
@DynamicUpdate
@Table(name = "job_task")
@org.hibernate.annotations.Table(appliesTo = "job_task",comment="任务表，即将执行的任务全在这个表;工作任务")
public class MythJobTask implements Serializable {
/**
* 当前版本节点主键
*/
@Id
@GeneratedValue(strategy= GenerationType.IDENTITY)
@Column(nullable = false,columnDefinition = "INT(13) COMMENT '当前版本节点主键'")
private Integer nodeId;
/**
* 当前节点的名称
*/
@Column(nullable = false,columnDefinition = "VARCHAR(255) COMMENT '当前节点的名称'")
private String nodeName;
/**
* 本地节点（插件方） 的节点的名字
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '本地节点（插件方） 的节点的名字'")
private String localNodeHandlerName;
/**
* 这个在设置节点执行时间跟随任务流的时候，他是没用的，但是设置不跟随的时候，节点的执行按照他自己的时间执行
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '这个在设置节点执行时间跟随任务流的时候，他是没用的，但是设置不跟随的时候，节点的执行按照他自己的时间执行'")
private String nodeCron;
/**
* 节点的说明
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '节点的说明'")
private String nodeDesc;
/**
* 插件端的url集合
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '插件端的url集合'")
private String pluginUrls;
/**
* 路由策略
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '路由策略'")
private String routingStrategy;
/**
* 阻塞策略
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '阻塞策略'")
private String blockStrategy;
/**
* 调度中心端请求插件时的令牌
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '调度中心端请求插件时的令牌'")
private String callbackToken;
/**
* 插件端请求调度中心的令牌
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '插件端请求调度中心的令牌'")
private String gatewayToken;
/**
* 节点的参数
*/
@Column(columnDefinition = "VARCHAR(256) COMMENT '节点的参数'")
private String runParam;
/**
* 节点的类型 node flow
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '节点的类型 node flow' ")
private String nodeType;
/**
* 工作流ID
*/
@Column(columnDefinition = "INT(13) COMMENT '工作流ID' ")
private Integer flowId;
/**
* 该节点的依赖节点
*/
@Column(columnDefinition = "VARCHAR(256) COMMENT '该节点的依赖节点' ")
private String dependencyNodes;
/**
* 节点的超时时间 -1不超时
*/
@Column(columnDefinition = "BIGINT COMMENT '节点的超时时间 -1不超时' ")
private Long nodeTimeout;
/**
* 脚本的文件服务器路径集
*/
@Column(columnDefinition = "VARCHAR(512) COMMENT '脚本的文件服务器路径集' ")
private String scriptUrls;
/**
* 源码
*/
@Column(columnDefinition = "LONGTEXT COMMENT '源码' ")
private String runSource;
/**
* 源码备注
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '源码备注' " )
private String runSourceDesc;
/**
* 源码的修改时间
*/
@Column(columnDefinition = "DATE COMMENT '源码的修改时间' " )
@JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mms:s")
private Date sourceUpdateTime;
/**
* 源码负责人
*/
@Column(columnDefinition = "varchar(32) COMMENT '源码负责人'")
private String sourcePrincipal;
/**
* 节点的任务流id
*/
@Column(columnDefinition = "int(13) COMMENT 'nodeOfFlowId'")
private Integer nodeOfFlowId;
/**
* 节点的剩余次数
*/
@Column(columnDefinition = "int(12) COMMENT '节点的剩余次数'")
private Integer remainingCount;
/**
* 当前节点总共重复次数
*/
@Column(columnDefinition = "int(12) COMMENT '当前节点总共重复次数'")
private Integer repeatCount ;
/**
* 当前节点的失败重试次数
*/
@Column(columnDefinition = "int(12) COMMENT '当前节点的失败重试次数'")
private Integer failedRetryCount;
/**
* 重试的间隔
*/
@Column(columnDefinition = "bigint(12) COMMENT '重试的间隔'")
private Long retryInterval;
/**
* 是否跟随任务流
*/
@Column(columnDefinition = "char(1) COMMENT '是否跟随任务流 1跟随 2不跟随'")
private String followTaskFlow;
/**
* 当前版本的节点的下次执行时间
*/
@Column(columnDefinition = "bigint(20) COMMENT '当前版本的节点的下次执行时间'")
private Long triggerNextTime;
/**
* 节点的运行标识
*/
@Column(columnDefinition = "varchar(36) COMMENT '运行标识'")
private String runId;
/**
* 调度状态：0-暂停，1-运行
*/
@Column(columnDefinition = "char(1) COMMENT '调度状态：0-暂停，1-运行'")
private String triggerStatus;
/**
* 当前任务的类型 java python shell sql script
*/
@Column(columnDefinition = "varchar(36) COMMENT '当前任务的类型 java python shell sql script'")
private String jobType;
/**
* 设置任务的优先级 1最低 2最高
*/
@Column(columnDefinition = "char(1) COMMENT '设置任务的优先级，1最低 2最高'")
private String priority;
/**
* 当前工作流版本的报警邮箱
*/
@Column(columnDefinition = "varchar(256) COMMENT '当前工作流版本的报警邮箱'")
private String alarmEmail;
/**
* 当前工作流版本的告警的时机
*/
@Column(columnDefinition = "char(1) COMMENT '当前工作流版本的告警的时机'")
private String mailAction;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobTaskRunLog.java deleted 100644 → 0
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
import java.util.Date;
/**
* @program: byit-myth-job->MythJobRunLog
* @description: 节点执行日志
* @author: huangfu
* @date: 2019/12/20 10:21
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
@EqualsAndHashCode
@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 {
/**
* 日志ID
*/
@Id
@GeneratedValue(strategy= GenerationType.IDENTITY)
@Column(nullable = false,columnDefinition = "int(13) COMMENT '日志ID'")
private Integer logId;
/**
* 执行机主键
*/
@Column(columnDefinition = "int(13) COMMENT '执行机主键'")
private Integer jobGroupId;
/**
* 任务流的主键
*/
@Column(columnDefinition = "int(13) COMMENT '任务流主键'")
private Integer jobFlowId;
/**
* 任务流版本的主键
*/
@Column(columnDefinition = "int(13) COMMENT '任务流版本的主键'")
private Integer flowVersionId;
/**
* 所属任务流的名称
*/
@Column(columnDefinition = "varchar(13) COMMENT '所属任务流的名称'")
private String jobFlowName;
/**
* 节点名称
*/
@Column(columnDefinition = "varchar(255) COMMENT '节点名称'")
private String nodeName;
/**
* 节点类型 node type
*/
@Column(columnDefinition = "varchar(36) COMMENT '节点类型 node type'")
private String nodeType;
/**
* 运行方式 1执行机运行，2本地运行
*/
@Column(columnDefinition = "char(1) COMMENT '运行方式 1执行机运行，2本地运行'")
private String runType;
/**
* 插件方节点key
*/
@Column(columnDefinition = "varchar(36) COMMENT '插件方节点key'")
private String localNodeHandlerName;
/**
* 运行参数
*/
@Column(columnDefinition = "varchar(36) COMMENT '运行参数'")
private String runParams;
/**
* 触发时间
*/
@Column(columnDefinition = "DATE COMMENT '触发时间'")
private Date triggerTime;
/**
* 调度结果 1成功 2失败
*/
@Column(columnDefinition = "varchar(13) COMMENT '调度结果 1成功 2失败'")
private String triggerCode;
/**
* 调度信息
*/
@Column(columnDefinition = "text COMMENT '调度信息'")
private String triggerMsg;
/**
* 执行时间
*/
@Column(columnDefinition = "DATE COMMENT '执行时间'")
private Date runTime;
/**
* 运行结果 0未执行完成
*/
@Column(columnDefinition = "varchar(13) COMMENT '运行结果'")
private String runCode;
/**
* 运行结果信息
*/
@Column(columnDefinition = "text COMMENT '运行结果信息'")
private String runMsg;
/**
* 当前节点的失败剩余重试次数 初始阶段为初始的重试次数
*/
@Column(columnDefinition = "int(12) COMMENT '当前节点的失败剩余重试次数 初始阶段为初始的重试次数'")
private Integer failedRemainingCount;
/**
* 当前工作流版本的报警邮箱
*/
@Column(columnDefinition = "varchar(256) COMMENT '当前工作流版本的报警邮箱'")
private String alarmEmail;
/**
* 当前工作流版本的告警的时机
*/
@Column(columnDefinition = "char(1) COMMENT '当前工作流版本的告警的时机'")
private String mailAction;
/**
* 告警结果 1-告警成功 2-告警失败
*/
@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 deleted 100644 → 0
package com.byit.model;
import lombok.*;
import org.hibernate.annotations.DynamicInsert;
import org.hibernate.annotations.DynamicUpdate;
import javax.persistence.*;
import java.io.Serializable;
import java.util.Date;
/**
* @program: byit-myth-job->MythJobTaskSchedule
* @description: 工作任务排期表，存储七秒内即将执行的任务节点,这个表也是对接工作流的
* @author: huangfu
* @date: 2019/12/20 10:16
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
@EqualsAndHashCode
@Builder
@Entity
@DynamicInsert
@DynamicUpdate
@Table(name = "job_task_schedule")
@org.hibernate.annotations.Table(appliesTo = "job_task_schedule",comment="工作任务排期表，存储七秒内即将执行的任务节点，这个表也是对接工作流的")
public class MythJobTaskSchedule implements Serializable {
/**
* 当前版本节点主键
*/
@Id
@GeneratedValue(strategy= GenerationType.IDENTITY)
@Column(nullable = false,columnDefinition = "INT(13) COMMENT '当前版本节点主键'")
private Integer nodeId;
/**
* 当前节点的名称
*/
@Column(nullable = false,columnDefinition = "VARCHAR(255) COMMENT '当前节点的名称'")
private String nodeName;
/**
* 本地节点（插件方） 的节点的名字
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '本地节点（插件方） 的节点的名字'")
private String localNodeHandlerName;
/**
* 这个在设置节点执行时间跟随任务流的时候，他是没用的，但是设置不跟随的时候，节点的执行按照他自己的时间执行
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '这个在设置节点执行时间跟随任务流的时候，他是没用的，但是设置不跟随的时候，节点的执行按照他自己的时间执行'")
private String nodeCron;
/**
* 节点的说明
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '节点的说明'")
private String nodeDesc;
/**
* 插件端的url集合
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '插件端的url集合'")
private String pluginUrls;
/**
* 路由策略
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '路由策略'")
private String routingStrategy;
/**
* 阻塞策略
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '阻塞策略'")
private String blockStrategy;
/**
* 调度中心端请求插件时的令牌
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '调度中心端请求插件时的令牌'")
private String callbackToken;
/**
* 插件端请求调度中心的令牌
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '插件端请求调度中心的令牌'")
private String gatewayToken;
/**
* 节点的参数
*/
@Column(columnDefinition = "VARCHAR(256) COMMENT '节点的参数'")
private String runParam;
/**
* 节点的类型 node flow
*/
@Column(columnDefinition = "VARCHAR(32) COMMENT '节点的类型 node flow' ")
private String nodeType;
/**
* 工作流ID
*/
@Column(columnDefinition = "INT(13) COMMENT '工作流ID' ")
private Integer flowId;
/**
* 该节点的依赖节点
*/
@Column(columnDefinition = "VARCHAR(256) COMMENT '该节点的依赖节点' ")
private String dependencyNodes;
/**
* 节点的超时时间 -1不超时
*/
@Column(columnDefinition = "BIGINT COMMENT '节点的超时时间 -1不超时' ")
private Long nodeTimeout;
/**
* 脚本的文件服务器路径集
*/
@Column(columnDefinition = "VARCHAR(512) COMMENT '脚本的文件服务器路径集' ")
private String scriptUrls;
/**
* 源码
*/
@Column(columnDefinition = "LONGTEXT COMMENT '源码' ")
private String runSource;
/**
* 源码备注
*/
@Column(columnDefinition = "VARCHAR(255) COMMENT '源码备注' " )
private String runSourceDesc;
/**
* 源码的修改时间
*/
@Column(columnDefinition = "DATE COMMENT '源码的修改时间' " )
private Date sourceUpdateTime;
/**
* 源码负责人
*/
@Column(columnDefinition = "varchar(32) COMMENT '源码负责人'")
private String sourcePrincipal;
/**
* 节点的任务流id
*/
@Column(columnDefinition = "int(13) COMMENT 'nodeOfFlowId'")
private Integer nodeOfFlowId;
/**
* 节点的剩余次数
*/
@Column(columnDefinition = "int(12) COMMENT '节点的剩余次数'")
private Integer remainingCount;
/**
* 当前节点总共重复次数
*/
@Column(columnDefinition = "int(12) COMMENT '当前节点总共重复次数'")
private Integer repeatCount ;
/**
* 当前节点的失败重试次数
*/
@Column(columnDefinition = "int(12) COMMENT '当前节点的失败重试次数'")
private Integer failedRetryCount;
/**
* 重试的间隔
*/
@Column(columnDefinition = "bigint(12) COMMENT '重试的间隔'")
private Long retryInterval;
/**
* 是否跟随任务流
*/
@Column(columnDefinition = "char(1) COMMENT '是否跟随任务流 1跟随 2不跟随'")
private String followTaskFlow;
/**
* 当前版本的节点的下次执行时间
*/
@Column(columnDefinition = "bigint(20) COMMENT '当前版本的节点的下次执行时间'")
private Long triggerNextTime;
/**
* 节点的运行标识
*/
@Column(columnDefinition = "varchar(36) COMMENT '运行标识'")
private String runId;
/**
* 调度状态：0-暂停，1-运行
*/
@Column(columnDefinition = "char(1) COMMENT '调度状态：0-暂停，1-运行'")
private String triggerStatus;
/**
* 当前任务的类型 java python shell sql script
*/
@Column(nullable = false,columnDefinition = "varchar(36) COMMENT '当前任务的类型 java python shell sql script'")
private String jobType;
/**
* 设置任务的优先级 1最低 2最高
*/
@Column(columnDefinition = "char(1) COMMENT '设置任务的优先级，1最低 2最高'")
private String priority;
/**
* 当前工作流版本的报警邮箱
*/
@Column(columnDefinition = "varchar(256) COMMENT '当前工作流版本的报警邮箱'")
private String alarmEmail;
/**
* 当前工作流版本的告警的时机
*/
@Column(columnDefinition = "char(1) COMMENT '当前工作流版本的告警的时机'")
private String mailAction;
/**
* 日志ID
*/
@Column(columnDefinition = "int(13) COMMENT '日志ID'")
private Integer logId;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/repository/MythJobTaskRepository.java deleted 100644 → 0
package com.byit.repository;
import com.byit.model.MythJobTask;
import org.springframework.data.jpa.repository.JpaRepository;
import java.util.List;
/**
* @program: byit-myth-job->MythJobTaskRepository
* @description: 对于即将执行人任务进行操作JobTask
* @author: huangfu
* @date: 2019/12/20 15:07
**/
public interface MythJobTaskRepository extends JpaRepository<MythJobTask,Integer> {
/**
* 根据下次执行时间+7000毫秒 查询所有任务节点
* @param maxNextTime
* @return
*/
List<MythJobTask> findMythJobTaskByTriggerNextTimeLessThanEqual(long maxNextTime);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/repository/MythJobTaskRunLogRepository.java deleted 100644 → 0
package com.byit.repository;
import com.byit.model.MythJobTaskRunLog;
import lombok.*;
import org.springframework.data.jpa.repository.JpaRepository;
import javax.persistence.*;
import java.io.Serializable;
import java.util.Date;
/**
* @program: byit-myth-job->MythJobRunLog
* @description: 节点执行日志
* @author: huangfu
* @date: 2019/12/20 10:21
**/
public interface MythJobTaskRunLogRepository extends JpaRepository<MythJobTaskRunLog,Integer> {
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/repository/MythJobTaskScheduleRepository.java deleted 100644 → 0
package com.byit.repository;
import com.byit.model.MythJobTaskSchedule;
import org.springframework.data.jpa.repository.JpaRepository;
import java.util.List;
/**
* @program: byit-myth-job->MythJobTaskSchedule
* @description: 排期表持久化操作
* @author: huangfu
* @date: 2019/12/20 17:27
**/
public interface MythJobTaskScheduleRepository extends JpaRepository<MythJobTaskSchedule,Integer> {
/**
* 根据下次执行时间+5000毫秒 查询所有任务节点
* @param maxNextTime
* @return
*/
List<MythJobTaskSchedule> findMythJobTaskScheduleByTriggerNextTimeLessThanEqual(long maxNextTime);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/MythJobTaskRunLogService.java → byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobTaskRunLogService.java
package com.byit.service;
import com.byit.model.MythJobTaskRunLog;
import com.byit.model.JobTaskRunLog;
/**
* @program: byit-myth-job->MythJobTaskRunLogService
* @program: byit-myth-job->JobTaskRunLogService
* @description: 日志业务表
* @author: huangfu
* @date: 2019/12/20 19:42
**/
public interface MythJobTaskRunLogService {
public interface JobTaskRunLogService {
/**
* 查询一条数据 根据ID
* @param id
* @return
*/
MythJobTaskRunLog findMythJobTaskRunLogById(Integer id);
JobTaskRunLog findJobTaskRunLogById(Integer id);
/**
* 既是保存接口又是修改接口
* @param jobTaskRunLog
* @return
*/
MythJobTaskRunLog save(MythJobTaskRunLog mythJobTaskRunLog);
int saveJobTaskRunLog(JobTaskRunLog jobTaskRunLog);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/MythJobTaskScheduleService.java → byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobTaskScheduleService.java
package com.byit.service;
import com.byit.model.MythJobTaskSchedule;
import com.byit.model.JobTaskSchedule;
import java.util.List;
/**
* @program: byit-myth-job->MythJobTaskScheduleService
* @program: byit-myth-job->JobTaskScheduleService
* @description: 排期表操作
* @author: huangfu
* @date: 2019/12/20 17:31
**/
public interface MythJobTaskScheduleService {
public interface JobTaskScheduleService {
/**
* 根据下次执行时间+5000毫秒 查询所有任务节点
* @param maxNextTime
* @return
*/
List<MythJobTaskSchedule> findMythJobTaskScheduleByTriggerNextTimeLessThanEqual(long maxNextTime);
List<JobTaskSchedule> findJobTaskScheduleByTriggerNextTimeLessThanEqual(long maxNextTime);
/**
* 保存任务表预读的数据
* @param mythJobTaskSchedules
* @return
*/
List<MythJobTaskSchedule> save(List<MythJobTaskSchedule> mythJobTaskSchedules);
int saveAllData(List<JobTaskSchedule> mythJobTaskSchedules);
/**
* 根据id删除
......@@ -31,10 +31,5 @@ public interface MythJobTaskScheduleService {
*/
void delete(Integer id);
/**
* 根据集合删除
* @param mythJobTaskSchedules
*/
void delete(List<MythJobTaskSchedule> mythJobTaskSchedules);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/MythJobTaskService.java → byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobTaskService.java
package com.byit.service;
import com.byit.model.MythJobTask;
import com.byit.model.JobTask;
import java.util.List;
/**
* @program: byit-myth-job->MythJobTaskService
* @program: byit-myth-job->JobTaskService
* @description: 对于任务表的操作
* @author: huangfu
* @date: 2019/12/20 15:13
**/
public interface MythJobTaskService {
public interface JobTaskService {
/**
* 根据下次执行时间+7000毫秒 查询所有任务节点
* @param maxNextTime
* @return
*/
List<MythJobTask> findMythJobTaskByTriggerNextTimeLessThanEqual(long maxNextTime);
List<JobTask> findJobTaskByTriggerNextTimeLessThanEqual(long maxNextTime);
/**
* 添加单个任务节点
* @param mythJobTask
* @param jobTask
*/
void addMythJobTask(MythJobTask mythJobTask);
void addMythJobTask(JobTask jobTask);
/**
* 删除已经被调度的任务根据ID
......@@ -32,7 +32,7 @@ public interface MythJobTaskService {
/**
* 根据集合删除
* @param mythJobTasks
* @param jobTasks
*/
void removeMythJobTaskInIds(List<MythJobTask> mythJobTasks);
void removeMythJobTaskInIds(List<JobTask> jobTasks);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/MythJobTaskRunLogServiceImpl.java → byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobTaskRunLogServiceImpl.java
package com.byit.service.impl;
import com.byit.model.MythJobTaskRunLog;
import com.byit.repository.MythJobTaskRunLogRepository;
import com.byit.service.MythJobTaskRunLogService;
import com.byit.mapper.JobTaskRunLogMapper;
import com.byit.model.JobTaskRunLog;
import com.byit.service.JobTaskRunLogService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
* @program: byit-myth-job->MythJobTaskRunLogServiceImpl
* @program: byit-myth-job->JobTaskRunLogServiceImpl
* @description: 日志业务表
* @author: huangfu
* @date: 2019/12/20 19:43
**/
@Service
public class MythJobTaskRunLogServiceImpl implements MythJobTaskRunLogService {
public class JobTaskRunLogServiceImpl implements JobTaskRunLogService {
@Autowired
private MythJobTaskRunLogRepository mythJobTaskRunLogRepository;
private JobTaskRunLogMapper jobTaskRunLogMapper;
@Override
public MythJobTaskRunLog findMythJobTaskRunLogById(Integer id) {
return mythJobTaskRunLogRepository.findOne(id);
public JobTaskRunLog findJobTaskRunLogById(Integer id) {
return jobTaskRunLogMapper.getById(id);
}
@Override
public MythJobTaskRunLog save(MythJobTaskRunLog mythJobTaskRunLog) {
return mythJobTaskRunLogRepository.save(mythJobTaskRunLog);
public int saveJobTaskRunLog(JobTaskRunLog jobTaskRunLog) {
return jobTaskRunLogMapper.insertSelective(jobTaskRunLog);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobTaskScheduleServiceImpl.java 0 → 100644
package com.byit.service.impl;
import com.byit.mapper.JobTaskScheduleMapper;
import com.byit.model.JobTaskSchedule;
import com.byit.service.JobTaskScheduleService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
* @program: byit-myth-job->JobTaskScheduleServiceImpl
* @description: 排期表实现
* @author: huangfu
* @date: 2019/12/20 17:32
**/
@Service
public class JobTaskScheduleServiceImpl implements JobTaskScheduleService {
@Autowired
private JobTaskScheduleMapper jobTaskScheduleMapper;
/**
* 查询五秒的数据
* @param maxNextTime
* @return
*/
@Override
public List<JobTaskSchedule> findJobTaskScheduleByTriggerNextTimeLessThanEqual(long maxNextTime) {
return jobTaskScheduleMapper.findJobTaskScheduleByTriggerNextTimeLessThanEqual(maxNextTime);
}
/**
* 保存一批数据
* @param jobTaskSchedules
* @return
*/
@Override
public int saveAllData(List<JobTaskSchedule> jobTaskSchedules) {
return jobTaskScheduleMapper.insertDataList(jobTaskSchedules);
}
@Override
public void delete(Integer id) {
jobTaskScheduleMapper.deleteById(id);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobTaskServiceImpl.java 0 → 100644
package com.byit.service.impl;
import com.byit.mapper.JobTaskMapper;
import com.byit.model.JobTask;
import com.byit.service.JobTaskService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
* @program: byit-myth-job->JobTaskService
* @description: 对于任务表的操作
* @author: huangfu
* @date: 2019/12/20 15:13
**/
@Service
public class JobTaskServiceImpl implements JobTaskService {
private final JobTaskMapper jobTaskMapper;
@Autowired
public JobTaskServiceImpl(JobTaskMapper jobTaskMapper) {
this.jobTaskMapper = jobTaskMapper;
}
/**
* 根据下次执行时间查询
* @param maxNextTime 预读数值
* @return 返回的是七秒内将要执行的数据
*/
@Override
public List<JobTask> findJobTaskByTriggerNextTimeLessThanEqual(long maxNextTime) {
return jobTaskMapper.findJobTaskByTriggerNextTimeLessThanEqual(maxNextTime);
}
/**
* 添加一个任务
* @param jobTask 任务实体
*/
@Override
public void addMythJobTask(JobTask jobTask) {
jobTaskMapper.insertSelective(jobTask);
}
/**
* 根据id删除一个任务
* @param id 任务的id
*/
@Override
public void removeMythJobTaskById(Integer id) {
jobTaskMapper.deleteById(id);
}
/**
* 删除多个任务 根据id
* @param jobTasks 任务实体
*/
@Override
public void removeMythJobTaskInIds(List<JobTask> jobTasks) {
jobTaskMapper.deleteByIds(jobTasks);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/MythJobTaskScheduleServiceImpl.java deleted 100644 → 0
package com.byit.service.impl;
import com.byit.model.MythJobTaskSchedule;
import com.byit.repository.MythJobTaskScheduleRepository;
import com.byit.service.MythJobTaskScheduleService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
* @program: byit-myth-job->MythJobTaskScheduleServiceImpl
* @description: 排期表实现
* @author: huangfu
* @date: 2019/12/20 17:32
**/
@Service
public class MythJobTaskScheduleServiceImpl implements MythJobTaskScheduleService {
@Autowired
private MythJobTaskScheduleRepository mythJobTaskScheduleRepository;
@Override
public List<MythJobTaskSchedule> findMythJobTaskScheduleByTriggerNextTimeLessThanEqual(long maxNextTime) {
return mythJobTaskScheduleRepository.findMythJobTaskScheduleByTriggerNextTimeLessThanEqual(maxNextTime);
}
@Override
public List<MythJobTaskSchedule> save(List<MythJobTaskSchedule> mythJobTaskSchedules) {
return mythJobTaskScheduleRepository.save(mythJobTaskSchedules);
}
@Override
public void delete(Integer id) {
mythJobTaskScheduleRepository.delete(id);
}
@Override
public void delete(List<MythJobTaskSchedule> mythJobTaskSchedules) {
mythJobTaskScheduleRepository.delete(mythJobTaskSchedules);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/MythJobTaskServiceImpl.java deleted 100644 → 0
package com.byit.service.impl;
import com.byit.model.MythJobTask;
import com.byit.repository.MythJobTaskRepository;
import com.byit.service.MythJobTaskService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
* @program: byit-myth-job->MythJobTaskService
* @description: 对于任务表的操作
* @author: huangfu
* @date: 2019/12/20 15:13
**/
@Service
public class MythJobTaskServiceImpl implements MythJobTaskService {
@Autowired
private MythJobTaskRepository mythJobTaskRepository;
@Override
public List<MythJobTask> findMythJobTaskByTriggerNextTimeLessThanEqual(long maxNextTime) {
return mythJobTaskRepository.findMythJobTaskByTriggerNextTimeLessThanEqual(maxNextTime);
}
@Override
public void addMythJobTask(MythJobTask mythJobTask) {
mythJobTaskRepository.save(mythJobTask);
}
@Override
public void removeMythJobTaskById(Integer id) {
mythJobTaskRepository.delete(id);
}
@Override
public void removeMythJobTaskInIds(List<MythJobTask> mythJobTasks) {
mythJobTaskRepository.delete(mythJobTasks);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaBeanJobTask.java
......@@ -6,11 +6,11 @@ import com.byit.job.dto.AdminSenPluginDto;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.exceptions.plugin.PluginException;
import com.byit.job.utils.IpUtil;
import com.byit.model.MythJobTaskRunLog;
import com.byit.model.MythJobTaskSchedule;
import com.byit.model.JobTaskRunLog;
import com.byit.model.JobTaskSchedule;
import com.byit.rpc.remoting.invoker.route.LoadBalance;
import com.byit.rpc.remoting.invoker.route.RpcLoadBalance;
import com.byit.service.impl.MythJobTaskRunLogServiceImpl;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.util.SpringUtil;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
......@@ -26,8 +26,8 @@ import java.util.Date;
**/
@Slf4j
public class JavaBeanJobTask implements TimerTask {
private MythJobTaskSchedule mythJobTaskSchedule;
public JavaBeanJobTask(MythJobTaskSchedule mythJobTaskSchedule) {
private JobTaskSchedule mythJobTaskSchedule;
public JavaBeanJobTask(JobTaskSchedule mythJobTaskSchedule) {
this.mythJobTaskSchedule = mythJobTaskSchedule;
}
......@@ -63,19 +63,19 @@ public class JavaBeanJobTask implements TimerTask {
}
}
private void saveLog(MythJobTaskSchedule mythJobTaskSchedule, String url, String result){
private void saveLog(JobTaskSchedule mythJobTaskSchedule, String url, String result){
DispatchResponseDto dispatchResponseDto = JSON.parseObject(result, DispatchResponseDto.class);
MythJobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(MythJobTaskRunLogServiceImpl.class);
MythJobTaskRunLog mythJobTaskRunLog = mythJobTaskRunLogService.findMythJobTaskRunLogById(mythJobTaskSchedule.getLogId());
JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
JobTaskRunLog jobTaskRunLog = new JobTaskRunLog();
jobTaskRunLog.setLogId(mythJobTaskSchedule.getLogId());
//版本id需要查验
//还需要携带版本的名字
mythJobTaskRunLog.setRunType("2");
mythJobTaskRunLog.setLocalNodeHandlerName(mythJobTaskSchedule.getLocalNodeHandlerName());
mythJobTaskRunLog.setTriggerTime(new Date());
mythJobTaskRunLog.setTriggerCode(dispatchResponseDto.getCode());
mythJobTaskRunLog.setTriggerMsg(dispatchResponseDto.getMsg());
mythJobTaskRunLogService.save(mythJobTaskRunLog);
jobTaskRunLog.setRunType("2");
jobTaskRunLog.setLocalNodeHandlerName(mythJobTaskSchedule.getLocalNodeHandlerName());
jobTaskRunLog.setTriggerTime(new Date());
jobTaskRunLog.setTriggerCode(dispatchResponseDto.getCode());
jobTaskRunLog.setTriggerMsg(dispatchResponseDto.getMsg());
mythJobTaskRunLogService.saveJobTaskRunLog(jobTaskRunLog);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/JobScheduleHelper.java
......@@ -2,12 +2,12 @@ package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.job.WorkRoulette;
import com.byit.model.MythJobTask;
import com.byit.model.MythJobTaskRunLog;
import com.byit.model.MythJobTaskSchedule;
import com.byit.service.MythJobTaskScheduleService;
import com.byit.service.MythJobTaskService;
import com.byit.service.impl.MythJobTaskRunLogServiceImpl;
import com.byit.model.JobTask;
import com.byit.model.JobTaskRunLog;
import com.byit.model.JobTaskSchedule;
import com.byit.service.JobTaskScheduleService;
import com.byit.service.JobTaskService;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.task.JavaBeanJobTask;
import com.byit.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;
......@@ -39,9 +39,9 @@ public class JobScheduleHelper{
@Autowired
private MythJobTaskService mythJobTaskService;
private JobTaskService jobTaskService;
@Autowired
private MythJobTaskScheduleService mythJobTaskScheduleService;
private JobTaskScheduleService jobTaskScheduleService;
/**
* 读取任务节点的预读
......@@ -104,27 +104,27 @@ public class JobScheduleHelper{
//行锁已经加上 后续处理
long nowTime = System.currentTimeMillis();
//开始寻找此时 不是暂停状态，而且七秒内即将运行的任务
List<MythJobTask> mythJobTasks = mythJobTaskService.findMythJobTaskByTriggerNextTimeLessThanEqual(nowTime + PRE_READ_MS);
List<JobTask> jobTasks = jobTaskService.findJobTaskByTriggerNextTimeLessThanEqual(nowTime + PRE_READ_MS);
if(CollectionUtil.isNotEmpty(mythJobTasks)){
List<MythJobTaskSchedule> mythJobTaskSchedules = new ArrayList<MythJobTaskSchedule>(15);
mythJobTasks.forEach(mythJobTask -> {
if(CollectionUtil.isNotEmpty(jobTasks)){
List<JobTaskSchedule> jobTaskSchedules = new ArrayList<JobTaskSchedule>(15);
jobTasks.forEach(jobTask -> {
/**
* 需要去检验当前任务的上级节点是否已经执行成功，没有执行，或者处于暂停状态则跳过该任务
* 大概思路，根据任务流id，从任务流执行回溯表查询该任务流的所有节点，查看上级节点是否已经执行成功
* //TODO 需要修改 判断父节点是否执行完毕 注意 父节点是一个集合
*/
if(StringUtils.isNotBlank(mythJobTask.getDependencyNodes())){
log.debug("任务：{}", mythJobTask);
MythJobTaskSchedule mythJobTaskSchedule = new MythJobTaskSchedule();
BeanUtils.copyProperties(mythJobTask,mythJobTaskSchedule);
mythJobTaskSchedules.add(mythJobTaskSchedule);
if(StringUtils.isNotBlank(jobTask.getDependencyNodes())){
log.debug("任务：{}", jobTask);
JobTaskSchedule jobTaskSchedule = new JobTaskSchedule();
BeanUtils.copyProperties(jobTask,jobTaskSchedule);
jobTaskSchedules.add(jobTaskSchedule);
}
});
if(CollectionUtil.isNotEmpty(mythJobTaskSchedules)) {
mythJobTaskScheduleService.save(mythJobTaskSchedules);
mythJobTaskService.removeMythJobTaskInIds(mythJobTasks);
if(CollectionUtil.isNotEmpty(jobTaskSchedules)) {
jobTaskScheduleService.saveAllData(jobTaskSchedules);
jobTaskService.removeMythJobTaskInIds(jobTasks);
}
}else{
......@@ -238,15 +238,15 @@ public class JobScheduleHelper{
//行锁已经加上 后续处理
long nowTime = System.currentTimeMillis();
//查询所有符合条件的任务节点
List<MythJobTaskSchedule> mythJobTaskSchedules = mythJobTaskScheduleService.findMythJobTaskScheduleByTriggerNextTimeLessThanEqual(nowTime + SCHEDULE_READ_MS);
if(CollectionUtil.isNotEmpty(mythJobTaskSchedules)){
List<JobTaskSchedule> jobTaskSchedules = jobTaskScheduleService.findJobTaskScheduleByTriggerNextTimeLessThanEqual(nowTime + SCHEDULE_READ_MS);
if(CollectionUtil.isNotEmpty(jobTaskSchedules)){
//循环遍历添加任务
mythJobTaskSchedules.forEach(mythJobTaskSchedule ->{
jobTaskSchedules.forEach(mythJobTaskSchedule ->{
if ("BEAN".equals(mythJobTaskSchedule.getJobType())) {
Integer logId = saveLog(mythJobTaskSchedule);
mythJobTaskSchedule.setLogId(logId);
JavaBeanJobTask javaBeanJobTask = new JavaBeanJobTask(mythJobTaskSchedule);
mythJobTaskScheduleService.delete(mythJobTaskSchedule.getNodeId());
jobTaskScheduleService.delete(mythJobTaskSchedule.getNodeId());
WorkRoulette.addJob(javaBeanJobTask,mythJobTaskSchedule.getTriggerNextTime());
}
});
......@@ -323,22 +323,22 @@ public class JobScheduleHelper{
this.dataSource = dataSource;
}
private Integer saveLog(MythJobTaskSchedule mythJobTaskSchedule){
MythJobTaskRunLog mythJobTaskRunLog = new MythJobTaskRunLog();
private Integer saveLog(JobTaskSchedule mythJobTaskSchedule){
JobTaskRunLog jobTaskRunLog = new JobTaskRunLog();
mythJobTaskRunLog.setJobFlowId(mythJobTaskSchedule.getFlowId());
jobTaskRunLog.setJobFlowId(mythJobTaskSchedule.getFlowId());
//版本id需要查验
//还需要携带版本的名字
mythJobTaskRunLog.setNodeName(mythJobTaskSchedule.getNodeName());
mythJobTaskRunLog.setNodeType(mythJobTaskSchedule.getNodeType());
mythJobTaskRunLog.setRunParams(mythJobTaskSchedule.getRunParam());
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);
jobTaskRunLog.setNodeName(mythJobTaskSchedule.getNodeName());
jobTaskRunLog.setNodeType(mythJobTaskSchedule.getNodeType());
jobTaskRunLog.setRunParams(mythJobTaskSchedule.getRunParam());
jobTaskRunLog.setFailedRemainingCount(mythJobTaskSchedule.getFailedRetryCount());
jobTaskRunLog.setAlarmEmail(mythJobTaskSchedule.getAlarmEmail());
jobTaskRunLog.setMailAction(mythJobTaskSchedule.getMailAction());
jobTaskRunLog.setJobType(mythJobTaskSchedule.getJobType());
JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
mythJobTaskRunLogService.saveJobTaskRunLog(jobTaskRunLog);
return mythJobTaskRunLogSave.getLogId();
return jobTaskRunLog.getLogId();
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/LogCallbackThread.java
package com.byit.thread;
import com.byit.job.dto.JobRunResultDto;
import com.byit.model.MythJobTaskRunLog;
import com.byit.service.impl.MythJobTaskRunLogServiceImpl;
import com.byit.model.JobTaskRunLog;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;
......@@ -24,11 +24,12 @@ public class LogCallbackThread implements Runnable {
@Override
public void run() {
log.debug("--------------------任务执行完成---------------------");
MythJobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(MythJobTaskRunLogServiceImpl.class);
MythJobTaskRunLog mythJobTaskRunLog = mythJobTaskRunLogService.findMythJobTaskRunLogById(jobRunResultDto.getLogId( ));
mythJobTaskRunLog.setRunTime(jobRunResultDto.getEndTime());
mythJobTaskRunLog.setRunCode(jobRunResultDto.getReturnResult().getCode());
mythJobTaskRunLog.setRunMsg(jobRunResultDto.getReturnResult().getMsg());
mythJobTaskRunLogService.save(mythJobTaskRunLog);
JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
JobTaskRunLog jobTaskRunLog = new JobTaskRunLog();
jobTaskRunLog.setLogId(jobRunResultDto.getLogId());
jobTaskRunLog.setRunTime(jobRunResultDto.getEndTime());
jobTaskRunLog.setRunCode(jobRunResultDto.getReturnResult().getCode());
jobTaskRunLog.setRunMsg(jobRunResultDto.getReturnResult().getMsg());
mythJobTaskRunLogService.saveJobTaskRunLog(jobTaskRunLog);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/util/SourceObj2TargetObjUtil.java
package com.byit.util;
import com.byit.job.dto.PluginBeanJobInfo;
import com.byit.model.MythJobTask;
import java.util.Date;
import java.util.UUID;
import com.byit.model.JobTask;
/**
* @program: byit-myth-job->SourceObj2TargetObjUtil
......@@ -19,22 +16,23 @@ public class SourceObj2TargetObjUtil {
* @param pluginBeanJobInfo
* @return
*/
public static MythJobTask pluginBeanJobInfo2MythJobTask(PluginBeanJobInfo pluginBeanJobInfo){
MythJobTask mythJobTask = new MythJobTask();
mythJobTask.setLocalNodeHandlerName(pluginBeanJobInfo.getJobHandelName());
mythJobTask.setPluginUrls(pluginBeanJobInfo.getUrl());
mythJobTask.setNodeCron(pluginBeanJobInfo.getMythCron());
mythJobTask.setRoutingStrategy(pluginBeanJobInfo.getRoutingStrategy());
mythJobTask.setBlockStrategy(pluginBeanJobInfo.getBlockingStrategy());
mythJobTask.setCallbackToken(pluginBeanJobInfo.getCallbackToken());
mythJobTask.setGatewayToken(pluginBeanJobInfo.getGatewayToken());
mythJobTask.setRunParam(pluginBeanJobInfo.getParam());
mythJobTask.setAlarmEmail(pluginBeanJobInfo.getAlarmEmail());
mythJobTask.setSourcePrincipal(pluginBeanJobInfo.getAuthor());
mythJobTask.setNodeName(pluginBeanJobInfo.getJobHandelName());
mythJobTask.setTriggerNextTime(System.currentTimeMillis()+20000);
mythJobTask.setDependencyNodes("0");
mythJobTask.setJobType("BEAN");
return mythJobTask;
public static JobTask pluginBeanJobInfo2JobTask(PluginBeanJobInfo pluginBeanJobInfo){
JobTask jobTask = new JobTask();
jobTask.setNodeId(1);
jobTask.setLocalNodeHandlerName(pluginBeanJobInfo.getJobHandelName());
jobTask.setPluginUrls(pluginBeanJobInfo.getUrl());
jobTask.setNodeCron(pluginBeanJobInfo.getMythCron());
jobTask.setRoutingStrategy(pluginBeanJobInfo.getRoutingStrategy());
jobTask.setBlockStrategy(pluginBeanJobInfo.getBlockingStrategy());
jobTask.setCallbackToken(pluginBeanJobInfo.getCallbackToken());
jobTask.setGatewayToken(pluginBeanJobInfo.getGatewayToken());
jobTask.setRunParam(pluginBeanJobInfo.getParam());
jobTask.setAlarmEmail(pluginBeanJobInfo.getAlarmEmail());
jobTask.setSourcePrincipal(pluginBeanJobInfo.getAuthor());
jobTask.setNodeName(pluginBeanJobInfo.getJobHandelName());
jobTask.setTriggerNextTime(System.currentTimeMillis()+20000);
jobTask.setDependencyNodes("0");
jobTask.setJobType("BEAN");
return jobTask;
}
}
byit-myth-core/myth-admin-core/src/main/resources/Generator-config.xml
......@@ -55,14 +55,6 @@
type="XMLMAPPER">
<property name="enableSubPackages" value="false"/>
</javaClientGenerator>
<table tableName="job_flow_current" domainObjectName="JobFlow" />
<table tableName="job_flow_dependent" domainObjectName="JobFlowDependent" />
<table tableName="job_flow_node_current" domainObjectName="JobFlowNode" />
<table tableName="job_flow_node_version" domainObjectName="JobFlowNodeVersion" />
<table tableName="job_flow_run_recording" domainObjectName="JobFlowRunRecording" />
<table tableName="job_flow_version" domainObjectName="JobFlowVersion" />
<table tableName="job_task" domainObjectName="JobTask" />
<table tableName="job_task_run_log" domainObjectName="JobTaskRunLog" />
<table tableName="job_task_schedule" domainObjectName="JobTaskSchedule" />
</context>
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.byit.mapper.JobTaskRunLogMapper">
<resultMap id="BaseResultMap" type="com.byit.model.JobTaskRunLog">
<!-- generated @mbg.generated date: 2019-12-23 -->
<id column="log_id" jdbcType="INTEGER" property="logId" />
<result column="alarm_email" jdbcType="VARCHAR" property="alarmEmail" />
<result column="alarm_status" jdbcType="CHAR" property="alarmStatus" />
......@@ -22,38 +22,29 @@
<result column="run_type" jdbcType="CHAR" property="runType" />
<result column="trigger_code" jdbcType="VARCHAR" property="triggerCode" />
<result column="trigger_time" jdbcType="DATE" property="triggerTime" />
</resultMap>
<resultMap extends="BaseResultMap" id="ResultMapWithBLOBs" type="com.byit.model.JobTaskRunLogWithBLOBs">
<!-- generated @mbg.generated date: 2019-12-23 -->
<result column="run_msg" jdbcType="LONGVARCHAR" property="runMsg" />
<result column="trigger_msg" jdbcType="LONGVARCHAR" property="triggerMsg" />
</resultMap>
<sql id="Base_Column_List">
<!-- generated @mbg.generated date: 2019-12-23 -->
log_id, alarm_email, alarm_status, failed_remaining_count, flow_version_id, job_flow_id,
job_flow_name, job_group_id, job_type, local_node_handler_name, mail_action, node_name,
node_type, run_code, run_params, run_time, run_type, trigger_code, trigger_time
</sql>
<sql id="Blob_Column_List">
<!-- generated @mbg.generated date: 2019-12-23 -->
run_msg, trigger_msg
node_type, run_code, run_params, run_time, run_type, trigger_code, trigger_time,run_msg, trigger_msg
</sql>
<select id="getById" parameterType="java.lang.Integer" resultMap="ResultMapWithBLOBs">
<!-- generated @mbg.generated date: 2019-12-23 -->
<select id="getById" parameterType="java.lang.Integer" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
,
<include refid="Blob_Column_List" />
from job_task_run_log
where log_id = #{logId,jdbcType=INTEGER}
</select>
<delete id="deleteById" parameterType="java.lang.Integer">
<!-- generated @mbg.generated date: 2019-12-23 -->
delete from job_task_run_log
where log_id = #{logId,jdbcType=INTEGER}
</delete>
<insert id="insert" parameterType="com.byit.model.JobTaskRunLogWithBLOBs">
<!-- generated @mbg.generated date: 2019-12-23 -->
<insert id="insert" parameterType="com.byit.model.JobTaskRunLog">
insert into job_task_run_log (log_id, alarm_email, alarm_status,
failed_remaining_count, flow_version_id, job_flow_id,
job_flow_name, job_group_id, job_type,
......@@ -71,8 +62,8 @@
#{triggerTime,jdbcType=DATE}, #{runMsg,jdbcType=LONGVARCHAR}, #{triggerMsg,jdbcType=LONGVARCHAR}
)
</insert>
<insert id="insertSelective" parameterType="com.byit.model.JobTaskRunLogWithBLOBs">
<!-- generated @mbg.generated date: 2019-12-23 -->
<insert id="insertSelective" parameterType="com.byit.model.JobTaskRunLog" useGeneratedKeys="true" keyProperty="id" keyColumn="log_id">
insert into job_task_run_log
<trim prefix="(" suffix=")" suffixOverrides=",">
<if test="logId != null">
......@@ -205,7 +196,8 @@
</if>
</trim>
</insert>
<update id="updateByIdSelective" parameterType="com.byit.model.JobTaskRunLogWithBLOBs">
<update id="updateByIdSelective" parameterType="com.byit.model.JobTaskRunLog">
<!-- generated @mbg.generated date: 2019-12-23 -->
update job_task_run_log
<set>
......@@ -272,7 +264,8 @@
</set>
where log_id = #{logId,jdbcType=INTEGER}
</update>
<update id="updateByPrimaryKeyWithBLOBs" parameterType="com.byit.model.JobTaskRunLogWithBLOBs">
<update id="updateByPrimaryKeyWithBLOBs" parameterType="com.byit.model.JobTaskRunLog">
<!-- generated @mbg.generated date: 2019-12-23 -->
update job_task_run_log
set alarm_email = #{alarmEmail,jdbcType=VARCHAR},
......@@ -297,6 +290,7 @@
trigger_msg = #{triggerMsg,jdbcType=LONGVARCHAR}
where log_id = #{logId,jdbcType=INTEGER}
</update>
<update id="updateById" parameterType="com.byit.model.JobTaskRunLog">
<!-- generated @mbg.generated date: 2019-12-23 -->
update job_task_run_log
......
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