Skip to content

B
byit-myth-job
Commit fef7d1cd by huangfusuper
Options

【实现流程】任务表到排期表到任务轮；调用插件，结果回调！ 

【遗留BUG】日志表被重复的插入
parent 574de89e
Showing with 316 additions and 1169 deletions
byit-myth-admin/src/main/java/com/byit/controller/JobController.java
...@@ -3,12 +3,10 @@ package com.byit.controller; ...@@ -3,12 +3,10 @@ package com.byit.controller;
import com.byit.conf.MythJobAutoConfigure; import com.byit.conf.MythJobAutoConfigure;
import com.byit.job.dto.JobRunResultDto; import com.byit.job.dto.JobRunResultDto;
import com.byit.job.dto.PluginBeanJobInfo; import com.byit.job.dto.PluginBeanJobInfo;
import com.byit.job.model.MythJobReadAhead;
import com.byit.job.utils.SourceObj2TargetObjUtil;
import com.byit.model.MythJobTask; import com.byit.model.MythJobTask;
import com.byit.service.JobReadAheadService;
import com.byit.service.MythJobTaskService; import com.byit.service.MythJobTaskService;
import com.byit.thread.LogCallbackThread; import com.byit.thread.LogCallbackThread;
import com.byit.util.SourceObj2TargetObjUtil;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.*;
...@@ -23,15 +21,19 @@ import java.util.List; ...@@ -23,15 +21,19 @@ import java.util.List;
@RestController @RestController
@RequestMapping("job") @RequestMapping("job")
public class JobController { public class JobController {
private final MythJobTaskService mythJobTaskService;
@Autowired @Autowired
private JobReadAheadService jobReadAheadService; public JobController(MythJobTaskService mythJobTaskService) {
@Autowired this.mythJobTaskService = mythJobTaskService;
private MythJobTaskService mythJobTaskService; }
@PostMapping(value = "addJob") @PostMapping(value = "addJob")
public String addJob(@RequestBody PluginBeanJobInfo pluginBeanJobInfo){ public String addJob(@RequestBody PluginBeanJobInfo pluginBeanJobInfo){
MythJobReadAhead mythJobReadAhead = SourceObj2TargetObjUtil.pluginBeanJobInfo2MythJobReadAhead(pluginBeanJobInfo); MythJobTask mythJobTask = SourceObj2TargetObjUtil.pluginBeanJobInfo2MythJobTask(pluginBeanJobInfo);
jobReadAheadService.addOneMythJobReadAhead(mythJobReadAhead); mythJobTaskService.addMythJobTask(mythJobTask);
return "SUCCESS"; return "SUCCESS";
} }
......
byit-myth-admin/src/main/resources/application.yml
...@@ -7,7 +7,7 @@ spring: ...@@ -7,7 +7,7 @@ spring:
jpa: jpa:
hibernate: hibernate:
ddl-auto: update ddl-auto: update
show-sql: true show-sql: false
mybatis: mybatis:
mapper-locations: /mapper/*.xml mapper-locations: /mapper/*.xml
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/dao/JobFlightScheduleMapper.java deleted 100644 → 0
package com.byit.dao;
import com.byit.job.model.MythJobFlightSchedule;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;
/**
* @program: byit-myth-job->JobFlightScheduleMapper
* @description: 任务排期表 这个就是预读五秒的数据，要添加进任务调度轮的数据
* @author: huangfu
* @date: 2019/12/10 15:06
**/
@Repository
public interface JobFlightScheduleMapper {
/**
* 查询预读五秒的数据
* @param maxNextTime
* @return
*/
List<MythJobFlightSchedule> findMythJobFlightScheduleByTriggerNextTime(@Param("maxNextTime")long maxNextTime);
/**
* 批量向排期表添加任务数据
* @param mythJobFlightSchedules
*/
void addJobFlightSchedules(@Param("mythJobFlightSchedule") List<MythJobFlightSchedule> mythJobFlightSchedules);
/**
* 删除已经加载的节点
* @param id
*/
void deleteMythJobFlightScheduleById(@Param("id")String id);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/dao/JobInfoMapper.java deleted 100644 → 0
package com.byit.dao;
import com.byit.job.model.MythJobReadAhead;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;
/**
* @program: byit-myth-job->JobInfoMapper
* @description: 对于任务节点的操作
* @author: huangfu
* @date: 2019/12/9 14:47
**/
@Repository
public interface JobInfoMapper {
/**
* 根据下次执行时间+7000毫秒 查询所有任务节点
* @param maxNextTime
* @return
*/
List<MythJobReadAhead> findMythJobReadAheadByTriggerNextTime(@Param("maxNextTime")long maxNextTime);
/**
* 添加单个任务节点
* @param mythJobReadAhead
*/
void addOneMythJobReadAhead(MythJobReadAhead mythJobReadAhead);
/**
* 删除已经被调度的任务根据ID
* @param id
*/
void removeMythJobReadAheadById(@Param("id") String id);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/dao/JobLogMapper.java deleted 100644 → 0
package com.byit.dao;
import com.byit.job.model.MythJobLog;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;
/**
* @program: byit-myth-job->JobLogMapper
* @description: 任务日志的操作
* @author: huangfu
* @date: 2019/12/16 14:35
**/
@Repository
public interface JobLogMapper {
/**
* 根据运行id和所有的父类id 查询没有完成的数据
* 后期查询一个节点的父节点是否完成时只需要根据运行标识和父节点id查询 返回结果为null的情况下成立
* @param runId
* @param parenIds
* @return
*/
List<MythJobLog> findUndoneMythJobLogByParenIdAndInRunId(@Param("runId") String runId, @Param("parenIds") List<String> parenIds);
/**
* 修改一条日志数据
* @param mythJobLog
*/
void updateOneMythJobLog(MythJobLog mythJobLog);
/**
* 添加一条数据
* @param mythJobLog
*/
void addOneMythJobLog(MythJobLog mythJobLog);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobTaskRunLog.java
...@@ -51,7 +51,7 @@ public class MythJobTaskRunLog implements Serializable { ...@@ -51,7 +51,7 @@ public class MythJobTaskRunLog implements Serializable {
/** /**
* 节点名称 * 节点名称
*/ */
@Column(columnDefinition = "int(13) COMMENT '节点名称'") @Column(columnDefinition = "varchar(255) COMMENT '节点名称'")
private String nodeName; private String nodeName;
/** /**
* 节点类型 node type * 节点类型 node type
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/MythJobTaskSchedule.java
...@@ -188,4 +188,9 @@ public class MythJobTaskSchedule implements Serializable { ...@@ -188,4 +188,9 @@ public class MythJobTaskSchedule implements Serializable {
*/ */
@Column(columnDefinition = "char(1) COMMENT '当前工作流版本的告警的时机'") @Column(columnDefinition = "char(1) COMMENT '当前工作流版本的告警的时机'")
private String mailAction; 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/MythJobTaskRunLogRepository.java 0 → 100644
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 0 → 100644
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/JobFlightScheduleService.java deleted 100644 → 0
package com.byit.service;
import com.byit.job.model.MythJobFlightSchedule;
import org.apache.ibatis.annotations.Param;
import java.util.List;
/**
* @program: byit-myth-job->JobFlightScheduleService
* @description: 任务排期表
* @author: huangfu
* @date: 2019/12/10 16:33
**/
public interface JobFlightScheduleService {
/**
* 查询预读五秒的数据
* @param maxNextTime
* @return
*/
List<MythJobFlightSchedule> findMythJobFlightScheduleByTriggerNextTime(long maxNextTime);
/**
* 批量向排期表添加任务数据
* @param mythJobFlightSchedules
*/
void addJobFlightSchedules(List<MythJobFlightSchedule> mythJobFlightSchedules);
/**
* 删除已经加载的节点
* @param id
*/
void deleteMythJobFlightScheduleById(String id);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobLogService.java deleted 100644 → 0
package com.byit.service;
import com.byit.job.dto.JobRunResultDto;
import com.byit.job.model.MythJobLog;
import java.util.List;
/**
* @program: byit-myth-job->JobLogService
* @description: 日志表操作
* @author: huangfu
* @date: 2019/12/16 17:01
**/
public interface JobLogService {
/**
* 根据运行id和所有的父类id 查询没有完成的数据
* 后期查询一个节点的父节点是否完成时只需要根据运行标识和父节点id查询 返回结果为null的情况下成立
* @param runId
* @param parenIds
* @return
*/
List<MythJobLog> findUndoneMythJobLogByParenIdAndInRunId(String runId,List<String> parenIds);
/**
* 修改一条日志数据
* @param jobRunResultDto
*/
void updateOneMythJobLog(JobRunResultDto jobRunResultDto);
/**
* 添加一条数据
* @param mythJobLog
*/
void addOneMythJobLog(MythJobLog mythJobLog);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobReadAheadService.java deleted 100644 → 0
package com.byit.service;
import com.byit.job.model.MythJobReadAhead;
import java.util.List;
/**
* @program: byit-myth-job->JobReadAheadService
* @description: 任务节点操作
* @author: huangfu
* @date: 2019/12/9 15:17
**/
public interface JobReadAheadService {
/**
* 查询7秒内要执行的数据
* @param maxNextTime
* @return
*/
List<MythJobReadAhead> findMythJobReadAheadByTriggerNextTime(long maxNextTime);
/**
* 添加单个任务节点
* @param mythJobReadAhead
*/
void addOneMythJobReadAhead(MythJobReadAhead mythJobReadAhead);
/**
* 删除已经被调度的任务根据ID
* @param id
*/
void removeMythJobReadAheadById(String id);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/MythJobTaskRunLogService.java 0 → 100644
package com.byit.service;
import com.byit.model.MythJobTaskRunLog;
/**
* @program: byit-myth-job->MythJobTaskRunLogService
* @description: 日志业务表
* @author: huangfu
* @date: 2019/12/20 19:42
**/
public interface MythJobTaskRunLogService {
/**
* 既是保存接口又是修改接口
* @return
*/
MythJobTaskRunLog save(MythJobTaskRunLog mythJobTaskRunLog);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/MythJobTaskScheduleService.java 0 → 100644
package com.byit.service;
import com.byit.model.MythJobTaskSchedule;
import java.util.List;
/**
* @program: byit-myth-job->MythJobTaskScheduleService
* @description: 排期表操作
* @author: huangfu
* @date: 2019/12/20 17:31
**/
public interface MythJobTaskScheduleService {
/**
* 根据下次执行时间+5000毫秒 查询所有任务节点
* @param maxNextTime
* @return
*/
List<MythJobTaskSchedule> findMythJobTaskScheduleByTriggerNextTimeLessThanEqual(long maxNextTime);
/**
* 保存任务表预读的数据
* @param mythJobTaskSchedules
* @return
*/
List<MythJobTaskSchedule> save(List<MythJobTaskSchedule> mythJobTaskSchedules);
/**
* 根据id删除
* @param id
*/
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
package com.byit.service; package com.byit.service;
import com.byit.job.model.MythJobReadAhead;
import com.byit.model.MythJobTask; import com.byit.model.MythJobTask;
import java.util.List; import java.util.List;
...@@ -30,4 +29,10 @@ public interface MythJobTaskService { ...@@ -30,4 +29,10 @@ public interface MythJobTaskService {
* @param id * @param id
*/ */
void removeMythJobTaskById(Integer id); void removeMythJobTaskById(Integer id);
/**
* 根据集合删除
* @param mythJobTasks
*/
void removeMythJobTaskInIds(List<MythJobTask> mythJobTasks);
} }
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobFlightScheduleServiceImpl.java deleted 100644 → 0
package com.byit.service.impl;
import com.byit.dao.JobFlightScheduleMapper;
import com.byit.job.model.MythJobFlightSchedule;
import com.byit.service.JobFlightScheduleService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
* @program: byit-myth-job->JobFlightScheduleServiceImpl
* @description: 任务排期表
* @author: huangfu
* @date: 2019/12/10 16:36
**/
@Service
public class JobFlightScheduleServiceImpl implements JobFlightScheduleService {
private final JobFlightScheduleMapper jobFlightScheduleMapper;
@Autowired
public JobFlightScheduleServiceImpl(JobFlightScheduleMapper jobFlightScheduleMapper) {
this.jobFlightScheduleMapper = jobFlightScheduleMapper;
}
@Override
public List<MythJobFlightSchedule> findMythJobFlightScheduleByTriggerNextTime(long maxNextTime) {
return jobFlightScheduleMapper.findMythJobFlightScheduleByTriggerNextTime(maxNextTime);
}
@Override
public void addJobFlightSchedules(List<MythJobFlightSchedule> mythJobFlightSchedules) {
jobFlightScheduleMapper.addJobFlightSchedules(mythJobFlightSchedules);
}
@Override
public void deleteMythJobFlightScheduleById(String id) {
jobFlightScheduleMapper.deleteMythJobFlightScheduleById(id);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobLogServiceImpl.java deleted 100644 → 0
package com.byit.service.impl;
import com.byit.dao.JobLogMapper;
import com.byit.job.dto.JobRunResultDto;
import com.byit.job.model.MythJobLog;
import com.byit.service.JobLogService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.Date;
import java.util.List;
/**
* @program: byit-myth-job->JobLogServiceImpl
* @description: 日志表操作
* @author: huangfu
* @date: 2019/12/16 17:02
**/
@Service
public class JobLogServiceImpl implements JobLogService {
private final JobLogMapper jobLogMapper;
@Autowired
public JobLogServiceImpl(JobLogMapper jobLogMapper) {
this.jobLogMapper = jobLogMapper;
}
@Override
public List<MythJobLog> findUndoneMythJobLogByParenIdAndInRunId(String runId, List<String> parenIds) {
return jobLogMapper.findUndoneMythJobLogByParenIdAndInRunId(runId,parenIds);
}
@Override
public void updateOneMythJobLog(JobRunResultDto jobRunResultDto) {
MythJobLog mythJobLog = new MythJobLog();
mythJobLog.setId(jobRunResultDto.getLogId());
mythJobLog.setHandleTime(new Date());
mythJobLog.setHandleCode(jobRunResultDto.getReturnResult().getCode());
mythJobLog.setHandleMsg(jobRunResultDto.getReturnResult().getMsg());
jobLogMapper.updateOneMythJobLog(mythJobLog);
}
@Override
public void addOneMythJobLog(MythJobLog mythJobLog) {
jobLogMapper.addOneMythJobLog(mythJobLog);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobReadAheadServiceImpl.java deleted 100644 → 0
package com.byit.service.impl;
import com.byit.dao.JobInfoMapper;
import com.byit.job.model.MythJobReadAhead;
import com.byit.service.JobReadAheadService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
* @program: byit-myth-job->JobReadAheadServiceImpl
* @description: 任务节点操作的实现类
* @author: huangfu
* @date: 2019/12/9 15:18
**/
@Service
@Slf4j
public class JobReadAheadServiceImpl implements JobReadAheadService {
private final JobInfoMapper jobInfoMapper;
@Autowired
public JobReadAheadServiceImpl(JobInfoMapper jobInfoMapper) {
this.jobInfoMapper = jobInfoMapper;
}
@Override
public List<MythJobReadAhead> findMythJobReadAheadByTriggerNextTime(long maxNextTime) {
return jobInfoMapper.findMythJobReadAheadByTriggerNextTime(maxNextTime);
}
@Override
public void addOneMythJobReadAhead(MythJobReadAhead mythJobReadAhead) {
jobInfoMapper.addOneMythJobReadAhead(mythJobReadAhead);
}
@Override
public void removeMythJobReadAheadById(String id) {
jobInfoMapper.removeMythJobReadAheadById(id);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/MythJobTaskRunLogServiceImpl.java 0 → 100644
package com.byit.service.impl;
import com.byit.model.MythJobTaskRunLog;
import com.byit.repository.MythJobTaskRunLogRepository;
import com.byit.service.MythJobTaskRunLogService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
* @program: byit-myth-job->MythJobTaskRunLogServiceImpl
* @description: 日志业务表
* @author: huangfu
* @date: 2019/12/20 19:43
**/
@Service
public class MythJobTaskRunLogServiceImpl implements MythJobTaskRunLogService {
@Autowired
private MythJobTaskRunLogRepository mythJobTaskRunLogRepository;
@Override
public MythJobTaskRunLog save(MythJobTaskRunLog mythJobTaskRunLog) {
return mythJobTaskRunLogRepository.save(mythJobTaskRunLog);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/MythJobTaskScheduleServiceImpl.java 0 → 100644
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
...@@ -33,4 +33,11 @@ public class MythJobTaskServiceImpl implements MythJobTaskService { ...@@ -33,4 +33,11 @@ public class MythJobTaskServiceImpl implements MythJobTaskService {
public void removeMythJobTaskById(Integer id) { public void removeMythJobTaskById(Integer id) {
mythJobTaskRepository.delete(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
...@@ -4,19 +4,18 @@ import cn.hutool.http.HttpUtil; ...@@ -4,19 +4,18 @@ import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
import com.byit.job.dto.AdminSenPluginDto; import com.byit.job.dto.AdminSenPluginDto;
import com.byit.job.exceptions.plugin.PluginException; import com.byit.job.exceptions.plugin.PluginException;
import com.byit.job.model.MythJobFlightSchedule;
import com.byit.job.model.MythJobLog;
import com.byit.job.utils.IpUtil; import com.byit.job.utils.IpUtil;
import com.byit.model.MythJobTaskRunLog;
import com.byit.model.MythJobTaskSchedule;
import com.byit.rpc.remoting.invoker.route.LoadBalance; import com.byit.rpc.remoting.invoker.route.LoadBalance;
import com.byit.rpc.remoting.invoker.route.RpcLoadBalance; import com.byit.rpc.remoting.invoker.route.RpcLoadBalance;
import com.byit.service.impl.JobLogServiceImpl; import com.byit.service.impl.MythJobTaskRunLogServiceImpl;
import com.byit.util.SpringUtil; import com.byit.util.SpringUtil;
import io.netty.util.Timeout; import io.netty.util.Timeout;
import io.netty.util.TimerTask; import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import java.util.Date; import java.util.Date;
import java.util.UUID;
/** /**
* @program: byit-myth-job->JvavBeanJobTask * @program: byit-myth-job->JvavBeanJobTask
...@@ -26,30 +25,30 @@ import java.util.UUID; ...@@ -26,30 +25,30 @@ import java.util.UUID;
**/ **/
@Slf4j @Slf4j
public class JavaBeanJobTask implements TimerTask { public class JavaBeanJobTask implements TimerTask {
private MythJobFlightSchedule mythJobFlightSchedule; private MythJobTaskSchedule mythJobTaskSchedule;
public JavaBeanJobTask(MythJobTaskSchedule mythJobTaskSchedule) {
public JavaBeanJobTask( MythJobFlightSchedule mythJobFlightSchedule) { this.mythJobTaskSchedule = mythJobTaskSchedule;
this.mythJobFlightSchedule = mythJobFlightSchedule;
} }
@Override @Override
public void run(Timeout timeout) { public void run(Timeout timeout) {
//根据负责均衡方案获取对应IP //根据负责均衡方案获取对应IP
RpcLoadBalance rpcInvokerRouter = LoadBalance.match(mythJobFlightSchedule.getRoutingStrategy( ), LoadBalance.ROUND).rpcInvokerRouter; RpcLoadBalance rpcInvokerRouter = LoadBalance.match(mythJobTaskSchedule.getRoutingStrategy( ), LoadBalance.ROUND).rpcInvokerRouter;
try{ try{
String url = IpUtil.electiveUrl(mythJobFlightSchedule.getPluginUrls( ), rpcInvokerRouter); String url = IpUtil.electiveUrl(mythJobTaskSchedule.getPluginUrls( ), rpcInvokerRouter);
String jobHandelName = mythJobFlightSchedule.getExecutorHandler( ); String jobHandelName = mythJobTaskSchedule.getLocalNodeHandlerName();
String param = mythJobFlightSchedule.getJobParam( ); String param = mythJobTaskSchedule.getRunParam( );
String runId = mythJobFlightSchedule.getRunID(); String runId = mythJobTaskSchedule.getRunId();
String logId = UUID.randomUUID( ).toString().replace("-","");
AdminSenPluginDto adminSenPluginDto = new AdminSenPluginDto(); AdminSenPluginDto adminSenPluginDto = new AdminSenPluginDto();
adminSenPluginDto.setCallbackUrl("http://127.0.0.1:8080/job/callbackRes"); adminSenPluginDto.setCallbackUrl("http://127.0.0.1:8080/job/callbackRes");
adminSenPluginDto.setJobHandelName(jobHandelName); adminSenPluginDto.setJobHandelName(jobHandelName);
adminSenPluginDto.setJobParam(param); adminSenPluginDto.setJobParam(param);
adminSenPluginDto.setRunId(runId); adminSenPluginDto.setRunId(runId);
adminSenPluginDto.setLogId(logId); adminSenPluginDto.setLogId(mythJobTaskSchedule.getLogId());
//这里获取的是调度结果 //这里获取的是调度结果
String result = HttpUtil.post(url, JSON.toJSONString(adminSenPluginDto),10*1000); String result = HttpUtil.post(url, JSON.toJSONString(adminSenPluginDto),10*1000);
saveLog(mythJobFlightSchedule,logId,url,result); //修改调度结果
saveLog(mythJobTaskSchedule, url, result);
log.debug("---------------{}------------",result); log.debug("---------------{}------------",result);
...@@ -60,29 +59,19 @@ public class JavaBeanJobTask implements TimerTask { ...@@ -60,29 +59,19 @@ public class JavaBeanJobTask implements TimerTask {
} }
} }
private void saveLog(MythJobFlightSchedule mythJobFlightSchedule,String logId,String url,String result){ private Integer saveLog(MythJobTaskSchedule mythJobTaskSchedule,String url,String result){
MythJobLog mythJobLog = new MythJobLog( ); MythJobTaskRunLog mythJobTaskRunLog = new MythJobTaskRunLog();
//设置邮件
mythJobLog.setId(logId);
mythJobLog.setJobGroup(url);
mythJobLog.setJobId(mythJobFlightSchedule.getId());
mythJobLog.setFlowId(mythJobFlightSchedule.getTaskFlowId());
mythJobLog.setRunId(mythJobFlightSchedule.getRunID());
mythJobLog.setExecutorHandler(mythJobFlightSchedule.getExecutorHandler());
mythJobLog.setExecutorParams(mythJobFlightSchedule.getJobParam());
//TODO 有问题
mythJobLog.setTriggerCode("1");
mythJobLog.setTriggerMsg(result);
mythJobLog.setTriggerTime(new Date());
mythJobLog.setJobName(mythJobFlightSchedule.getJobName());
mythJobLog.setAlarmEmail(mythJobFlightSchedule.getAlarmEmail());
JobLogServiceImpl bean = SpringUtil.getBean(JobLogServiceImpl.class);
bean.addOneMythJobLog(mythJobLog);
//版本id需要查验
//还需要携带版本的名字
mythJobTaskRunLog.setLogId(mythJobTaskSchedule.getLogId());
mythJobTaskRunLog.setRunType("2");
mythJobTaskRunLog.setLocalNodeHandlerName(mythJobTaskSchedule.getLocalNodeHandlerName());
mythJobTaskRunLog.setTriggerTime(new Date());
mythJobTaskRunLog.setTriggerCode(result);
mythJobTaskRunLog.setTriggerMsg("SUCCESS");
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/thread/JobScheduleHelper.java
...@@ -2,11 +2,14 @@ package com.byit.thread; ...@@ -2,11 +2,14 @@ package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil; import cn.hutool.core.collection.CollectionUtil;
import com.byit.job.WorkRoulette; import com.byit.job.WorkRoulette;
import com.byit.job.model.MythJobFlightSchedule; import com.byit.model.MythJobTask;
import com.byit.job.model.MythJobReadAhead; import com.byit.model.MythJobTaskRunLog;
import com.byit.service.JobFlightScheduleService; import com.byit.model.MythJobTaskSchedule;
import com.byit.service.JobReadAheadService; import com.byit.service.MythJobTaskScheduleService;
import com.byit.service.MythJobTaskService;
import com.byit.service.impl.MythJobTaskRunLogServiceImpl;
import com.byit.task.JavaBeanJobTask; import com.byit.task.JavaBeanJobTask;
import com.byit.util.SpringUtil;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils; import org.springframework.beans.BeanUtils;
...@@ -33,10 +36,12 @@ import java.util.concurrent.TimeUnit; ...@@ -33,10 +36,12 @@ import java.util.concurrent.TimeUnit;
@Component @Component
public class JobScheduleHelper{ public class JobScheduleHelper{
private DataSource dataSource; private DataSource dataSource;
@Autowired @Autowired
private JobReadAheadService jobReadAheadService; private MythJobTaskService mythJobTaskService;
@Autowired @Autowired
private JobFlightScheduleService jobFlightScheduleService; private MythJobTaskScheduleService mythJobTaskScheduleService;
/** /**
* 读取任务节点的预读 * 读取任务节点的预读
...@@ -99,27 +104,27 @@ public class JobScheduleHelper{ ...@@ -99,27 +104,27 @@ public class JobScheduleHelper{
//行锁已经加上 后续处理 //行锁已经加上 后续处理
long nowTime = System.currentTimeMillis(); long nowTime = System.currentTimeMillis();
//开始寻找此时 不是暂停状态，而且七秒内即将运行的任务 //开始寻找此时 不是暂停状态，而且七秒内即将运行的任务
List<MythJobReadAhead> mythJobReadAheadByTriggerNextTime = jobReadAheadService.findMythJobReadAheadByTriggerNextTime(nowTime + PRE_READ_MS); List<MythJobTask> mythJobTasks = mythJobTaskService.findMythJobTaskByTriggerNextTimeLessThanEqual(nowTime + PRE_READ_MS);
if(CollectionUtil.isNotEmpty(mythJobReadAheadByTriggerNextTime)){ if(CollectionUtil.isNotEmpty(mythJobTasks)){
List<MythJobFlightSchedule> mythJobFlightSchedules = new ArrayList<>(15); List<MythJobTaskSchedule> mythJobTaskSchedules = new ArrayList<MythJobTaskSchedule>(15);
mythJobReadAheadByTriggerNextTime.forEach(mythJobInfo -> { mythJobTasks.forEach(mythJobTask -> {
/** /**
* 需要去检验当前任务的上级节点是否已经执行成功，没有执行，或者处于暂停状态则跳过该任务 * 需要去检验当前任务的上级节点是否已经执行成功，没有执行，或者处于暂停状态则跳过该任务
* 大概思路，根据任务流id，从任务流执行回溯表查询该任务流的所有节点，查看上级节点是否已经执行成功 * 大概思路，根据任务流id，从任务流执行回溯表查询该任务流的所有节点，查看上级节点是否已经执行成功
* //TODO 需要修改 判断父节点是否执行完毕 注意 父节点是一个集合 * //TODO 需要修改 判断父节点是否执行完毕 注意 父节点是一个集合
*/ */
if(StringUtils.isNotBlank(mythJobInfo.getParentId())){ if(StringUtils.isNotBlank(mythJobTask.getDependencyNodes())){
log.debug("任务：{}", mythJobInfo); log.debug("任务：{}", mythJobTask);
MythJobFlightSchedule mythJobFlightSchedule = new MythJobFlightSchedule(); MythJobTaskSchedule mythJobTaskSchedule = new MythJobTaskSchedule();
BeanUtils.copyProperties(mythJobInfo,mythJobFlightSchedule); BeanUtils.copyProperties(mythJobTask,mythJobTaskSchedule);
mythJobFlightSchedules.add(mythJobFlightSchedule); mythJobTaskSchedules.add(mythJobTaskSchedule);
jobReadAheadService.removeMythJobReadAheadById(mythJobInfo.getId());
} }
}); });
if(CollectionUtil.isNotEmpty(mythJobFlightSchedules)) { if(CollectionUtil.isNotEmpty(mythJobTaskSchedules)) {
jobFlightScheduleService.addJobFlightSchedules(mythJobFlightSchedules); mythJobTaskScheduleService.save(mythJobTaskSchedules);
mythJobTaskService.removeMythJobTaskInIds(mythJobTasks);
} }
}else{ }else{
...@@ -233,14 +238,16 @@ public class JobScheduleHelper{ ...@@ -233,14 +238,16 @@ public class JobScheduleHelper{
//行锁已经加上 后续处理 //行锁已经加上 后续处理
long nowTime = System.currentTimeMillis(); long nowTime = System.currentTimeMillis();
//查询所有符合条件的任务节点 //查询所有符合条件的任务节点
List<MythJobFlightSchedule> mythJobFlightScheduleByTriggerNextTimes = jobFlightScheduleService.findMythJobFlightScheduleByTriggerNextTime(nowTime + SCHEDULE_READ_MS); List<MythJobTaskSchedule> mythJobTaskSchedules = mythJobTaskScheduleService.findMythJobTaskScheduleByTriggerNextTimeLessThanEqual(nowTime + SCHEDULE_READ_MS);
if(CollectionUtil.isNotEmpty(mythJobFlightScheduleByTriggerNextTimes)){ if(CollectionUtil.isNotEmpty(mythJobTaskSchedules)){
//循环遍历添加任务 //循环遍历添加任务
mythJobFlightScheduleByTriggerNextTimes.forEach(mythJobFlightScheduleByTriggerNextTime ->{ mythJobTaskSchedules.forEach(mythJobTaskSchedule ->{
if ("BEAN".equals(mythJobFlightScheduleByTriggerNextTime.getJobType())) { Integer logId = saveLog(mythJobTaskSchedule);
JavaBeanJobTask javaBeanJobTask = new JavaBeanJobTask(mythJobFlightScheduleByTriggerNextTime); mythJobTaskSchedule.setLogId(logId);
jobFlightScheduleService.deleteMythJobFlightScheduleById(mythJobFlightScheduleByTriggerNextTime.getId()); if ("BEAN".equals(mythJobTaskSchedule.getJobType())) {
WorkRoulette.addJob(javaBeanJobTask,mythJobFlightScheduleByTriggerNextTime.getTriggerNextTime()); JavaBeanJobTask javaBeanJobTask = new JavaBeanJobTask(mythJobTaskSchedule);
mythJobTaskScheduleService.delete(mythJobTaskSchedule.getNodeId());
WorkRoulette.addJob(javaBeanJobTask,mythJobTaskSchedule.getTriggerNextTime());
} }
}); });
}else{ }else{
...@@ -317,4 +324,20 @@ public class JobScheduleHelper{ ...@@ -317,4 +324,20 @@ public class JobScheduleHelper{
this.dataSource = dataSource; this.dataSource = dataSource;
} }
private Integer saveLog(MythJobTaskSchedule mythJobTaskSchedule){
MythJobTaskRunLog mythJobTaskRunLog = new MythJobTaskRunLog();
mythJobTaskRunLog.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());
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/thread/LogCallbackThread.java
package com.byit.thread; package com.byit.thread;
import com.byit.job.dto.JobRunResultDto; import com.byit.job.dto.JobRunResultDto;
import com.byit.service.JobLogService; import com.byit.model.MythJobTaskRunLog;
import com.byit.service.impl.JobLogServiceImpl; import com.byit.service.impl.MythJobTaskRunLogServiceImpl;
import com.byit.util.SpringUtil; import com.byit.util.SpringUtil;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
...@@ -24,7 +24,12 @@ public class LogCallbackThread implements Runnable { ...@@ -24,7 +24,12 @@ public class LogCallbackThread implements Runnable {
@Override @Override
public void run() { public void run() {
log.debug("--------------------任务执行完成---------------------"); log.debug("--------------------任务执行完成---------------------");
JobLogService jobLogService = SpringUtil.getBean(JobLogServiceImpl.class); MythJobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(MythJobTaskRunLogServiceImpl.class);
jobLogService.updateOneMythJobLog(jobRunResultDto); MythJobTaskRunLog mythJobTaskRunLog = new MythJobTaskRunLog( );
mythJobTaskRunLog.setLogId(jobRunResultDto.getLogId());
mythJobTaskRunLog.setRunTime(jobRunResultDto.getEndTime());
mythJobTaskRunLog.setRunCode(jobRunResultDto.getReturnResult().getCode());
mythJobTaskRunLog.setRunMsg(jobRunResultDto.getReturnResult().getMsg());
mythJobTaskRunLogService.save(mythJobTaskRunLog);
} }
} }
byit-myth-core/myth-admin-core/src/main/java/com/byit/util/SourceObj2TargetObjUtil.java 0 → 100644
package com.byit.util;
import com.byit.job.dto.PluginBeanJobInfo;
import com.byit.model.MythJobTask;
import java.util.Date;
import java.util.UUID;
/**
* @program: byit-myth-job->SourceObj2TargetObjUtil
* @description: 这是一个转换的工具类，定义两个类之间的相互转换
* @author: huangfu
* @date: 2019/12/14 13:00
**/
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("1");
return mythJobTask;
}
}
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobFlightScheduleMapper.xml deleted 100644 → 0
<?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.dao.JobFlightScheduleMapper">
<sql id="Base_Column_List">
t.id,
t.job_name,
t.executor_handler,
t.job_cron,
t.job_desc,
t.plugin_urls,
t.routing_strategy,
t.blocking_strategy,
t.callback_token,
t.gateway_token,
t.request_host,
t.request_port,
t.job_param,
t.job_type,
t.taskflow_id,
t.parent_id,
t.taskflow_is_cron,
t.alarm_email,
t.executor_timeout,
t.source_urls,
t.glue_source,
t.glue_remark,
t.glue_updatetime,
t.trigger_next_time,
t.author,
t.add_time,
t.update_time,
t.run_id
</sql>
<resultMap id="mythJobFlightSchedule" type="com.byit.job.model.MythJobFlightSchedule">
<id column="id" property="id"/>
<result column="job_name" property="jobName"/>
<result column="executor_handler" property="executorHandler"/>
<result column="job_cron" property="jobCron"/>
<result column="job_desc" property="jobDesc"/>
<result column="plugin_urls" property="pluginUrls"/>
<result column="routing_strategy" property="routingStrategy"/>
<result column="blocking_strategy" property="blockingStrategy"/>
<result column="callback_token" property="callbackToken"/>
<result column="gateway_token" property="gatewayToken"/>
<result column="request_host" property="requestHost"/>
<result column="request_port" property="requestPort"/>
<result column="job_param" property="jobParam"/>
<result column="job_type" property="jobType"/>
<result column="taskflow_id" property="taskFlowId"/>
<result column="parent_id" property="parentId"/>
<result column="taskflow_is_cron" property="taskFlowIsCron"/>
<result column="alarm_email" property="alarmEmail"/>
<result column="executor_timeout" property="executorTimeout"/>
<result column="source_urls" property="sourceUrls"/>
<result column="glue_source" property="glueSource"/>
<result column="glue_remark" property="glueRemark"/>
<result column="glue_updatetime" property="glueUpdateTime"/>
<result column="trigger_next_time" property="triggerNextTime"/>
<result column="author" property="author"/>
<result column="add_time" property="addTime"/>
<result column="update_time" property="updateTime"/>
<result column="run_id" property="runID"/>
</resultMap>
<select id="findMythJobFlightScheduleByTriggerNextTime" resultMap="mythJobFlightSchedule">
SELECT <include refid="Base_Column_List" />
FROM job_flight_schedule AS t
WHERE t.trigger_next_time<![CDATA[ <= ]]> #{maxNextTime}
</select>
<insert id="addJobFlightSchedules" parameterType="com.byit.job.model.MythJobFlightSchedule">
INSERT INTO job_flight_schedule
(
id,
job_name,
executor_handler,
job_cron,
job_desc,
plugin_urls,
routing_strategy,
blocking_strategy,
callback_token,
gateway_token,
request_host,
request_port,
job_param,
job_type,
taskflow_id,
parent_id,
taskflow_is_cron,
alarm_email,
executor_timeout,
source_urls,
glue_source,
glue_remark,
glue_updatetime,
trigger_next_time,
author,
add_time,
update_time,
run_id
)
VALUES
<foreach collection="mythJobFlightSchedule" separator="," item="item" index="index">
(
#{item.id},
#{item.jobName},
#{item.executorHandler},
#{item.jobCron},
#{item.jobDesc},
#{item.pluginUrls},
#{item.routingStrategy},
#{item.blockingStrategy},
#{item.callbackToken},
#{item.gatewayToken},
#{item.requestHost},
#{item.requestPort},
#{item.jobParam},
#{item.jobType},
#{item.taskFlowId},
#{item.parentId},
#{item.taskFlowIsCron},
#{item.alarmEmail},
#{item.executorTimeout},
#{item.sourceUrls},
#{item.glueSource},
#{item.glueRemark},
#{item.glueUpdateTime},
#{item.triggerNextTime},
#{item.author},
#{item.addTime},
#{item.updateTime},
#{item.runID}
)
</foreach>
</insert>
<delete id="deleteMythJobFlightScheduleById">
DELETE FROM job_flight_schedule WHERE id= #{id}
</delete>
</mapper>
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobLogMapper.xml deleted 100644 → 0
<?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.dao.JobLogMapper">
<sql id="Base_Column_List">
t.id,
t.job_group,
t.job_id,
t.flow_id,
t.run_id,
t.executor_handler,
t.executor_params,
t.trigger_time,
t.trigger_code,
t.trigger_msg,
t.handle_time,
t.handle_code,
t.handle_msg,
t.alarm_status,
t.job_name,
t.alarm_email
</sql>
<resultMap id="mythJobLog" type="com.byit.job.model.MythJobLog">
<id column="id" property="id"/>
<result column="job_group" property="jobGroup"/>
<result column="job_id" property="jobId"/>
<result column="flow_id" property="flowId"/>
<result column="run_id" property="runId"/>
<result column="executor_handler" property="executorHandler"/>
<result column="executor_params" property="executorParams"/>
<result column="trigger_time" property="triggerTime"/>
<result column="trigger_code" property="triggerCode"/>
<result column="trigger_msg" property="triggerMsg"/>
<result column="handle_time" property="handleTime"/>
<result column="handle_code" property="handleCode"/>
<result column="handle_msg" property="handleMsg"/>
<result column="alarm_status" property="alarmStatus"/>
<result column="job_name" property="jobName"/>
<result column="alarm_email" property="alarmEmail"/>
</resultMap>
<select id="findUndoneMythJobLogByParenIdAndInRunId" resultMap="mythJobLog">
SELECT <include refid="Base_Column_List"/> FROM job_log t
<where>
run_id = #{runId} AND handle_code = '0' AND
job_id IN
<foreach collection="parenIds" item="jobId" index="index" open="(" close=")" separator=",">
#{jobId}
</foreach>
</where>
</select>
<update id="updateOneMythJobLog" parameterType="com.byit.job.model.MythJobLog">
UPDATE job_log
<trim prefix="SET" suffixOverrides=",">
<if test="jobGroup!= null and jobGroup != ''">job_group=#{jobGroup},</if>
<if test="jobId!= null and jobId != ''">job_id=#{jobId},</if>
<if test="flowId!= null and flowId != ''">flow_id=#{flowId},</if>
<if test="runId!= null and runId != ''">run_id=#{runId},</if>
<if test="executorHandler!= null and executorHandler != ''">executor_handler=#{executorHandler},</if>
<if test="executorParams!= null and executorParams != ''">executor_params=#{executorParams},</if>
<if test="triggerTime != null">trigger_time=#{triggerTime},</if>
<if test="triggerCode!= null and triggerCode != ''">trigger_code=#{triggerCode},</if>
<if test="triggerMsg!= null and triggerMsg != ''">trigger_msg=#{triggerMsg},</if>
<if test="handleTime!= null">handle_time=#{handleTime},</if>
<if test="handleCode!= null and handleCode != ''">handle_code=#{handleCode},</if>
<if test="handleMsg!= null and handleMsg != ''">handle_msg=#{handleMsg},</if>
<if test="alarmStatus!= null and alarmStatus != ''">alarm_status=#{alarmStatus},</if>
<if test="jobName!= null and jobName != ''">job_name=#{jobName},</if>
<if test="alarmEmail!= null and alarmEmail != ''">alarm_email=#{alarmEmail},</if>
</trim>
WHERE id = #{id}
</update>
<insert id="addOneMythJobLog" parameterType="com.byit.job.model.MythJobLog">
INSERT INTO job_log (
id,job_group,job_id,flow_id,run_id,executor_handler,
executor_params,trigger_time,trigger_code,trigger_msg,handle_time,
handle_code,handle_msg,alarm_status,job_name,alarm_email
)
values (
#{id},#{jobGroup},#{jobId},#{flowId},#{runId},#{executorHandler},#{executorParams},
#{triggerTime},#{triggerCode},#{triggerMsg},#{handleTime},#{handleCode},#{handleMsg},#{alarmStatus},
#{jobName},#{alarmEmail}
)
</insert>
</mapper>
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/resources/mapper/MythJobInfoMapper.xml deleted 100644 → 0
<?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.dao.JobInfoMapper">
<sql id="Base_Column_List">
t.id,
t.job_name,
t.executor_handler,
t.job_cron,
t.job_desc,
t.plugin_urls,
t.routing_strategy,
t.blocking_strategy,
t.callback_token,
t.gateway_token,
t.request_host,
t.request_port,
t.job_param,
t.job_type,
t.taskflow_id,
t.parent_id,
t.taskflow_is_cron,
t.alarm_email,
t.executor_timeout,
t.source_urls,
t.trigger_status,
t.glue_source,
t.glue_remark,
t.glue_updatetime,
t.trigger_next_time,
t.author,
t.add_time,
t.update_time,
t.run_id
</sql>
<resultMap id="MythJobInfo" type="com.byit.job.model.MythJobReadAhead" >
<id column="id" property="id"/>
<result column="job_name" property="jobName"/>
<result column="executor_handler" property="executorHandler"/>
<result column="job_cron" property="jobCron"/>
<result column="job_desc" property="jobDesc"/>
<result column="plugin_urls" property="pluginUrls"/>
<result column="routing_strategy" property="routingStrategy"/>
<result column="blocking_strategy" property="blockingStrategy"/>
<result column="callback_token" property="callbackToken"/>
<result column="gateway_token" property="gatewayToken"/>
<result column="request_host" property="requestHost"/>
<result column="request_port" property="requestPort"/>
<result column="job_param" property="jobParam"/>
<result column="job_type" property="jobType"/>
<result column="taskflow_id" property="taskFlowId"/>
<result column="parent_id" property="parentId"/>
<result column="taskflow_is_cron" property="taskFlowIsCron"/>
<result column="alarm_email" property="alarmEmail"/>
<result column="executor_timeout" property="executorTimeout"/>
<result column="source_urls" property="sourceUrls"/>
<result column="trigger_status" property="triggerStatus"/>
<result column="glue_source" property="glueSource"/>
<result column="glue_remark" property="glueRemark"/>
<result column="glue_updatetime" property="glueUpdateTime"/>
<result column="trigger_next_time" property="triggerNextTime"/>
<result column="author" property="author"/>
<result column="add_time" property="addTime"/>
<result column="update_time" property="updateTime"/>
<result column="run_id" property="runID"/>
</resultMap>
<select id="findMythJobReadAheadByTriggerNextTime" resultMap="MythJobInfo">
SELECT <include refid="Base_Column_List" />
FROM job_read_ahead AS t
WHERE t.trigger_status=1
AND t.trigger_next_time<![CDATA[ <= ]]> #{maxNextTime}
</select>
<insert id="addOneMythJobReadAhead" parameterType="com.byit.job.model.MythJobReadAhead">
insert into job_read_ahead (id,job_name,executor_handler,job_cron,job_desc,plugin_urls,routing_strategy,
blocking_strategy,callback_token,gateway_token,request_host,request_port,job_param,job_type,taskflow_id,
parent_id,taskflow_is_cron,alarm_email,executor_timeout,source_urls,trigger_status,glue_source,glue_remark,
glue_updatetime,trigger_next_time,author,add_time,update_time,run_id)
values(#{id},#{jobName},#{executorHandler},#{jobCron},#{jobDesc},#{pluginUrls},#{routingStrategy},#{blockingStrategy},
#{callbackToken},#{gatewayToken},#{requestHost},#{requestPort},#{jobParam},#{jobType},#{taskFlowId},
#{parentId},#{taskFlowIsCron},#{alarmEmail},#{executorTimeout},#{sourceUrls},#{triggerStatus},#{glueSource},#{glueRemark},
#{glueUpdateTime},#{triggerNextTime},#{author},#{addTime},#{updateTime},#{runID})
</insert>
<delete id="removeMythJobReadAheadById">
DELETE FROM job_read_ahead WHERE id=#{id}
</delete>
</mapper>
\ No newline at end of file
byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/AdminSenPluginDto.java
...@@ -39,5 +39,5 @@ public class AdminSenPluginDto { ...@@ -39,5 +39,5 @@ public class AdminSenPluginDto {
/** /**
* 日志ID * 日志ID
*/ */
private String logId; private Integer logId;
} }
byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/JobRunResultDto.java
...@@ -36,6 +36,6 @@ public class JobRunResultDto { ...@@ -36,6 +36,6 @@ public class JobRunResultDto {
*/ */
private Date endTime; private Date endTime;
private String logId; private Integer logId;
} }
byit-myth-core/myth-core-common/src/main/java/com/byit/job/model/MythJobFlightSchedule.java deleted 100644 → 0
package com.byit.job.model;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;
import java.util.Date;
/**
* @program: byit-myth-job->MythJobFlightSchedule
* @description: 任务排期表
* @author: huangfu
* @date: 2019/12/10 15:05
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class MythJobFlightSchedule {
/**
* 任务节点的id
*/
private String id;
/**
* 任务节点的名字
*/
private String jobName;
/**
* 插件任务调度的key,调度中心会根据这个key找到对应的插件端任务，执行
*/
private String executorHandler;
/**
* 任务周期调度cron表达式
*/
private String jobCron;
/**
* 任务详情，展示在调度中心平台的备注
*/
private String jobDesc;
/**
* 插件地址的集合
*/
private String pluginUrls;
/**
* 路由策略
* 1随机（默认）
* 2轮询
* 3最近最少使用
* 4最近最久未使用算法
* 5哈希算法
*/
private String routingStrategy;
/**
* 阻塞策略：
* 1丢弃
* 2阻塞等待（默认）
*/
private String blockingStrategy;
/**
* 回调时的身份认证
*/
private String callbackToken;
/**
* 请求调度中心的token
*/
private String gatewayToken;
/**
* 调度中心主机名
*/
private String requestHost;
/**
* 调度中心端口号
*/
private String requestPort;
/**
* 任务参数
*/
private String jobParam;
/**
* 任务类型，java,python,php,script,sql.shell
*/
private String jobType;
/**
* 所属任务流的id
*/
private String taskFlowId;
/**
* 上级节点
*/
private String parentId;
/**
* 是否跟随任务流的时间设置？1不跟随(默认),2跟随
*/
private String taskFlowIsCron;
/**
* 报警邮件
*/
private String alarmEmail;
/**
* 任务的超时时间
*/
private Integer executorTimeout;
/**
* 脚本文件地址(或远程文件服务器地址)支持多个，逗号分割
*/
private String sourceUrls;
/**
*调度任务源码
*/
private String glueSource;
/**
* 源码备注
*/
private String glueRemark;
/**
* 源码的修改时间
*/
private Date glueUpdateTime;
/**
* 任务下次调度时间
*/
private Long triggerNextTime;
/**
* 任务创建者
*/
private String author;
/**
* 任务节点添加时间
*/
private Date addTime;
/**
* 任务节点修改时间
*/
private Date updateTime;
/**
* 运行标识
*/
private String runID;
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/model/MythJobLog.java deleted 100644 → 0
package com.byit.job.model;
import lombok.*;
import java.util.Date;
/**
* @program: byit-myth-job->MythJobLog
* @description: 任务日志表实体
* @author: huangfu
* @date: 2019/12/16 14:22
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
@EqualsAndHashCode
public class MythJobLog {
private String id;
/**
* 执行器的id
*/
private String jobGroup;
/**
* 任务节点主键
*/
private String jobId;
/**
* 任务流id
*/
private String flowId;
/**
* 运行标识
*/
private String runId;
/**
* 执行器任务handler
*/
private String executorHandler;
/**
* 执行器的参数
*/
private String executorParams;
/**
*调度时间
*/
private Date triggerTime;
/**
* 调度结果 1成功 2失败
*/
private String triggerCode;
/**
* 调度日志
*/
private String triggerMsg;
/**
*执行时间
*/
private Date handleTime;
/**
* 执行结果0未执行完成 1成功 2失败
*/
private String handleCode;
/**
* 执行日志
*/
private String handleMsg;
/**
*告警状态 1- 默认无需警告 2-告警成功 3-告警失败
*/
private String alarmStatus;
/**
* 任务名称
*/
private String jobName;
/**
* 告警邮箱
*/
private String alarmEmail;
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/model/MythJobReadAhead.java deleted 100644 → 0
package com.byit.job.model;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.*;
import java.util.Date;
/**
* @program: byit-myth-job->MythJobReadAhead
* @description: 任务预读表映射实体，他作为任务流节点的快照，其实所有的修改操作都应该在这个类上做
* @author: huangfu
* @date: 2019/12/9 10:18
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
@EqualsAndHashCode
public class MythJobReadAhead {
/**
* 任务节点的id
*/
private String id;
/**
* 任务节点的名字
*/
private String jobName;
/**
* 插件任务调度的key,调度中心会根据这个key找到对应的插件端任务，执行
*/
private String executorHandler;
/**
* 任务周期调度cron表达式
*/
private String jobCron;
/**
* 任务详情，展示在调度中心平台的备注
*/
private String jobDesc;
/**
* 插件地址的集合
*/
private String pluginUrls;
/**
* 路由策略
* 1随机（默认）
* 2轮询
* 3最近最少使用
* 4最近最久未使用算法
* 5哈希算法
*/
private String routingStrategy;
/**
* 阻塞策略：
* 1丢弃
* 2阻塞等待（默认）
*/
private String blockingStrategy;
/**
* 回调时的身份认证
*/
private String callbackToken;
/**
* 请求调度中心的token
*/
private String gatewayToken;
/**
* 调度中心主机名
*/
private String requestHost;
/**
* 调度中心端口号
*/
private String requestPort;
/**
* 任务参数
*/
private String jobParam;
/**
* 任务类型，java,python,php,script,sql.shell
*/
private String jobType;
/**
* 所属任务流的id
*/
private String taskFlowId;
/**
* 上级节点
*/
private String parentId;
/**
* 是否跟随任务流的时间设置？1不跟随(默认),2跟随
*/
private String taskFlowIsCron;
/**
* 报警邮件
*/
private String alarmEmail;
/**
* 任务的超时时间
*/
private Integer executorTimeout;
/**
* 脚本文件地址(或远程文件服务器地址)支持多个，逗号分割
*/
private String sourceUrls;
/**
* 调度状态：0-暂停，1-运行
*/
private String triggerStatus;
/**
*调度任务源码
*/
private String glueSource;
/**
* 源码备注
*/
private String glueRemark;
/**
* 源码的修改时间
*/
@JSONField(format = "yyyy-MM-dd HH:mm:ss")
private Date glueUpdateTime;
/**
* 任务下次调度时间
*/
private Long triggerNextTime;
/**
* 任务创建者
*/
private String author;
/**
* 任务节点添加时间
*/
private Date addTime;
/**
* 任务节点修改时间
*/
private Date updateTime;
/**
* 任务流的唯一运行标识
*/
private String runID;
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/utils/SourceObj2TargetObjUtil.java deleted 100644 → 0
package com.byit.job.utils;
import com.byit.job.model.MythJobReadAhead;
import com.byit.job.dto.PluginBeanJobInfo;
import java.util.Date;
import java.util.UUID;
/**
* @program: byit-myth-job->SourceObj2TargetObjUtil
* @description: 这是一个转换的工具类，定义两个类之间的相互转换
* @author: huangfu
* @date: 2019/12/14 13:00
**/
public class SourceObj2TargetObjUtil {
/**
* 插件端的实体对象转换为对任务节点的对象
* @param pluginBeanJobInfo
* @return
*/
public static MythJobReadAhead pluginBeanJobInfo2MythJobReadAhead(PluginBeanJobInfo pluginBeanJobInfo){
MythJobReadAhead mythJobReadAhead = new MythJobReadAhead();
mythJobReadAhead.setId(UUID.randomUUID( ).toString().replace("-",""));
mythJobReadAhead.setExecutorHandler(pluginBeanJobInfo.getJobHandelName());
mythJobReadAhead.setPluginUrls(pluginBeanJobInfo.getUrl());
mythJobReadAhead.setJobCron(pluginBeanJobInfo.getMythCron());
mythJobReadAhead.setRoutingStrategy(pluginBeanJobInfo.getRoutingStrategy());
mythJobReadAhead.setBlockingStrategy(pluginBeanJobInfo.getBlockingStrategy());
mythJobReadAhead.setCallbackToken(pluginBeanJobInfo.getCallbackToken());
mythJobReadAhead.setGatewayToken(pluginBeanJobInfo.getGatewayToken());
mythJobReadAhead.setJobParam(pluginBeanJobInfo.getParam());
mythJobReadAhead.setAlarmEmail(pluginBeanJobInfo.getAlarmEmail());
mythJobReadAhead.setAuthor(pluginBeanJobInfo.getAuthor());
mythJobReadAhead.setAddTime(new Date());
mythJobReadAhead.setExecutorTimeout(30);
mythJobReadAhead.setJobDesc("这是一个测试任务，来自插件端的测试任务");
mythJobReadAhead.setJobName("测试任务");
mythJobReadAhead.setJobType("BEAN");
mythJobReadAhead.setTriggerStatus("1");
mythJobReadAhead.setParentId("1");
mythJobReadAhead.setTaskFlowIsCron("1");
//TODO 暂时随机分配 未来这个东西是线程自动添加的
mythJobReadAhead.setRunID(UUID.randomUUID( ).toString().replace("-",""));
mythJobReadAhead.setTriggerNextTime(System.currentTimeMillis()+20000);
return mythJobReadAhead;
}
}
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