Commit ed4219d0 by huangfusuper

重构折线图

parent cf84567e
......@@ -170,6 +170,8 @@ public class ApiFlowController {
return ResponseResult.ok("SUCCESS");
}
/**
* 获取整体运行的统计数据
* @param param
......
package com.byit.dto;
import lombok.Data;
@Data
public class StatisticsConditionDto {
private long startTime;
private long endTime;
private Integer workspaceId;
private String flowName;
}
......@@ -7,21 +7,23 @@ package com.byit.enums;
*/
public enum ScheduleStatusEnum {
UN_START(1, "未运行"),
STARTING(2, "运行中"),
FINISH(4, "暂停"),
STOP(3, "完成"),
UN_START("1", "未运行"),
STARTING("2", "运行中"),
FINISH("4", "成功"),
STOP("3", "暂停"),
ERROR("5", "失败"),
KILL("6", "杀死"),
;
private Integer code;
private String code;
private String msg;
private ScheduleStatusEnum(Integer code, String msg){
ScheduleStatusEnum(String code, String msg){
this.code = code;
this.msg = msg;
}
public Integer getCode(){
public String getCode(){
return this.code;
}
......
......@@ -37,13 +37,13 @@ public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunH
dateAligned(INIT_SLEEP_DATE,threadName);
log.info("---------------{}线程启动成功----------------",threadName);
while (!THREAD_GROUP_STOP){
dateAligned(CYCLE_INTERVAL,threadName);
//定义睡眠变量
Long sleepTime = 0L;
try{
//加锁
RedissLockUtil.lock(lockName,120);
log.debug("------------{},加锁成功,锁名称为{}----------",threadName,lockName);
dateAligned(CYCLE_INTERVAL,threadName);
//调用业务操作
sleepTime = examplesValue.start();
}catch (Exception e) {
......
package com.byit.mapper;
import com.byit.dto.FlowConditionDto;
import com.byit.dto.StatisticsConditionDto;
import com.byit.model.Flow;
import org.apache.ibatis.annotations.Param;
......@@ -8,6 +9,12 @@ import java.util.List;
public interface FlowMapper {
/**
* 查询当天将要运行的工作流
* @param statisticsConditionDto
* @return 所有的流信息
*/
List<Flow> findAllByThisDayFlow(StatisticsConditionDto statisticsConditionDto);
/**
* 查询全部的任务流数据
* @return
*/
......
package com.byit.mapper;
import com.byit.dto.StatisticsConditionDto;
import com.byit.model.FlowStatusSnapshoot;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;
@Repository
public interface FlowStatusSnapshootMapper {
List<FlowStatusSnapshoot> findAallByCon(StatisticsConditionDto statisticsConditionDto);
/**
* 删除当前时间的工作流快照
*/
void removeThisDateFlowStatusSnapshoot (@Param("thisDay") String thisDay,@Param("thisHour") String thisHour);
int deleteById(Integer statusId);
int insert(FlowStatusSnapshoot record);
......
package com.byit.mapper;
import com.byit.dto.FlowConditionDto;
import com.byit.dto.StatisticsConditionDto;
import com.byit.dto.plugin.StatisticData;
import com.byit.model.RunRecording;
import org.apache.ibatis.annotations.Param;
......@@ -15,6 +16,13 @@ import java.util.List;
*/
@Repository
public interface RunRecordingMapper {
/**
* 查询一个时间段的数据
* @param statisticsConditionDto 查询条件
* @return 一个时间段的数据
*/
List<RunRecording> findThisDayRunRecording(StatisticsConditionDto statisticsConditionDto);
/**
* 查询全部的数据
* @return
......
package com.byit.mapper;
import com.byit.dto.StatisticsConditionDto;
import com.byit.dto.plugin.StatisticData;
import com.byit.model.WaitingRecord;
import org.apache.ibatis.annotations.Param;
......@@ -13,6 +14,12 @@ import java.util.List;
@Repository
public interface WaitingRecordMapper {
/**
*
* @param statisticsConditionDto
* @return
*/
List<WaitingRecord> findAllByCondition(StatisticsConditionDto statisticsConditionDto);
/**
* 查询全部的工作流,同时根据执行时间查询对应的工作流数据
* @param triggerTime 本次的执行时间
* @return 即将执行的数据
......
......@@ -89,6 +89,9 @@ public class FlowStatusSnapshoot implements Serializable {
@ApiModelProperty("杀死的节点数目")
private int killNode;
@ApiModelProperty("工作流快照的时间戳")
private Long triggerTime;
/**
*/
private static final long serialVersionUID = 1L;
......
......@@ -117,6 +117,9 @@ public class WaitingRecord implements Serializable {
*/
@ApiModelProperty("补批时间")
private String repeatTime;
@ApiModelProperty("工作空间id")
private Integer workspaceId;
/**
*/
private static final long serialVersionUID = 1L;
......
......@@ -19,6 +19,13 @@ import java.util.List;
public interface FlowService {
/**
* 查询当天将要运行的工作流
* @param triggerNextTime 下次运行中
* @return 所有的流信息
*/
List<Flow> findAllByThisDayFlow(Long triggerNextTime);
/**
* 查询全部的任务流数据
* @return
*/
......
package com.byit.service;
import com.byit.model.FlowStatusSnapshoot;
import java.util.List;
/**
* 工作流快照业务接口
* @author huangfu
*/
public interface FlowStatusSnapshootService {
/**
* 删除和保存新的工作流快照
* @param allFlow
*/
void removeAndSaveSnapshoot(List<FlowStatusSnapshoot> allFlow);
}
......@@ -11,6 +11,14 @@ import java.util.List;
* @author huangfu
*/
public interface RunRecordingService {
/**
* 查询当天的实例 查询一个时间段的范围
* @param startTime 开始时间
* @param endTime 结束时间
* @return 一个时间段的数据
*/
List<RunRecording> findThisDayRunRecording(Long startTime,Long endTime);
/**
* 查询全部数据 映射成VO
* @return
......
......@@ -4,6 +4,7 @@ import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.byit.dto.FlowConditionDto;
import com.byit.dto.StatisticsConditionDto;
import com.byit.dto.web.ResponseResult;
import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.NodePropertyEnum;
......@@ -53,6 +54,19 @@ public class FlowServiceImpl implements FlowService {
private WorkspaceMapper workspaceMapper;
/**
* 查询当天所有待运行的任务流
* @param triggerNextTime 下次运行中
* @return
*/
@Override
public List<Flow> findAllByThisDayFlow(Long triggerNextTime) {
StatisticsConditionDto statisticsConditionDto = new StatisticsConditionDto();
statisticsConditionDto.setStartTime(0L);
statisticsConditionDto.setEndTime(triggerNextTime);
return flowMapper.findAllByThisDayFlow(statisticsConditionDto);
}
@Override
public List<FlowViewVo> findAllFlowViewVo(FlowConditionDto flowConditionDto) {
List<Flow> allFlow = findAllFlow(flowConditionDto);
......
package com.byit.service.impl;
import cn.hutool.core.date.DateUnit;
import com.byit.job.utils.DateUtil;
import com.byit.mapper.FlowStatusSnapshootMapper;
import com.byit.model.FlowStatusSnapshoot;
import com.byit.service.FlowStatusSnapshootService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.Date;
import java.util.List;
/**
* 工作流快照操作
*/
@Service
@Slf4j
@Transactional(rollbackFor = Exception.class)
public class FlowStatusSnapshootServiceImpl implements FlowStatusSnapshootService {
private final FlowStatusSnapshootMapper flowStatusSnapshootMapper;
public FlowStatusSnapshootServiceImpl(FlowStatusSnapshootMapper flowStatusSnapshootMapper) {
this.flowStatusSnapshootMapper = flowStatusSnapshootMapper;
}
@Override
public void removeAndSaveSnapshoot(List<FlowStatusSnapshoot> allFlow) {
//获取当天的日期
String thisDateStr = DateUtil.dateFormat(new Date(), "yyyyMMdd");
String thisHourStr = DateUtil.dateFormat(new Date(), "HH");
String hourBefStr = String.valueOf(Integer.parseInt(thisHourStr)+1);
//20201212 12:12:12
Date date = DateUtil.strFormatDate(thisDateStr + " " + hourBefStr + ":00:00", "yyyyMMdd HH:mm:dd");
long between = cn.hutool.core.date.DateUtil.between(new Date(), date, DateUnit.MINUTE);
if (between > 3) {
log.info("--------开始删除当前点数的工作流快照----------");
flowStatusSnapshootMapper.removeThisDateFlowStatusSnapshoot(thisDateStr,thisHourStr);
log.info("------删除成功,保存新的工作流快照---------");
flowStatusSnapshootMapper.saveList(allFlow);
}else {
log.info("---------------------------距离下个小时数小于三分钟,跳过本次快照记录-----------------");
}
}
}
package com.byit.service.impl;
import com.byit.dto.FlowConditionDto;
import com.byit.dto.StatisticsConditionDto;
import com.byit.enums.RunRecordingEnum;
import com.byit.mapper.RunRecordingMapper;
import com.byit.model.RunRecording;
import com.byit.model.vo.RunRecordingViewVo;
import com.byit.service.RunRecordingService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.List;
import java.util.UUID;
......@@ -21,6 +24,8 @@ import java.util.stream.Collectors;
* @date: 2019/12/26 15:08
**/
@Service
@Transactional(rollbackFor = Exception.class)
@Slf4j
public class RunRecordingServiceImpl implements RunRecordingService {
private final RunRecordingMapper runRecordingMapper;
......@@ -29,6 +34,21 @@ public class RunRecordingServiceImpl implements RunRecordingService {
this.runRecordingMapper = runRecordingMapper;
}
/**
* 查询当天的数据
* @param startTime 开始时间
* @param endTime 结束时间
* @return
*/
@Override
public List<RunRecording> findThisDayRunRecording(Long startTime, Long endTime) {
log.debug("--------------查询时间范围为:{} --> {} 的数据--------------",startTime,endTime);
StatisticsConditionDto statisticsConditionDto = new StatisticsConditionDto();
statisticsConditionDto.setStartTime(startTime);
statisticsConditionDto.setEndTime(endTime);
return runRecordingMapper.findThisDayRunRecording(statisticsConditionDto);
}
@Override
public List<RunRecordingViewVo> findAllRunRecordingViewVo(FlowConditionDto flowConditionDt) {
List<RunRecording> runRecordings = findAll(flowConditionDt);
......
......@@ -102,7 +102,7 @@ public class ScriptExecutorJobTask implements TimerTask {
scriptDto.setRunId(mythJobTaskSchedule.getRunId());
scriptDto.setRemotePath(mythJobTaskSchedule.getScriptUrls());
scriptDto.setCallbackUrl(HTTP_PRE+ ServiceInfoUtil.getIpAndPort()+HTTP_SUFFIX);
//二次执行的情况下 会有这个信息
//二次执行的情况下 会有这个信息 //TODO 还是有问题
scriptDto.setLogRemotePath(jobTaskRunLogById.getLogRemotelyPath());
DispatchResponseDto dispatchResponseDto = new DispatchResponseDto();
try{
......
package com.byit.thread.helper;
import com.byit.dto.plugin.StatisticData;
import com.byit.enums.ScheduleStatusEnum;
import com.byit.mapper.*;
import com.byit.model.*;
import com.byit.mapper.FlowMapper;
import com.byit.mapper.FlowStatusSnapshootMapper;
import com.byit.mapper.JobTaskRunLogMapper;
import com.byit.mapper.RunRecordingMapper;
import com.byit.model.Flow;
import com.byit.model.FlowStatusSnapshoot;
import com.byit.model.RunRecording;
import com.byit.thread.BaseThreadRunHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
......@@ -13,7 +17,10 @@ import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import javax.sql.DataSource;
import java.time.*;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Date;
......
......@@ -34,6 +34,23 @@
schedule_follow, is_update, scan_mark
</sql>
<select id="findAllByThisDayFlow" resultMap="BaseResultMap" parameterType="com.byit.dto.StatisticsConditionDto">
select
<include refid="Base_Column_List" />
from flow
where
trigger_next_time between #{startTime} and #{endTime}
and remaining_count != 0
and start_up = '0'
<if test="workspaceId != null">
and workspace_id = #{workspaceId}
</if>
<if test="flowName != null and flowName != ''">
flow_name = #{flowName}
</if>
</select>
<select id="findAllFlow" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
......
......@@ -16,12 +16,32 @@
<result column="success_node" jdbcType="INTEGER" property="successNode" />
<result column="fail_node" jdbcType="INTEGER" property="failNode" />
<result column="kill_node" jdbcType="INTEGER" property="killNode" />
<result column="trigger_time" jdbcType="BIGINT" property="triggerTime" />
</resultMap>
<sql id="Base_Column_List">
<!-- generated @mbg.generated date: 2020-04-02 -->
status_id, workspace_id, flow_id, flow_name, `day`, `hour`, flow_status, snapshoot_time,
unstart_node, runing_node, success_node, fail_node, kill_node
unstart_node, runing_node, success_node, fail_node, kill_node, trigger_time
</sql>
<select id="findAallByCon" resultMap="BaseResultMap" parameterType="com.byit.dto.StatisticsConditionDto">
select
<include refid="Base_Column_List" />
from flow_status_snapshoot
where trigger_time between #{startTime} and #{endTime}
<if test="workspaceId != null">
and workspace_id = #{workspaceId}
</if>
<if test="flowName != null and flowName != ''">
flow_name = #{flowName}
</if>
</select>
<delete id="removeThisDateFlowStatusSnapshoot">
delete from flow_status_snapshoot where day=#{thisDay} and hour=#{thisHour}
</delete>
<select id="getById" parameterType="java.lang.Integer" resultMap="BaseResultMap">
<!-- generated @mbg.generated date: 2020-04-02 -->
select
......@@ -40,12 +60,12 @@
flow_name, `day`, `hour`,
flow_status, snapshoot_time, unstart_node,
runing_node, success_node, fail_node,
kill_node)
kill_node,trigger_time)
values (#{statusId,jdbcType=INTEGER}, #{workspaceId,jdbcType=INTEGER}, #{flowId,jdbcType=INTEGER},
#{flowName,jdbcType=VARCHAR}, #{day,jdbcType=VARCHAR}, #{hour,jdbcType=VARCHAR},
#{flowStatus,jdbcType=INTEGER}, #{snapshootTime,jdbcType=BIGINT}, #{unstartNode,jdbcType=INTEGER},
#{runingNode,jdbcType=INTEGER}, #{successNode,jdbcType=INTEGER}, #{failNode,jdbcType=INTEGER},
#{killNode,jdbcType=INTEGER})
#{killNode,jdbcType=INTEGER},#{triggerTime,jdbcType=BIGINT})
</insert>
<insert id="insertSelective" parameterType="com.byit.model.FlowStatusSnapshoot">
<!-- generated @mbg.generated date: 2020-04-02 -->
......@@ -173,6 +193,9 @@
<if test="killNode != null">
kill_node = #{killNode,jdbcType=INTEGER},
</if>
<if test="triggerTime != null">
trigger_time = #{triggerTime,jdbcType=BIGINT},
</if>
</set>
where status_id = #{statusId,jdbcType=INTEGER}
</update>
......@@ -190,7 +213,8 @@
runing_node = #{runingNode,jdbcType=INTEGER},
success_node = #{successNode,jdbcType=INTEGER},
fail_node = #{failNode,jdbcType=INTEGER},
kill_node = #{killNode,jdbcType=INTEGER}
kill_node = #{killNode,jdbcType=INTEGER},
trigger_time = #{triggerTime,jdbcType=BIGINT},
where status_id = #{statusId,jdbcType=INTEGER}
</update>
<select id="exist" resultType="java.lang.Integer">
......@@ -228,7 +252,7 @@
flow_name, `day`, `hour`,
flow_status, snapshoot_time, unstart_node,
runing_node, success_node, fail_node,
kill_node
kill_node,trigger_time
)
VALUES
<foreach collection="flowStatusSnapshootList" item="flowStatusSnapshoot" separator=",">
......@@ -244,7 +268,8 @@
#{flowStatusSnapshoot.runingNode,jdbcType=INTEGER},
#{flowStatusSnapshoot.successNode,jdbcType=INTEGER},
#{flowStatusSnapshoot.failNode,jdbcType=INTEGER},
#{flowStatusSnapshoot.killNode,jdbcType=INTEGER}
#{flowStatusSnapshoot.killNode,jdbcType=INTEGER},
#{flowStatusSnapshoot.triggerTime,jdbcType=BIGINT},
)
</foreach>
</insert>
......
......@@ -35,6 +35,21 @@
,workspace_id, repeat_time
</sql>
<select id="findThisDayRunRecording" resultMap="BaseResultMap" parameterType="com.byit.dto.StatisticsConditionDto">
select
<include refid="Base_Column_List" />
from run_recording
where trigger_time between #{startTime} and #{endTime}
<if test="workspaceId != null">
and workspace_id = #{workspaceId}
</if>
<if test="flowName != null and flowName != ''">
flow_name = #{flowName}
</if>
</select>
<select id="findAll" parameterType="com.byit.dto.FlowConditionDto" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
......
......@@ -20,14 +20,31 @@
<result column="wait_order" jdbcType="INTEGER" property="waitOrder" />
<result column="run_id" jdbcType="VARCHAR" property="runId" />
<result column="repeat_time" jdbcType="VARCHAR" property="repeatTime" />
<result column="workspace_id" jdbcType="INTEGER" property="workspaceId" />
</resultMap>
<sql id="Base_Column_List">
<!-- generated @mbg.generated date: 2020-03-12 -->
wait_id, flow_id, flow_name, flow_version_name, flow_timeout, alarm_email, alarml_action,
priority, trigger_time, principal, is_inner, flow_node_count, schedule_type, `operator`,
wait_order, run_id, repeat_time
wait_order, run_id, repeat_time, workspace_id
</sql>
<select id="findAllByCondition" resultMap="BaseResultMap" parameterType="com.byit.dto.StatisticsConditionDto">
select
<include refid="Base_Column_List" />
from waiting_record
where trigger_time
between #{startTime} and #{endTime}
<if test="workspaceId != null">
and workspace_id = #{workspaceId}
</if>
<if test="flowName != null and flowName != ''">
flow_name = #{flowName}
</if>
</select>
<select id="findAllByTriggerTime" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
......@@ -129,6 +146,9 @@
<if test="repeatTime != null">
repeat_time,
</if>
<if test="workspaceId != null">
workspace_id,
</if>
</trim>
<trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="waitId != null">
......@@ -182,6 +202,9 @@
<if test="repeatTime != null">
#{repeatTime,jdbcType=VARCHAR},
</if>
<if test="workspaceId != null">
#{workspaceId,jdbcType=INTEGER},
</if>
</trim>
</insert>
<update id="updateByIdSelective" parameterType="com.byit.model.WaitingRecord">
......@@ -236,6 +259,9 @@
<if test="repeatTime != null">
repeat_time = #{repeatTime,jdbcType=VARCHAR},
</if>
<if test="workspaceId != null">
workspace_id = #{workspaceId,jdbcType=INTEGER},
</if>
</set>
where wait_id = #{waitId,jdbcType=INTEGER}
</update>
......
......@@ -21,7 +21,7 @@ public class DateUtil {
* @param formatStr 转换格式
* @return 标准的时间格式
*/
public static String dateFormat(Date date,@NotNull String formatStr){
public static String dateFormat(Date date, String formatStr){
LocalDateTime localDateTime = dateToLocalDateTime(date);
String dateFormat;
if(StringUtils.isNotBlank(formatStr)){
......@@ -134,6 +134,18 @@ public class DateUtil {
}
/**
* 字符串转换时间
* @param dateStr 标准时间字符串
* @param dateFormat 时间字符串的格式
* @return 标准时间类型
*/
public static Date strFormatDate(String dateStr,String dateFormat){
DateTimeFormatter dtf = DateTimeFormatter.ofPattern(dateFormat);
LocalDateTime parse = LocalDateTime.parse(dateStr, dtf);
return localDateTimeToDate(parse);
}
/**
* 时间运算 天数相减
* @param date 标准的时间格式
* @param day 相加天数
......@@ -148,16 +160,6 @@ public class DateUtil {
/**
* 字符串转换时间
* @param dateStr 标准时间字符串
* @return 标准时间类型
*/
public static Date strToDate(String dateStr){
LocalDateTime parse = LocalDateTime.parse(dateStr);
return localDateTimeToDate(parse);
}
/**
* date转换为java8的localDate
* @param date 标准时间
* @return java8的时间类型
......@@ -192,8 +194,32 @@ public class DateUtil {
return Date.from(zonedDateTime.toInstant());
}
/**
* 获取当天结束时间 xxxx-xx-xx 23:59:59
* @return
*/
public static Date getThisDayEndDate(){
LocalDateTime todayEnd = LocalDateTime.of(LocalDate.now(), LocalTime.MAX);
return localDateTimeToDate(todayEnd);
}
// public static void main(String[] args) {
// System.out.println(DateUtil.dateLessDayStr(new Date(), "yyyyMMddHHmmss", 1));
// }
public static Date getThisDayStartDate() {
//当天零点
LocalDateTime todayStart = LocalDateTime.of(LocalDate.now(), LocalTime.MIN);
return localDateTimeToDate(todayStart);
}
public static void main(String[] args) {
String hh = DateUtil.dateFormat(new Date(), "HH");
System.out.println(hh);
String thisDateStr = DateUtil.dateFormat(new Date(), "yyyyMMdd");
String test = thisDateStr + " " + hh + ":00:00";
System.out.println(test);
Date date = DateUtil.strFormatDate(test, "yyyyMMdd HH:mm:ss");
System.out.println(DateUtil.dateFormat(date,"yyyy-MM-dd HH:mm:ss"));
}
}
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