gtsoft/gtsoft-snail-job-server
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects 1 Releases Wiki Activity

feat: 2.0.0

Browse Source 
1. 优化日志存储逻辑
2. 新增日志保存时间配置
3. 新增定时清除日志功能
4. 优化看板页面日志的查询统计
5. 新增日志详情页查询调度日志列表
6. 优化部分代码
This commit is contained in:
byteblogs168 2023-06-16 17:38:19 +08:00
parent a2729dea15
commit 3e52fcf5a8
40 changed files with 628 additions and 231 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
doc/sql/easy_retry.sql
View File

@ -49,6 +49,7 @@ CREATE TABLE `retry_dead_letter_0`
KEY `idx_group_name_scene_name` (`group_name`, `scene_name`),
KEY `idx_idempotent_id` (`idempotent_id`),
KEY `idx_biz_no` (`biz_no`),
KEY `idx_create_dt` (`create_dt`),
UNIQUE KEY `uk_name_unique_id` (`group_name`, `unique_id`)
) ENGINE=InnoDB AUTO_INCREMENT=0 DEFAULT CHARSET=utf8mb4 COMMENT='重试死信队列'
;
@ -75,6 +76,7 @@ CREATE TABLE `retry_task_0`
KEY `idx_retry_status` (`retry_status`),
KEY `idx_idempotent_id` (`idempotent_id`),
KEY `idx_biz_no` (`biz_no`),
KEY `idx_create_dt` (`create_dt`),
UNIQUE KEY `uk_name_unique_id` (`group_name`, `unique_id`)
) ENGINE=InnoDB AUTO_INCREMENT=0 DEFAULT CHARSET=utf8mb4 COMMENT='重试表'
;
@ -90,18 +92,30 @@ CREATE TABLE `retry_task_log`
`executor_name` varchar(512) NOT NULL DEFAULT '' COMMENT '执行器名称',
`args_str` text NOT NULL COMMENT '执行方法参数',
`ext_attrs` text NOT NULL COMMENT '扩展字段',
`retry_status` tinyint(4) NOT NULL DEFAULT '0' COMMENT '重试状态 0、失败 1、成功',
`retry_status` tinyint(4) NOT NULL DEFAULT '0' COMMENT '重试状态 0、重试中 1、成功 2、最大次数',
`task_type` tinyint(4) NOT NULL DEFAULT '1' COMMENT '任务类型 1、重试数据 2、回调数据',
`create_dt` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`update_dt` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',
`error_message` text NOT NULL COMMENT '异常信息',
PRIMARY KEY (`id`),
`create_dt` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间'
PRIMARY KEY (`id`),
KEY `idx_group_name_scene_name` (`group_name`, `scene_name`),
KEY `idx_retry_status` (`retry_status`),
KEY `idx_idempotent_id` (`idempotent_id`),
KEY `idx_unique_id` (`unique_id`),
KEY `idx_biz_no` (`biz_no`)
) ENGINE=InnoDB AUTO_INCREMENT=0 DEFAULT CHARSET=utf8mb4 COMMENT='重试日志表'
KEY `idx_biz_no` (`biz_no`),
KEY `idx_create_dt` (`create_dt`)
) ENGINE=InnoDB AUTO_INCREMENT=0 DEFAULT CHARSET=utf8mb4 COMMENT='重试日志基础信息表'
;
CREATE TABLE `retry_task_log_message`
(
`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
`group_name` varchar(64) NOT NULL COMMENT '组名称',
`unique_id` varchar(64) NOT NULL COMMENT '同组下id唯一',
`create_dt` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`message` text NOT NULL COMMENT '异常信息',
PRIMARY KEY (`id`),
KEY `idx_group_name_unique_id` (`group_name`, `unique_id`),
KEY `idx_create_dt` (`create_dt`)
) ENGINE=InnoDB AUTO_INCREMENT=0 DEFAULT CHARSET=utf8mb4 COMMENT='重试日志异常信息记录表'
;
CREATE TABLE `scene_config`

2
easy-retry-client-core/src/main/java/com/aizuda/easy/retry/client/core/client/RetryEndPoint.java
View File

@ -140,7 +140,7 @@ public class RetryEndPoint {
retryCompleteCallback.doSuccessCallback(retryerInfo.getScene(), retryerInfo.getExecutorClassName(), deSerialize);
}
if (RetryStatusEnum.MAX_RETRY_COUNT.getStatus().equals(callbackDTO.getRetryStatus())) {
if (RetryStatusEnum.MAX_COUNT.getStatus().equals(callbackDTO.getRetryStatus())) {
retryCompleteCallback.doMaxRetryCallback(retryerInfo.getScene(), retryerInfo.getExecutorClassName(), deSerialize);
}

8
easy-retry-client-core/src/main/java/com/aizuda/easy/retry/client/core/intercepter/RetryAspect.java
View File

@ -11,14 +11,12 @@ import com.aizuda.easy.retry.client.core.annotation.Retryable;
import com.aizuda.easy.retry.client.core.retryer.RetryerResultContext;
import com.aizuda.easy.retry.common.core.alarm.Alarm;
import com.aizuda.easy.retry.common.core.alarm.AlarmContext;
import com.aizuda.easy.retry.common.core.alarm.AltinAlarmFactory;
import com.aizuda.easy.retry.common.core.constant.SystemConstants;
import com.aizuda.easy.retry.common.core.alarm.EasyRetryAlarmFactory;
import com.aizuda.easy.retry.common.core.enums.NotifySceneEnum;
import com.aizuda.easy.retry.common.core.enums.RetryResultStatusEnum;
import com.aizuda.easy.retry.common.core.log.LogUtils;
import com.aizuda.easy.retry.common.core.model.EasyRetryHeaders;
import com.aizuda.easy.retry.common.core.util.EnvironmentUtils;
import com.aizuda.easy.retry.common.core.util.JsonUtil;
import com.aizuda.easy.retry.server.model.dto.ConfigDTO;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
@ -58,7 +56,7 @@ public class RetryAspect implements Ordered {
@Qualifier("localRetryStrategies")
private RetryStrategy retryStrategy;
@Autowired
private AltinAlarmFactory altinAlarmFactory;
private EasyRetryAlarmFactory easyRetryAlarmFactory;
@Autowired
private StandardEnvironment standardEnvironment;
@ -186,7 +184,7 @@ public class RetryAspect implements Ordered {
.title("retry component handling exception:[{}]", EasyRetryProperties.getGroup())
.notifyAttribute(notifyAttribute.getNotifyAttribute());
Alarm<AlarmContext> alarmType = altinAlarmFactory.getAlarmType(notifyAttribute.getNotifyType());
Alarm<AlarmContext> alarmType = easyRetryAlarmFactory.getAlarmType(notifyAttribute.getNotifyType());
alarmType.asyncSendMessage(context);
}
} catch (Exception e1) {

6
easy-retry-client-core/src/main/java/com/aizuda/easy/retry/client/core/report/ReportListener.java
View File

@ -12,7 +12,7 @@ import com.aizuda.easy.retry.client.core.cache.GroupVersionCache;
import com.aizuda.easy.retry.client.core.executor.GuavaRetryExecutor;
import com.aizuda.easy.retry.common.core.alarm.Alarm;
import com.aizuda.easy.retry.common.core.alarm.AlarmContext;
import com.aizuda.easy.retry.common.core.alarm.AltinAlarmFactory;
import com.aizuda.easy.retry.common.core.alarm.EasyRetryAlarmFactory;
import com.aizuda.easy.retry.common.core.context.SpringContext;
import com.aizuda.easy.retry.common.core.enums.NotifySceneEnum;
import com.aizuda.easy.retry.common.core.log.LogUtils;
@ -125,8 +125,8 @@ public class ReportListener implements Listener<RetryTaskDTO> {
.title("上报异常:[{}]", EasyRetryProperties.getGroup())
.notifyAttribute(notifyAttribute.getNotifyAttribute());
AltinAlarmFactory altinAlarmFactory = SpringContext.getBeanByType(AltinAlarmFactory.class);
Alarm<AlarmContext> alarmType = altinAlarmFactory.getAlarmType(notifyAttribute.getNotifyType());
EasyRetryAlarmFactory easyRetryAlarmFactory = SpringContext.getBeanByType(EasyRetryAlarmFactory.class);
Alarm<AlarmContext> alarmType = easyRetryAlarmFactory.getAlarmType(notifyAttribute.getNotifyType());
alarmType.asyncSendMessage(context);
}
} catch (Exception e1) {

4
easy-retry-common/easy-retry-common-core/src/main/java/com/aizuda/easy/retry/common/core/alarm/AltinAlarmFactory.java → easy-retry-common/easy-retry-common-core/src/main/java/com/aizuda/easy/retry/common/core/alarm/EasyRetryAlarmFactory.java
View File

@ -11,12 +11,12 @@ import java.util.concurrent.ConcurrentHashMap;
* @date : 2021-11-25 09:20
*/
@Component
public class AltinAlarmFactory {
public class EasyRetryAlarmFactory {
private final Map<Integer, Alarm> alarmMap = new ConcurrentHashMap<>();
@Autowired
public AltinAlarmFactory(Map<String, Alarm> map) {
public EasyRetryAlarmFactory(Map<String, Alarm> map) {
for (Map.Entry<String, Alarm> entry : map.entrySet()) {
alarmMap.put(entry.getValue().getAlarmType(), entry.getValue());
}

15
easy-retry-common/easy-retry-common-core/src/main/java/com/aizuda/easy/retry/common/core/constant/SystemConstants.java
View File

@ -66,19 +66,4 @@ public interface SystemConstants {
}
interface CALL_BACK {
/**
* 回调id前缀
*/
String CB_ = "CB_";
/**
* 最大重试次数
*/
int MAX_RETRY_COUNT = 288;
/**
* 间隔时间
*/
int TRIGGER_INTERVAL = 15 * 60;
}
}

4
easy-retry-common/easy-retry-common-core/src/main/java/com/aizuda/easy/retry/common/core/enums/RetryStatusEnum.java
View File

@ -25,9 +25,9 @@ public enum RetryStatusEnum {
FINISH(1),
/**
* 到达最大重试次数
* 到达最大次数
*/
MAX_RETRY_COUNT(2),
MAX_COUNT(2),
/**
* 暂停重试

25
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/akka/ActorGenerator.java
View File

@ -2,15 +2,15 @@ package com.aizuda.easy.retry.server.akka;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import com.aizuda.easy.retry.common.core.context.SpringContext;
import com.aizuda.easy.retry.server.support.dispatch.actor.exec.ExecCallbackUnitActor;
import com.aizuda.easy.retry.server.support.dispatch.actor.exec.ExecUnitActor;
import com.aizuda.easy.retry.server.support.dispatch.actor.log.LogActor;
import com.aizuda.easy.retry.server.support.dispatch.actor.result.FailureActor;
import com.aizuda.easy.retry.server.support.dispatch.actor.result.FinishActor;
import com.aizuda.easy.retry.server.support.dispatch.actor.result.NoRetryActor;
import com.aizuda.easy.retry.server.support.dispatch.actor.scan.AbstractScanGroup;
import com.aizuda.easy.retry.server.support.dispatch.actor.scan.ScanCallbackGroupActor;
import com.aizuda.easy.retry.server.support.dispatch.actor.scan.ScanGroupActor;
import com.aizuda.easy.retry.common.core.context.SpringContext;
/**
* Actor生成器
@ -50,7 +50,7 @@ public class ActorGenerator {
}
/**
* 不触发重试actor
* 回调处理
*
* @return actor 引用
*/
@ -85,6 +85,16 @@ public class ActorGenerator {
return getDispatchRetryActorSystem().actorOf(getSpringExtension().props(ScanCallbackGroupActor.BEAN_NAME));
}
/**
* 生成扫描重试数据的actor
*
* @return actor 引用
*/
public static ActorRef logActor() {
return getLogActorSystemSystem().actorOf(getSpringExtension().props(LogActor.BEAN_NAME));
}
public static SpringExtension getSpringExtension() {
return SpringContext.getBeanByType(SpringExtension.class);
}
@ -104,6 +114,7 @@ public class ActorGenerator {
public static ActorSystem getDispatchExecUnitActorSystem() {
return SpringContext.getBean("dispatchExecUnitActorSystem", ActorSystem.class);
}
/**
* 重试任务结果分发器
* @return
@ -112,4 +123,12 @@ public class ActorGenerator {
return SpringContext.getBean("dispatchResultActorSystem", ActorSystem.class);
}
/**
* 日志记录分发器
*
* @return
*/
public static ActorSystem getLogActorSystemSystem() {
return SpringContext.getBean("logActorSystem", ActorSystem.class);
}
}

13
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/akka/AkkaConfiguration.java
View File

@ -18,6 +18,7 @@ public class AkkaConfiguration {
private static final String DISPATCH_RETRY_ACTOR_SYSTEM = "DISPATCH_RETRY_ACTOR_SYSTEM";
private static final String DISPATCH_EXEC_UNIT_RETRY_ACTOR_SYSTEM = "DISPATCH_EXEC_UNIT_RETRY_ACTOR_SYSTEM";
private static final String DISPATCH_RESULT_ACTOR_SYSTEM = "DISPATCH_RESULT_ACTOR_SYSTEM";
private static final String LOG_ACTOR_SYSTEM = "LOG_ACTOR_SYSTEM";
@Autowired
private ApplicationContext applicationContext;
@ -60,4 +61,16 @@ public class AkkaConfiguration {
springExtension.initialize(applicationContext);
return system;
}
/**
* 日志处理
*
* @return {@link ActorSystem} 顶级actor
*/
@Bean("logActorSystem")
public ActorSystem createLogActorSystem() {
ActorSystem system = ActorSystem.create(LOG_ACTOR_SYSTEM);
springExtension.initialize(applicationContext);
return system;
}
}

34
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/config/SystemProperties.java
View File

@ -46,4 +46,38 @@ public class SystemProperties {
*/
private int step = 100;
/**
* 日志默认保存天数
*/
private int logStorage = 90;
/**
* 回调配置
*/
private Callback callback = new Callback();
/**
* 回调配置
*/
@Data
public static class Callback {
/**
* 回调id前缀
*/
String prefix = "CB_";
/**
* 回调的最大执行次数
*/
private int maxCount = 288;
/**
* 间隔时间
*/
private long triggerInterval = 15 * 60;
}
}

1
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/persistence/mybatis/mapper/RetryTaskLogMapper.java
View File

@ -11,7 +11,6 @@ import java.util.List;
public interface RetryTaskLogMapper extends BaseMapper<RetryTaskLog> {
int deleteByPrimaryKey(Long id);
RetryTaskLog selectByPrimaryKey(Long id);

18
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/persistence/mybatis/mapper/RetryTaskLogMessageMapper.java Normal file
View File

@ -0,0 +1,18 @@
package com.aizuda.easy.retry.server.persistence.mybatis.mapper;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLogMessage;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import org.apache.ibatis.annotations.Mapper;
/**
* <p>
* 重试日志异常信息记录表 Mapper 接口
* </p>
*
* @author www.byteblogs.com
* @since 2023-06-16
*/
@Mapper
public interface RetryTaskLogMessageMapper extends BaseMapper<RetryTaskLogMessage> {
}

2
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/persistence/mybatis/po/RetryTaskLog.java
View File

@ -33,8 +33,6 @@ public class RetryTaskLog implements Serializable {
private Integer taskType;
private String errorMessage;
private LocalDateTime createDt;
private static final long serialVersionUID = 1L;

52
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/persistence/mybatis/po/RetryTaskLogMessage.java Normal file
View File

@ -0,0 +1,52 @@
package com.aizuda.easy.retry.server.persistence.mybatis.po;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Getter;
import lombok.Setter;
import java.io.Serializable;
import java.time.LocalDateTime;
/**
* <p>
* 重试日志异常信息记录表
* </p>
*
* @author www.byteblogs.com
* @since 2023-06-16
*/
@Getter
@Setter
@TableName("retry_task_log_message")
public class RetryTaskLogMessage implements Serializable {
private static final long serialVersionUID = 1L;
/**
* 主键
*/
@TableId(value = "id", type = IdType.AUTO)
private Long id;
/**
* 组名称
*/
private String groupName;
/**
* 同组下id唯一
*/
private String uniqueId;
/**
* 创建时间
*/
private LocalDateTime createDt;
/**
* 异常信息
*/
private String message;
}

4
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/service/RetryTaskLogService.java
View File

@ -1,7 +1,9 @@
package com.aizuda.easy.retry.server.service;
import com.aizuda.easy.retry.server.web.model.base.PageResult;
import com.aizuda.easy.retry.server.web.model.request.RetryTaskLogMessageQueryVO;
import com.aizuda.easy.retry.server.web.model.request.RetryTaskLogQueryVO;
import com.aizuda.easy.retry.server.web.model.response.RetryTaskLogMessageResponseVO;
import com.aizuda.easy.retry.server.web.model.response.RetryTaskLogResponseVO;
import java.util.List;
@ -15,6 +17,8 @@ public interface RetryTaskLogService {
PageResult<List<RetryTaskLogResponseVO>> getRetryTaskLogPage(RetryTaskLogQueryVO queryVO);
PageResult<List<RetryTaskLogMessageResponseVO>> getRetryTaskLogMessagePage(RetryTaskLogMessageQueryVO queryVO);
RetryTaskLogResponseVO getRetryTaskLogById(Long id);
}

8
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/service/convert/RetryTaskConverter.java
View File

@ -1,9 +1,12 @@
package com.aizuda.easy.retry.server.service.convert;
import com.aizuda.easy.retry.server.model.dto.RetryTaskDTO;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryDeadLetter;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTask;
import com.aizuda.easy.retry.server.web.model.request.RetryTaskSaveRequestVO;
import org.mapstruct.Mapper;
import org.mapstruct.Mapping;
import org.mapstruct.Mappings;
import org.mapstruct.factory.Mappers;
import java.util.List;
@ -21,6 +24,11 @@ public interface RetryTaskConverter {
RetryTask toRetryTask(RetryTask retryTask);
@Mappings({
@Mapping(target = "id", ignore = true),
})
RetryTask toRetryTask(RetryDeadLetter retryDeadLetter);
RetryTask toRetryTask(RetryTaskSaveRequestVO retryTaskSaveRequestVO);
List<RetryTask> toRetryTaskList(List<RetryTaskDTO> retryTaskDTOList);

5
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/service/convert/RetryTaskLogConverter.java
View File

@ -3,6 +3,8 @@ package com.aizuda.easy.retry.server.service.convert;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTask;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import org.mapstruct.Mapper;
import org.mapstruct.Mapping;
import org.mapstruct.Mappings;
import org.mapstruct.factory.Mappers;
/**
@ -14,5 +16,8 @@ public interface RetryTaskLogConverter {
RetryTaskLogConverter INSTANCE = Mappers.getMapper(RetryTaskLogConverter.class);
@Mappings({
@Mapping(target = "id", ignore = true),
})
RetryTaskLog toRetryTask(RetryTask retryTask);
}

4
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/service/convert/RetryTaskLogResponseVOConverter.java
View File

@ -1,6 +1,8 @@
package com.aizuda.easy.retry.server.service.convert;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLogMessage;
import com.aizuda.easy.retry.server.web.model.response.RetryTaskLogMessageResponseVO;
import com.aizuda.easy.retry.server.web.model.response.RetryTaskLogResponseVO;
import org.mapstruct.Mapper;
import org.mapstruct.factory.Mappers;
@ -19,4 +21,6 @@ public interface RetryTaskLogResponseVOConverter {
RetryTaskLogResponseVO convert(RetryTaskLog retryTaskLog);
List<RetryTaskLogResponseVO> batchConvert(List<RetryTaskLog> retryTaskLogs);
List<RetryTaskLogMessageResponseVO> toRetryTaskLogMessageResponseVO(List<RetryTaskLogMessage> retryTaskLogs);
}

14
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/service/impl/DashBoardServiceImpl.java
View File

@ -1,8 +1,11 @@
package com.aizuda.easy.retry.server.service.impl;
import com.aizuda.easy.retry.server.enums.TaskTypeEnum;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMessageMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.ServerNodeMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLogMessage;
import com.aizuda.easy.retry.server.persistence.mybatis.po.ServerNode;
import com.aizuda.easy.retry.server.service.convert.ServerNodeResponseVOConverter;
import com.aizuda.easy.retry.server.web.model.base.PageResult;
@ -43,6 +46,9 @@ public class DashBoardServiceImpl implements DashBoardService {
@Autowired
private RetryTaskLogMapper retryTaskLogMapper;
@Autowired
private RetryTaskLogMessageMapper retryTaskLogMessageMapper;
@Autowired
private ServerNodeMapper serverNodeMapper;
@ -53,7 +59,7 @@ public class DashBoardServiceImpl implements DashBoardService {
taskQuantityResponseVO.setTotal(retryTaskLogMapper.countTaskTotal());
taskQuantityResponseVO.setFinish(retryTaskLogMapper.countTaskByRetryStatus(RetryStatusEnum.FINISH.getStatus()));
taskQuantityResponseVO.setMaxRetryCount(retryTaskLogMapper.countTaskByRetryStatus(RetryStatusEnum.MAX_RETRY_COUNT.getStatus()));
taskQuantityResponseVO.setMaxRetryCount(retryTaskLogMapper.countTaskByRetryStatus(RetryStatusEnum.MAX_COUNT.getStatus()));
taskQuantityResponseVO.setRunning(taskQuantityResponseVO.getTotal() - taskQuantityResponseVO.getFinish() - taskQuantityResponseVO.getMaxRetryCount());
return taskQuantityResponseVO;
@ -63,7 +69,8 @@ public class DashBoardServiceImpl implements DashBoardService {
public DispatchQuantityResponseVO countDispatch() {
DispatchQuantityResponseVO dispatchQuantityResponseVO = new DispatchQuantityResponseVO();
Long total = retryTaskLogMapper.selectCount(null);
// 任务的总调度量
Long total = retryTaskLogMessageMapper.selectCount(null);
dispatchQuantityResponseVO.setTotal(total);
if (total == 0) {
@ -71,8 +78,7 @@ public class DashBoardServiceImpl implements DashBoardService {
}
Long success = retryTaskLogMapper.selectCount(new LambdaQueryWrapper<RetryTaskLog>()
.in(RetryTaskLog::getRetryStatus, RetryStatusEnum.MAX_RETRY_COUNT.getStatus(),
RetryStatusEnum.FINISH.getStatus()));
.eq(RetryTaskLog::getRetryStatus, RetryStatusEnum.FINISH.getStatus()));
dispatchQuantityResponseVO.setSuccessPercent(BigDecimal.valueOf(success).divide(BigDecimal.valueOf(total), 2, RoundingMode.HALF_UP).multiply(BigDecimal.valueOf(100)));
return dispatchQuantityResponseVO;

35
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/service/impl/RetryDeadLetterServiceImpl.java
View File

@ -1,14 +1,21 @@
package com.aizuda.easy.retry.server.service.impl;
import cn.hutool.core.lang.Assert;
import com.aizuda.easy.retry.common.core.enums.RetryStatusEnum;
import com.aizuda.easy.retry.server.enums.TaskTypeEnum;
import com.aizuda.easy.retry.server.exception.EasyRetryServerException;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryDeadLetterMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryDeadLetter;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTask;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import com.aizuda.easy.retry.server.service.convert.RetryDeadLetterResponseVOConverter;
import com.aizuda.easy.retry.server.service.convert.RetryTaskConverter;
import com.aizuda.easy.retry.server.service.convert.RetryTaskLogConverter;
import com.aizuda.easy.retry.server.support.strategy.WaitStrategies;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.PageDTO;
import com.aizuda.easy.retry.server.config.RequestDataHelper;
import com.aizuda.easy.retry.server.service.RetryDeadLetterService;
@ -24,6 +31,7 @@ import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
/**
@ -35,9 +43,10 @@ public class RetryDeadLetterServiceImpl implements RetryDeadLetterService {
@Autowired
private RetryDeadLetterMapper retryDeadLetterMapper;
@Autowired
private RetryTaskMapper retryTaskMapper;
@Autowired
private RetryTaskLogMapper retryTaskLogMapper;
@Override
public PageResult<List<RetryDeadLetterResponseVO>> getRetryDeadLetterPage(RetryDeadLetterQueryVO queryVO) {
@ -83,19 +92,37 @@ public class RetryDeadLetterServiceImpl implements RetryDeadLetterService {
public boolean rollback(String groupName, Long id) {
RequestDataHelper.setPartition(groupName);
RetryDeadLetter retryDeadLetter = retryDeadLetterMapper.selectById(id);
Assert.notNull(retryDeadLetter, () -> new EasyRetryServerException("数据不存在"));
RetryTask retryTask = new RetryTask();
BeanUtils.copyProperties(retryDeadLetter, retryTask);
RetryTask retryTask = RetryTaskConverter.INSTANCE.toRetryTask(retryDeadLetter);
retryTask.setNextTriggerAt(WaitStrategies.randomWait(1, TimeUnit.SECONDS, 60, TimeUnit.SECONDS).computeRetryTime(null));
retryTask.setCreateDt(LocalDateTime.now());
retryTask.setUpdateDt(LocalDateTime.now());
retryTask.setId(null);
RequestDataHelper.setPartition(groupName);
Assert.isTrue(1 == retryTaskMapper.insert(retryTask), () -> new EasyRetryServerException("新增重试任务失败"));
RequestDataHelper.setPartition(groupName);
Assert.isTrue(1 == retryDeadLetterMapper.deleteById(retryDeadLetter.getId()), () -> new EasyRetryServerException("删除死信队列数据失败"));
// 变动日志的状态
RetryTaskLog retryTaskLog = new RetryTaskLog();
retryTaskLog.setRetryStatus(RetryStatusEnum.RUNNING.getStatus());
int update = retryTaskLogMapper.update(retryTaskLog, new LambdaUpdateWrapper<RetryTaskLog>()
.eq(RetryTaskLog::getUniqueId, retryTask.getUniqueId())
.eq(RetryTaskLog::getGroupName, retryTask.getGroupName()));
// 若日志不存在则初始化一个
if (update == 0) {
// 初始化回调日志
retryTaskLog = RetryTaskLogConverter.INSTANCE.toRetryTask(retryTask);
retryTaskLog.setTaskType(retryTask.getTaskType());
retryTaskLog.setRetryStatus(RetryStatusEnum.RUNNING.getStatus());
retryTaskLog.setCreateDt(LocalDateTime.now());
Assert.isTrue(1 == retryTaskLogMapper.insert(retryTaskLog),
() -> new EasyRetryServerException("新增重试日志失败"));
}
return true;
}

23
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/service/impl/RetryServiceImpl.java
View File

@ -4,17 +4,21 @@ import cn.hutool.core.lang.Assert;
import com.aizuda.easy.retry.server.enums.DelayLevelEnum;
import com.aizuda.easy.retry.server.enums.StatusEnum;
import com.aizuda.easy.retry.common.core.log.LogUtils;
import com.aizuda.easy.retry.server.enums.TaskTypeEnum;
import com.aizuda.easy.retry.server.exception.EasyRetryServerException;
import com.aizuda.easy.retry.server.model.dto.RetryTaskDTO;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryDeadLetterMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.SceneConfigMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.po.GroupConfig;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryDeadLetter;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTask;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import com.aizuda.easy.retry.server.persistence.mybatis.po.SceneConfig;
import com.aizuda.easy.retry.server.persistence.support.ConfigAccess;
import com.aizuda.easy.retry.server.persistence.support.RetryTaskAccess;
import com.aizuda.easy.retry.server.service.convert.RetryTaskLogConverter;
import com.aizuda.easy.retry.server.support.generator.IdGenerator;
import com.aizuda.easy.retry.server.support.strategy.WaitStrategies;
import com.aizuda.easy.retry.server.support.strategy.WaitStrategies.WaitStrategyEnum;
@ -64,7 +68,8 @@ public class RetryServiceImpl implements RetryService {
private RetryDeadLetterMapper retryDeadLetterMapper;
@Autowired
private SceneConfigMapper sceneConfigMapper;
@Autowired
private RetryTaskLogMapper retryTaskLogMapper;
@Transactional
@Override
@ -100,10 +105,12 @@ public class RetryServiceImpl implements RetryService {
return Boolean.TRUE;
}
LocalDateTime now = LocalDateTime.now();
RetryTask retryTask = RetryTaskConverter.INSTANCE.toRetryTask(retryTaskDTO);
retryTask.setUniqueId(getIdGenerator(retryTaskDTO.getGroupName()));
retryTask.setCreateDt(LocalDateTime.now());
retryTask.setUpdateDt(LocalDateTime.now());
retryTask.setTaskType(TaskTypeEnum.RETRY.getType());
retryTask.setCreateDt(now);
retryTask.setUpdateDt(now);
if (StringUtils.isBlank(retryTask.getExtAttrs())) {
retryTask.setExtAttrs(StringUtils.EMPTY);
@ -112,6 +119,14 @@ public class RetryServiceImpl implements RetryService {
retryTask.setNextTriggerAt(WaitStrategies.randomWait(1, TimeUnit.SECONDS, 60, TimeUnit.SECONDS).computeRetryTime(null));
Assert.isTrue(1 == retryTaskAccess.saveRetryTask(retryTask), () -> new EasyRetryServerException("failed to report data"));
// 初始化日志
RetryTaskLog retryTaskLog = RetryTaskLogConverter.INSTANCE.toRetryTask(retryTask);
retryTaskLog.setTaskType(TaskTypeEnum.RETRY.getType());
retryTaskLog.setCreateDt(now);
Assert.isTrue(1 == retryTaskLogMapper.insert(retryTaskLog),
() -> new EasyRetryServerException("新增重试日志失败"));
return Boolean.TRUE;
}
@ -149,7 +164,7 @@ public class RetryServiceImpl implements RetryService {
// 清除重试完成的数据
clearFinishRetryData(groupId);
List<RetryTask> retryTasks = retryTaskAccess.listRetryTaskByRetryCount(groupId, RetryStatusEnum.MAX_RETRY_COUNT.getStatus());
List<RetryTask> retryTasks = retryTaskAccess.listRetryTaskByRetryCount(groupId, RetryStatusEnum.MAX_COUNT.getStatus());
if (CollectionUtils.isEmpty(retryTasks)) {
return Boolean.TRUE;
}

30
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/service/impl/RetryTaskLogServiceImpl.java
View File

@ -1,7 +1,11 @@
package com.aizuda.easy.retry.server.service.impl;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMessageMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLogMessage;
import com.aizuda.easy.retry.server.web.model.request.RetryTaskLogMessageQueryVO;
import com.aizuda.easy.retry.server.web.model.response.RetryTaskLogMessageResponseVO;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.PageDTO;
import com.aizuda.easy.retry.server.web.model.base.PageResult;
@ -24,6 +28,8 @@ public class RetryTaskLogServiceImpl implements RetryTaskLogService {
@Autowired
private RetryTaskLogMapper retryTaskLogMapper;
@Autowired
private RetryTaskLogMessageMapper retryTaskLogMessageMapper;
@Override
public PageResult<List<RetryTaskLogResponseVO>> getRetryTaskLogPage(RetryTaskLogQueryVO queryVO) {
@ -48,7 +54,7 @@ public class RetryTaskLogServiceImpl implements RetryTaskLogService {
}
retryTaskLogLambdaQueryWrapper.select(RetryTaskLog::getGroupName, RetryTaskLog::getId, RetryTaskLog::getSceneName,
RetryTaskLog::getIdempotentId, RetryTaskLog::getBizNo, RetryTaskLog::getErrorMessage, RetryTaskLog::getRetryStatus,
RetryTaskLog::getIdempotentId, RetryTaskLog::getBizNo, RetryTaskLog::getRetryStatus,
RetryTaskLog::getCreateDt, RetryTaskLog::getUniqueId, RetryTaskLog::getTaskType);
PageDTO<RetryTaskLog> retryTaskLogPageDTO = retryTaskLogMapper.selectPage(pageDTO, retryTaskLogLambdaQueryWrapper.orderByDesc(RetryTaskLog::getCreateDt));
@ -57,6 +63,28 @@ public class RetryTaskLogServiceImpl implements RetryTaskLogService {
RetryTaskLogResponseVOConverter.INSTANCE.batchConvert(retryTaskLogPageDTO.getRecords()));
}
@Override
public PageResult<List<RetryTaskLogMessageResponseVO>> getRetryTaskLogMessagePage(
RetryTaskLogMessageQueryVO queryVO) {
PageDTO<RetryTaskLogMessage> pageDTO = new PageDTO<>(queryVO.getPage(), queryVO.getSize());
LambdaQueryWrapper<RetryTaskLogMessage> retryTaskLogLambdaQueryWrapper = new LambdaQueryWrapper<>();
if (StringUtils.isNotBlank(queryVO.getGroupName())) {
retryTaskLogLambdaQueryWrapper.eq(RetryTaskLogMessage::getGroupName, queryVO.getGroupName());
}
if (StringUtils.isNotBlank(queryVO.getUniqueId())) {
retryTaskLogLambdaQueryWrapper.eq(RetryTaskLogMessage::getUniqueId, queryVO.getUniqueId());
}
PageDTO<RetryTaskLogMessage> retryTaskLogPageDTO = retryTaskLogMessageMapper.selectPage(pageDTO, retryTaskLogLambdaQueryWrapper.orderByDesc(RetryTaskLogMessage::getCreateDt));
return new PageResult<>(
retryTaskLogPageDTO,
RetryTaskLogResponseVOConverter.INSTANCE.toRetryTaskLogMessageResponseVO(retryTaskLogPageDTO.getRecords()));
}
@Override
public RetryTaskLogResponseVO getRetryTaskLogById(Long id) {
RetryTaskLog retryTaskLog = retryTaskLogMapper.selectById(id);

65
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/support/dispatch/actor/exec/ExecCallbackUnitActor.java
View File

@ -1,24 +1,20 @@
package com.aizuda.easy.retry.server.support.dispatch.actor.exec;
import akka.actor.AbstractActor;
import cn.hutool.core.lang.Assert;
import com.aizuda.easy.retry.client.model.DispatchRetryResultDTO;
import akka.actor.ActorRef;
import com.aizuda.easy.retry.client.model.RetryCallbackDTO;
import com.aizuda.easy.retry.common.core.constant.SystemConstants;
import com.aizuda.easy.retry.server.akka.ActorGenerator;
import com.aizuda.easy.retry.server.dto.RegisterNodeInfo;
import com.aizuda.easy.retry.server.enums.StatusEnum;
import com.aizuda.easy.retry.common.core.log.LogUtils;
import com.aizuda.easy.retry.common.core.model.EasyRetryHeaders;
import com.aizuda.easy.retry.common.core.model.Result;
import com.aizuda.easy.retry.common.core.util.JsonUtil;
import com.aizuda.easy.retry.server.exception.EasyRetryServerException;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTask;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import com.aizuda.easy.retry.server.persistence.mybatis.po.ServerNode;
import com.aizuda.easy.retry.server.service.convert.RetryTaskLogConverter;
import com.aizuda.easy.retry.server.support.IdempotentStrategy;
import com.aizuda.easy.retry.server.support.context.CallbackRetryContext;
import com.aizuda.easy.retry.server.support.dispatch.actor.log.RetryTaskLogDTO;
import com.aizuda.easy.retry.server.support.retry.RetryExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
@ -32,7 +28,6 @@ import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import java.text.MessageFormat;
import java.time.LocalDateTime;
import java.util.Objects;
import java.util.concurrent.Callable;
@ -55,8 +50,6 @@ public class ExecCallbackUnitActor extends AbstractActor {
@Qualifier("bitSetIdempotentStrategyHandler")
private IdempotentStrategy<String, Integer> idempotentStrategy;
@Autowired
private RetryTaskLogMapper retryTaskLogMapper;
@Autowired
private RestTemplate restTemplate;
@Override
@ -67,34 +60,44 @@ public class ExecCallbackUnitActor extends AbstractActor {
RetryTask retryTask = context.getRetryTask();
RegisterNodeInfo serverNode = context.getServerNode();
RetryTaskLog retryTaskLog = RetryTaskLogConverter.INSTANCE.toRetryTask(retryTask);
retryTaskLog.setErrorMessage(StringUtils.EMPTY);
RetryTaskLogDTO retryTaskLog = new RetryTaskLogDTO();
retryTaskLog.setGroupName(retryTask.getGroupName());
retryTaskLog.setUniqueId(retryTask.getUniqueId());
retryTaskLog.setRetryStatus(retryTask.getRetryStatus());
try {
if (Objects.nonNull(serverNode)) {
retryExecutor.call((Callable<Result<Void>>) () -> callClient(retryTask, retryTaskLog, serverNode));
retryExecutor.call((Callable<Result<Void>>) () -> {
Result<Void> result = callClient(retryTask, serverNode);
if (StatusEnum.YES.getStatus() != result.getStatus() && StringUtils.isNotBlank(result.getMessage())) {
retryTaskLog.setMessage(result.getMessage());
} else {
retryTaskLog.setMessage("调度成功");
}
return result;
});
if (context.hasException()) {
retryTaskLog.setErrorMessage(context.getException().getMessage());
retryTaskLog.setMessage(context.getException().getMessage());
}
} else {
retryTaskLog.setErrorMessage("There are currently no available client PODs.");
retryTaskLog.setMessage("There are currently no available client PODs.");
}
}catch (Exception e) {
LogUtils.error(log, "回调客户端失败 retryTask:[{}]", JsonUtil.toJsonString(retryTask), e);
retryTaskLog.setErrorMessage(StringUtils.isBlank(e.getMessage()) ? StringUtils.EMPTY : e.getMessage());
LogUtils.error(log, "callback client error. retryTask:[{}]", JsonUtil.toJsonString(retryTask), e);
retryTaskLog.setMessage(StringUtils.isBlank(e.getMessage()) ? StringUtils.EMPTY : e.getMessage());
} finally {
// 清除幂等标识位
idempotentStrategy.clear(retryTask.getGroupName(), retryTask.getId().intValue());
ActorRef actorRef = ActorGenerator.logActor();
actorRef.tell(retryTaskLog, actorRef);
getContext().stop(getSelf());
// 记录重试日志
retryTaskLog.setCreateDt(LocalDateTime.now());
retryTaskLog.setId(null);
Assert.isTrue(1 == retryTaskLogMapper.insert(retryTaskLog),
() -> new EasyRetryServerException("新增重试日志失败"));
}
}).build();
@ -106,7 +109,7 @@ public class ExecCallbackUnitActor extends AbstractActor {
* @param retryTask {@link RetryTask} 需要重试的数据
* @return 重试结果返回值
*/
private Result<Void> callClient(RetryTask retryTask, RetryTaskLog retryTaskLog, RegisterNodeInfo serverNode) {
private Result<Void> callClient(RetryTask retryTask, RegisterNodeInfo serverNode) {
// 回调参数
RetryCallbackDTO retryCallbackDTO = new RetryCallbackDTO();
@ -129,20 +132,8 @@ public class ExecCallbackUnitActor extends AbstractActor {
String format = MessageFormat.format(URL, serverNode.getHostIp(), serverNode.getHostPort().toString(), serverNode.getContextPath());
Result result = restTemplate.postForObject(format, requestEntity, Result.class);
LogUtils.info(log, "回调请求客户端 response:[{}}] ", JsonUtil.toJsonString(result));
if (StatusEnum.YES.getStatus() != result.getStatus() && StringUtils.isNotBlank(result.getMessage())) {
retryTaskLog.setErrorMessage(result.getMessage());
} else {
DispatchRetryResultDTO data = JsonUtil.parseObject(JsonUtil.toJsonString(result.getData()), DispatchRetryResultDTO.class);
result.setData(data);
if (Objects.nonNull(data) && StringUtils.isNotBlank(data.getExceptionMsg())) {
retryTaskLog.setErrorMessage(data.getExceptionMsg());
}
}
LogUtils.info(log, "请求客户端 response:[{}}] ", JsonUtil.toJsonString(result));
LogUtils.info(log, "请求客户端 format:[{}] response:[{}}] ", format, JsonUtil.toJsonString(result));
return result;
}

73
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/support/dispatch/actor/exec/ExecUnitActor.java
View File

@ -1,27 +1,24 @@
package com.aizuda.easy.retry.server.support.dispatch.actor.exec;
import akka.actor.AbstractActor;
import cn.hutool.core.lang.Assert;
import akka.actor.ActorRef;
import com.aizuda.easy.retry.client.model.DispatchRetryDTO;
import com.aizuda.easy.retry.client.model.DispatchRetryResultDTO;
import com.aizuda.easy.retry.common.core.constant.SystemConstants;
import com.aizuda.easy.retry.common.core.log.LogUtils;
import com.aizuda.easy.retry.common.core.model.Result;
import com.aizuda.easy.retry.common.core.model.EasyRetryHeaders;
import com.aizuda.easy.retry.common.core.model.Result;
import com.aizuda.easy.retry.common.core.util.JsonUtil;
import com.aizuda.easy.retry.server.akka.ActorGenerator;
import com.aizuda.easy.retry.server.enums.StatusEnum;
import com.aizuda.easy.retry.server.dto.RegisterNodeInfo;
import com.aizuda.easy.retry.server.exception.EasyRetryServerException;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTask;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import com.aizuda.easy.retry.server.persistence.mybatis.po.ServerNode;
import com.aizuda.easy.retry.server.service.convert.RetryTaskLogConverter;
import com.aizuda.easy.retry.server.support.IdempotentStrategy;
import com.aizuda.easy.retry.server.support.context.MaxAttemptsPersistenceRetryContext;
import com.aizuda.easy.retry.server.support.dispatch.actor.log.RetryTaskLogDTO;
import com.aizuda.easy.retry.server.support.retry.RetryExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
@ -32,7 +29,6 @@ import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import java.text.MessageFormat;
import java.time.LocalDateTime;
import java.util.Objects;
import java.util.concurrent.Callable;
@ -55,8 +51,6 @@ public class ExecUnitActor extends AbstractActor {
@Qualifier("bitSetIdempotentStrategyHandler")
private IdempotentStrategy<String, Integer> idempotentStrategy;
@Autowired
private RetryTaskLogMapper retryTaskLogMapper;
@Autowired
private RestTemplate restTemplate;
@Override
@ -67,34 +61,54 @@ public class ExecUnitActor extends AbstractActor {
RetryTask retryTask = context.getRetryTask();
RegisterNodeInfo serverNode = context.getServerNode();
RetryTaskLog retryTaskLog = RetryTaskLogConverter.INSTANCE.toRetryTask(retryTask);
retryTaskLog.setErrorMessage(StringUtils.EMPTY);
RetryTaskLogDTO retryTaskLog = new RetryTaskLogDTO();
retryTaskLog.setGroupName(retryTask.getGroupName());
retryTaskLog.setUniqueId(retryTask.getUniqueId());
retryTaskLog.setRetryStatus(retryTask.getRetryStatus());
try {
if (Objects.nonNull(serverNode)) {
retryExecutor.call((Callable<Result<DispatchRetryResultDTO>>) () -> callClient(retryTask, retryTaskLog, serverNode));
retryExecutor.call((Callable<Result<DispatchRetryResultDTO>>) () -> {
Result<DispatchRetryResultDTO> result = callClient(retryTask, serverNode);
// 回调接口请求成功，处理返回值
if (StatusEnum.YES.getStatus() != result.getStatus() && StringUtils.isNotBlank(result.getMessage())) {
retryTaskLog.setMessage(result.getMessage());
} else {
DispatchRetryResultDTO data = JsonUtil.parseObject(JsonUtil.toJsonString(result.getData()), DispatchRetryResultDTO.class);
result.setData(data);
if (Objects.nonNull(data) && StringUtils.isNotBlank(data.getExceptionMsg())) {
retryTaskLog.setMessage(data.getExceptionMsg());
} else {
retryTaskLog.setMessage("调度成功");
}
}
return result;
});
// 请求发生异常
if (context.hasException()) {
retryTaskLog.setErrorMessage(context.getException().getMessage());
retryTaskLog.setMessage(context.getException().getMessage());
}
} else {
retryTaskLog.setErrorMessage("暂无可用的客户端POD");
retryTaskLog.setMessage("There are currently no available client PODs.");
}
}catch (Exception e) {
LogUtils.error(log, "回调客户端失败 retryTask:[{}]", JsonUtil.toJsonString(retryTask), e);
retryTaskLog.setErrorMessage(StringUtils.isBlank(e.getMessage()) ? StringUtils.EMPTY : e.getMessage());
LogUtils.error(log, "callback client error. retryTask:[{}]", JsonUtil.toJsonString(retryTask), e);
retryTaskLog.setMessage(e.getMessage());
} finally {
// 清除幂等标识位
idempotentStrategy.clear(retryTask.getGroupName(), retryTask.getId().intValue());
ActorRef actorRef = ActorGenerator.logActor();
actorRef.tell(retryTaskLog, actorRef);
getContext().stop(getSelf());
// 记录重试日志
retryTaskLog.setCreateDt(LocalDateTime.now());
retryTaskLog.setId(null);
Assert.isTrue(1 == retryTaskLogMapper.insert(retryTaskLog),
() -> new EasyRetryServerException("新增重试日志失败"));
}
}).build();
@ -106,7 +120,7 @@ public class ExecUnitActor extends AbstractActor {
* @param retryTask {@link RetryTask} 需要重试的数据
* @return 重试结果返回值
*/
private Result<DispatchRetryResultDTO> callClient(RetryTask retryTask, RetryTaskLog retryTaskLog, RegisterNodeInfo serverNode) {
private Result<DispatchRetryResultDTO> callClient(RetryTask retryTask, RegisterNodeInfo serverNode) {
DispatchRetryDTO dispatchRetryDTO = new DispatchRetryDTO();
dispatchRetryDTO.setIdempotentId(retryTask.getIdempotentId());
@ -128,17 +142,6 @@ public class ExecUnitActor extends AbstractActor {
String format = MessageFormat.format(URL, serverNode.getHostIp(), serverNode.getHostPort().toString(), serverNode.getContextPath());
Result<DispatchRetryResultDTO> result = restTemplate.postForObject(format, requestEntity, Result.class);
if (1 != result.getStatus() && StringUtils.isNotBlank(result.getMessage())) {
retryTaskLog.setErrorMessage(result.getMessage());
} else {
DispatchRetryResultDTO data = JsonUtil.parseObject(JsonUtil.toJsonString(result.getData()), DispatchRetryResultDTO.class);
result.setData(data);
if (Objects.nonNull(data) && StringUtils.isNotBlank(data.getExceptionMsg())) {
retryTaskLog.setErrorMessage(data.getExceptionMsg());
}
}
LogUtils.info(log, "请求客户端 response:[{}}] ", JsonUtil.toJsonString(result));
return result;

77
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/support/dispatch/actor/log/LogActor.java Normal file
View File

@ -0,0 +1,77 @@
package com.aizuda.easy.retry.server.support.dispatch.actor.log;
import akka.actor.AbstractActor;
import com.aizuda.easy.retry.common.core.enums.RetryStatusEnum;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMessageMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLogMessage;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
/**
* 处理日志信息
*
* @author: www.byteblogs.com
* @date : 2023-06-16 11:33
* @since 2.0.0
*/
@Component(LogActor.BEAN_NAME)
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
@Slf4j
public class LogActor extends AbstractActor {
public static final String BEAN_NAME = "LogActor";
@Autowired
private RetryTaskLogMessageMapper retryTaskLogMessageMapper;
@Autowired
private RetryTaskLogMapper retryTaskLogMapper;
@Override
public Receive createReceive() {
return receiveBuilder().match(RetryTaskLogDTO.class,
retryTaskLogDTO -> RetryStatusEnum.RUNNING.getStatus().equals(retryTaskLogDTO.getRetryStatus()),
retryTaskLogDTO -> {
saveRetryTaskLogMessage(retryTaskLogDTO);
getContext().stop(getSelf());
}).match(RetryTaskLogDTO.class, (retryTaskLogDTO) ->
RetryStatusEnum.MAX_COUNT.getStatus().equals(retryTaskLogDTO.getRetryStatus())
|| RetryStatusEnum.FINISH.getStatus().equals(retryTaskLogDTO.getRetryStatus()),
retryTaskLogDTO -> {
// 变动日志的状态
RetryTaskLog retryTaskLog = new RetryTaskLog();
retryTaskLog.setRetryStatus(retryTaskLogDTO.getRetryStatus());
retryTaskLogMapper.update(retryTaskLog, new LambdaUpdateWrapper<RetryTaskLog>()
.eq(RetryTaskLog::getUniqueId, retryTaskLogDTO.getUniqueId())
.eq(RetryTaskLog::getGroupName, retryTaskLogDTO.getGroupName()));
getContext().stop(getSelf());
}).build();
}
/**
* 报错日志详情
*/
private void saveRetryTaskLogMessage(final RetryTaskLogDTO retryTaskLogDTO) {
// 记录重试日志
RetryTaskLogMessage retryTaskLogMessage = new RetryTaskLogMessage();
retryTaskLogMessage.setUniqueId(retryTaskLogDTO.getUniqueId());
retryTaskLogMessage.setGroupName(retryTaskLogDTO.getGroupName());
String errorMessage = retryTaskLogDTO.getMessage();
retryTaskLogMessage.setMessage(
StringUtils.isBlank(errorMessage) ? StringUtils.EMPTY : errorMessage);
retryTaskLogMessage.setCreateDt(LocalDateTime.now());
retryTaskLogMessageMapper.insert(retryTaskLogMessage);
}
}

35
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/support/dispatch/actor/log/RetryTaskLogDTO.java Normal file
View File

@ -0,0 +1,35 @@
package com.aizuda.easy.retry.server.support.dispatch.actor.log;
import lombok.Data;
/**
* 日志上下文模型
*
* @author: www.byteblogs.com
* @date : 2023-06-16 14:14
* @since 2.0.0
*/
@Data
public class RetryTaskLogDTO {
/**
* 组名称
*/
private String groupName;
/**
* 同组下id唯一
*/
private String uniqueId;
/**
* 异常信息
*/
private String message;
/**
* 重试状态
*/
private Integer retryStatus;
}

56
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/support/dispatch/actor/result/FailureActor.java
View File

@ -1,21 +1,19 @@
package com.aizuda.easy.retry.server.support.dispatch.actor.result;
import akka.actor.AbstractActor;
import cn.hutool.core.lang.Assert;
import akka.actor.ActorRef;
import com.aizuda.easy.retry.common.core.constant.SystemConstants;
import com.aizuda.easy.retry.common.core.enums.RetryStatusEnum;
import com.aizuda.easy.retry.server.enums.TaskTypeEnum;
import com.aizuda.easy.retry.server.support.handler.CallbackRetryTaskHandler;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.PageDTO;
import com.aizuda.easy.retry.common.core.log.LogUtils;
import com.aizuda.easy.retry.server.exception.EasyRetryServerException;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMapper;
import com.aizuda.easy.retry.server.akka.ActorGenerator;
import com.aizuda.easy.retry.server.config.SystemProperties;
import com.aizuda.easy.retry.server.enums.TaskTypeEnum;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTask;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import com.aizuda.easy.retry.server.persistence.mybatis.po.SceneConfig;
import com.aizuda.easy.retry.server.persistence.support.ConfigAccess;
import com.aizuda.easy.retry.server.persistence.support.RetryTaskAccess;
import com.aizuda.easy.retry.server.support.dispatch.actor.log.RetryTaskLogDTO;
import com.aizuda.easy.retry.server.support.handler.CallbackRetryTaskHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
@ -25,14 +23,9 @@ import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.CollectionUtils;
import java.util.List;
/**
* 重试完成执行器
* 1、更新重试任务
* 2、记录重试日志
* 重试完成执行器 1、更新重试任务 2、记录重试日志
*
* @author www.byteblogs.com
* @date 2021-10-30
@ -52,11 +45,11 @@ public class FailureActor extends AbstractActor {
@Qualifier("configAccessProcessor")
private ConfigAccess configAccess;
@Autowired
private RetryTaskLogMapper retryTaskLogMapper;
@Autowired
private CallbackRetryTaskHandler callbackRetryTaskHandler;
@Autowired
private TransactionTemplate transactionTemplate;
@Autowired
private SystemProperties systemProperties;
@Override
public Receive createReceive() {
@ -65,23 +58,22 @@ public class FailureActor extends AbstractActor {
// 超过最大等级
SceneConfig sceneConfig =
configAccess.getSceneConfigByGroupNameAndSceneName(retryTask.getGroupName(), retryTask.getSceneName());
configAccess.getSceneConfigByGroupNameAndSceneName(retryTask.getGroupName(), retryTask.getSceneName());
try {
transactionTemplate.execute(new TransactionCallbackWithoutResult() {
@Override
protected void doInTransactionWithoutResult(TransactionStatus status) {
Integer maxRetryCount;
if (TaskTypeEnum.CALLBACK.getType().equals(retryTask.getTaskType())) {
maxRetryCount = SystemConstants.CALL_BACK.MAX_RETRY_COUNT;
maxRetryCount = systemProperties.getCallback().getMaxCount();
} else {
maxRetryCount = sceneConfig.getMaxRetryCount();
}
if (maxRetryCount <= retryTask.getRetryCount()) {
retryTask.setRetryStatus(RetryStatusEnum.MAX_RETRY_COUNT.getStatus());
retryTask.setRetryStatus(RetryStatusEnum.MAX_COUNT.getStatus());
// 创建一个回调任务
callbackRetryTaskHandler.create(retryTask);
}
@ -90,24 +82,20 @@ public class FailureActor extends AbstractActor {
}
});
} catch (Exception e) {
LogUtils.error(log,"更新重试任务失败", e);
LogUtils.error(log, "更新重试任务失败", e);
} finally {
getContext().stop(getSelf());
// 记录重试日志
PageDTO<RetryTaskLog> retryTaskLogPageDTO = retryTaskLogMapper.selectPage(new PageDTO<>(1, 1),
new LambdaQueryWrapper<RetryTaskLog>()
.eq(RetryTaskLog::getIdempotentId, retryTask.getIdempotentId())
.orderByDesc(RetryTaskLog::getId));
List<RetryTaskLog> records = retryTaskLogPageDTO.getRecords();
if (!CollectionUtils.isEmpty(records)) {
RetryTaskLog retryTaskLog = records.get(0);
retryTaskLog.setRetryStatus(retryTask.getRetryStatus());
Assert.isTrue(1 == retryTaskLogMapper.updateById(retryTaskLog),
() -> new EasyRetryServerException("更新重试日志失败"));
if (RetryStatusEnum.MAX_COUNT.getStatus().equals(retryTask.getRetryStatus())) {
RetryTaskLogDTO retryTaskLogDTO = new RetryTaskLogDTO();
retryTaskLogDTO.setGroupName(retryTask.getGroupName());
retryTaskLogDTO.setUniqueId(retryTask.getUniqueId());
retryTaskLogDTO.setRetryStatus(retryTask.getRetryStatus());
retryTaskLogDTO.setMessage("任务已经到达最大执行次数了.");
ActorRef actorRef = ActorGenerator.logActor();
actorRef.tell(retryTaskLogDTO, actorRef);
}
getContext().stop(getSelf());
}
}).build();

35
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/support/dispatch/actor/result/FinishActor.java
View File

@ -1,17 +1,14 @@
package com.aizuda.easy.retry.server.support.dispatch.actor.result;
import akka.actor.AbstractActor;
import cn.hutool.core.lang.Assert;
import akka.actor.ActorRef;
import com.aizuda.easy.retry.common.core.enums.RetryStatusEnum;
import com.aizuda.easy.retry.server.akka.ActorGenerator;
import com.aizuda.easy.retry.server.enums.TaskTypeEnum;
import com.aizuda.easy.retry.server.support.dispatch.actor.log.RetryTaskLogDTO;
import com.aizuda.easy.retry.server.support.handler.CallbackRetryTaskHandler;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.PageDTO;
import com.aizuda.easy.retry.common.core.log.LogUtils;
import com.aizuda.easy.retry.server.exception.EasyRetryServerException;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTask;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import com.aizuda.easy.retry.server.persistence.support.RetryTaskAccess;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
@ -22,9 +19,6 @@ import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.CollectionUtils;
import java.util.List;
/**
* 重试完成执行器
@ -46,8 +40,6 @@ public class FinishActor extends AbstractActor {
@Qualifier("retryTaskAccessProcessor")
private RetryTaskAccess<RetryTask> retryTaskAccess;
@Autowired
private RetryTaskLogMapper retryTaskLogMapper;
@Autowired
private CallbackRetryTaskHandler callbackRetryTaskHandler;
@Autowired
private TransactionTemplate transactionTemplate;
@ -76,21 +68,16 @@ public class FinishActor extends AbstractActor {
LogUtils.error(log, "更新重试任务失败", e);
} finally {
RetryTaskLogDTO retryTaskLogDTO = new RetryTaskLogDTO();
retryTaskLogDTO.setGroupName(retryTask.getGroupName());
retryTaskLogDTO.setUniqueId(retryTask.getUniqueId());
retryTaskLogDTO.setRetryStatus(retryTask.getRetryStatus());
retryTaskLogDTO.setMessage("任务已经执行成功了.");
ActorRef actorRef = ActorGenerator.logActor();
actorRef.tell(retryTaskLogDTO, actorRef);
getContext().stop(getSelf());
// 记录重试日志
PageDTO<RetryTaskLog> retryTaskLogPageDTO = retryTaskLogMapper.selectPage(new PageDTO<>(1, 1),
new LambdaQueryWrapper<RetryTaskLog>()
.eq(RetryTaskLog::getIdempotentId, retryTask.getIdempotentId())
.orderByDesc(RetryTaskLog::getId));
List<RetryTaskLog> records = retryTaskLogPageDTO.getRecords();
if (!CollectionUtils.isEmpty(records)) {
RetryTaskLog retryTaskLog = records.get(0);
retryTaskLog.setRetryStatus(retryTask.getRetryStatus());
Assert.isTrue(1 == retryTaskLogMapper.updateById(retryTaskLog),
() -> new EasyRetryServerException("更新重试日志失败"));
}
}

18
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/support/handler/CallbackRetryTaskHandler.java
View File

@ -3,11 +3,15 @@ package com.aizuda.easy.retry.server.support.handler;
import cn.hutool.core.lang.Assert;
import com.aizuda.easy.retry.common.core.constant.SystemConstants;
import com.aizuda.easy.retry.common.core.enums.RetryStatusEnum;
import com.aizuda.easy.retry.server.config.SystemProperties;
import com.aizuda.easy.retry.server.enums.TaskTypeEnum;
import com.aizuda.easy.retry.server.exception.EasyRetryServerException;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTask;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import com.aizuda.easy.retry.server.persistence.support.RetryTaskAccess;
import com.aizuda.easy.retry.server.service.convert.RetryTaskConverter;
import com.aizuda.easy.retry.server.service.convert.RetryTaskLogConverter;
import com.aizuda.easy.retry.server.support.strategy.WaitStrategies;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
@ -29,6 +33,10 @@ public class CallbackRetryTaskHandler {
@Autowired
@Qualifier("retryTaskAccessProcessor")
private RetryTaskAccess<RetryTask> retryTaskAccess;
@Autowired
private RetryTaskLogMapper retryTaskLogMapper;
@Autowired
private SystemProperties systemProperties;
@Transactional
public void create(RetryTask retryTask) {
@ -36,7 +44,7 @@ public class CallbackRetryTaskHandler {
callbackRetryTask.setTaskType(TaskTypeEnum.CALLBACK.getType());
callbackRetryTask.setId(null);
callbackRetryTask.setUniqueId(SystemConstants.CALL_BACK.CB_ + retryTask.getUniqueId());
callbackRetryTask.setUniqueId(systemProperties.getCallback().getPrefix() + retryTask.getUniqueId());
callbackRetryTask.setRetryStatus(RetryStatusEnum.RUNNING.getStatus());
callbackRetryTask.setRetryCount(0);
callbackRetryTask.setCreateDt(LocalDateTime.now());
@ -46,6 +54,14 @@ public class CallbackRetryTaskHandler {
Assert.isTrue(1 == retryTaskAccess.saveRetryTask(callbackRetryTask), () -> new EasyRetryServerException("failed to report data"));
// 初始化回调日志
RetryTaskLog retryTaskLog = RetryTaskLogConverter.INSTANCE.toRetryTask(callbackRetryTask);
// 记录重试日志
retryTaskLog.setTaskType(TaskTypeEnum.CALLBACK.getType());
retryTaskLog.setCreateDt(LocalDateTime.now());
Assert.isTrue(1 == retryTaskLogMapper.insert(retryTaskLog),
() -> new EasyRetryServerException("新增重试日志失败"));
}
}

8
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/support/schedule/AlarmNotifyThreadSchedule.java
View File

@ -2,7 +2,7 @@ package com.aizuda.easy.retry.server.support.schedule;
import com.aizuda.easy.retry.common.core.alarm.Alarm;
import com.aizuda.easy.retry.common.core.alarm.AlarmContext;
import com.aizuda.easy.retry.common.core.alarm.AltinAlarmFactory;
import com.aizuda.easy.retry.common.core.alarm.EasyRetryAlarmFactory;
import com.aizuda.easy.retry.common.core.enums.NotifySceneEnum;
import com.aizuda.easy.retry.common.core.enums.RetryStatusEnum;
import com.aizuda.easy.retry.common.core.log.LogUtils;
@ -52,7 +52,7 @@ public class AlarmNotifyThreadSchedule {
@Autowired
private RetryTaskMapper retryTaskMapper;
@Autowired
private AltinAlarmFactory altinAlarmFactory;
private EasyRetryAlarmFactory easyRetryAlarmFactory;
@Autowired
@Qualifier("configAccessProcessor")
private ConfigAccess configAccess;
@ -84,7 +84,7 @@ public class AlarmNotifyThreadSchedule {
.title("组:[{}])重试数据过多", groupConfig.getGroupName())
.notifyAttribute(notifyConfig.getNotifyAttribute());
Alarm<AlarmContext> alarmType = altinAlarmFactory.getAlarmType(notifyConfig.getNotifyType());
Alarm<AlarmContext> alarmType = easyRetryAlarmFactory.getAlarmType(notifyConfig.getNotifyType());
alarmType.asyncSendMessage(context);
}
}
@ -122,7 +122,7 @@ public class AlarmNotifyThreadSchedule {
.title("组:[{}] 环境重试失败数据监控", groupConfig.getGroupName())
.notifyAttribute(notifyConfig.getNotifyAttribute());
Alarm<AlarmContext> alarmType = altinAlarmFactory.getAlarmType(notifyConfig.getNotifyType());
Alarm<AlarmContext> alarmType = easyRetryAlarmFactory.getAlarmType(notifyConfig.getNotifyType());
alarmType.asyncSendMessage(context);
}
}

32
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/support/schedule/ClearThreadSchedule.java
View File

@ -1,16 +1,18 @@
package com.aizuda.easy.retry.server.support.schedule;
import cn.hutool.core.lang.Assert;
import com.aizuda.easy.retry.common.core.log.LogUtils;
import com.aizuda.easy.retry.server.config.SystemProperties;
import com.aizuda.easy.retry.server.dto.RegisterNodeInfo;
import com.aizuda.easy.retry.server.exception.EasyRetryServerException;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMessageMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.mapper.ServerNodeMapper;
import com.aizuda.easy.retry.server.persistence.mybatis.po.ServerNode;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLog;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLogMessage;
import com.aizuda.easy.retry.server.persistence.support.ConfigAccess;
import com.aizuda.easy.retry.server.service.RetryService;
import com.aizuda.easy.retry.server.support.cache.CacheRegisterTable;
import com.aizuda.easy.retry.server.support.register.ServerRegister;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import lombok.extern.slf4j.Slf4j;
import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
import org.springframework.beans.factory.annotation.Autowired;
@ -43,6 +45,13 @@ public class ClearThreadSchedule {
@Qualifier("configAccessProcessor")
private ConfigAccess configAccess;
@Autowired
private RetryTaskLogMessageMapper retryTaskLogMessageMapper;
@Autowired
private RetryTaskLogMapper retryTaskLogMapper;
@Autowired
private SystemProperties systemProperties;
/**
* 删除过期下线机器
*/
@ -94,4 +103,19 @@ public class ClearThreadSchedule {
}
/**
* 清理日志 一小时运行一次
*/
@Scheduled(cron = "0 0 0/1 * * ? ")
@SchedulerLock(name = "clearLog", lockAtMostFor = "PT1H", lockAtLeastFor = "PT1H")
public void clearLog() {
try {
LocalDateTime endTime = LocalDateTime.now().minusDays(systemProperties.getLogStorage());
retryTaskLogMapper.delete(new LambdaUpdateWrapper<RetryTaskLog>().le(RetryTaskLog::getCreateDt, endTime));
retryTaskLogMessageMapper.delete(new LambdaUpdateWrapper<RetryTaskLogMessage>().le(RetryTaskLogMessage::getCreateDt, endTime));
} catch (Exception e) {
LogUtils.error(log, "clear log error", e);
}
}
}

6
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/support/strategy/WaitStrategies.java
View File

@ -1,6 +1,7 @@
package com.aizuda.easy.retry.server.support.strategy;
import com.aizuda.easy.retry.common.core.constant.SystemConstants;
import com.aizuda.easy.retry.server.config.SystemProperties;
import com.aizuda.easy.retry.server.enums.TaskTypeEnum;
import com.aizuda.easy.retry.server.exception.EasyRetryServerException;
import com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTask;
@ -161,10 +162,11 @@ public class WaitStrategies {
@Override
public LocalDateTime computeRetryTime(RetryContext retryContext) {
RetryTask retryTask = retryContext.getRetryTask();
int triggerInterval;
long triggerInterval;
if (TaskTypeEnum.CALLBACK.getType().equals(retryTask.getTaskType())) {
// 回调失败的默认15分钟执行一次重试
triggerInterval = SystemConstants.CALL_BACK.TRIGGER_INTERVAL;
SystemProperties systemProperties = SpringContext.CONTEXT.getBean(SystemProperties.class);
triggerInterval = systemProperties.getCallback().getTriggerInterval();
} else {
ConfigAccess configAccess = SpringContext.CONTEXT.getBean("configAccessProcessor", ConfigAccess.class);
SceneConfig sceneConfig =

8
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/web/controller/RetryTaskLogController.java
View File

@ -2,8 +2,10 @@ package com.aizuda.easy.retry.server.web.controller;
import com.aizuda.easy.retry.server.service.RetryTaskLogService;
import com.aizuda.easy.retry.server.web.model.base.PageResult;
import com.aizuda.easy.retry.server.web.model.request.RetryTaskLogMessageQueryVO;
import com.aizuda.easy.retry.server.web.model.request.RetryTaskLogQueryVO;
import com.aizuda.easy.retry.server.web.annotation.LoginRequired;
import com.aizuda.easy.retry.server.web.model.response.RetryTaskLogMessageResponseVO;
import com.aizuda.easy.retry.server.web.model.response.RetryTaskLogResponseVO;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
@ -32,6 +34,12 @@ public class RetryTaskLogController {
return retryTaskLogService.getRetryTaskLogPage(queryVO);
}
@LoginRequired
@GetMapping("/message/list")
public PageResult<List<RetryTaskLogMessageResponseVO>> getRetryTaskLogPage(RetryTaskLogMessageQueryVO queryVO) {
return retryTaskLogService.getRetryTaskLogMessagePage(queryVO);
}
@LoginRequired
@GetMapping("{id}")
public RetryTaskLogResponseVO getRetryTaskLogById(@PathVariable("id") Long id) {

16
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/web/model/request/RetryTaskLogMessageQueryVO.java Normal file
View File

@ -0,0 +1,16 @@
package com.aizuda.easy.retry.server.web.model.request;
import com.aizuda.easy.retry.server.web.model.base.BaseQueryVO;
import lombok.Data;
/**
* @author: www.byteblogs.com
* @date : 2022-02-28 09:08
*/
@Data
public class RetryTaskLogMessageQueryVO extends BaseQueryVO {
private String groupName;
private String uniqueId;
}

18
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/web/model/response/RetryTaskLogMessageResponseVO.java Normal file
View File

@ -0,0 +1,18 @@
package com.aizuda.easy.retry.server.web.model.response;
import lombok.Data;
import java.time.LocalDateTime;
/**
* @author: www.byteblogs.com
* @date : 2022-02-28 09:09
*/
@Data
public class RetryTaskLogMessageResponseVO {
private String message;
private LocalDateTime createDt;
}

2
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/web/model/response/RetryTaskLogResponseVO.java
View File

@ -35,8 +35,6 @@ public class RetryTaskLogResponseVO {
private Integer taskType;
private String errorMessage;
private LocalDateTime createDt;
}

6
easy-retry-server/src/main/resources/application.yml
View File

@ -43,5 +43,11 @@ easy-retry:
netty-port: 1788 # 服务端netty端口
total-partition: 2 # 重试和死信表的分区总数
limiter: 10 # 一个客户端每秒最多接收的重试数量指令
step: 100 # 号段模式下步长配置
log-storage: 90 # 日志保存时间(单位: day)
callback:
max-count: 288
trigger-interval: 900

49
easy-retry-server/src/main/resources/mapper/RetryTaskLogMapper.xml
View File

@ -13,11 +13,11 @@
<result column="ext_attrs" jdbcType="VARCHAR" property="extAttrs"/>
<result column="retry_status" jdbcType="TINYINT" property="retryStatus"/>
<result column="task_type" jdbcType="TINYINT" property="taskType"/>
<result column="error_message" jdbcType="VARCHAR" property="errorMessage"/>
<result column="create_dt" jdbcType="TIMESTAMP" property="createDt"/>
</resultMap>
<sql id="Base_Column_List">
id, unique_id, group_name, scene_name, idempotent_id, biz_no, executor_name, args_str, ext_attrs, retry_status, error_message,
id
, unique_id, group_name, scene_name, idempotent_id, biz_no, executor_name, args_str, ext_attrs, retry_status,
create_dt, task_type
</sql>
<select id="selectByPrimaryKey" parameterType="java.lang.Long" resultMap="BaseResultMap">
@ -28,40 +28,31 @@
</select>
<select id="countTaskTotal" resultType="java.lang.Long">
select count(*)
from (
select group_name, scene_name, idempotent_id from retry_task_log group by group_name, scene_name, idempotent_id) a
from retry_task_log
</select>
<select id="countTaskByRetryStatus" resultType="java.lang.Long">
select count(*)
from (
select group_name, scene_name, idempotent_id
from retry_task_log
where retry_status = #{retryStatus}
group by group_name, scene_name, idempotent_id) a
from retry_task_log
where retry_status = #{retryStatus}
</select>
<select id="rankSceneQuantity"
resultType="com.aizuda.easy.retry.server.web.model.response.SceneQuantityRankResponseVO">
select group_name, scene_name, count(*) total
from (
select group_name, scene_name, idempotent_id, count(*)
resultType="com.aizuda.easy.retry.server.web.model.response.SceneQuantityRankResponseVO">
select group_name, scene_name, count(*) as total
from retry_task_log
<where>
task_type = 1
<if test="groupName != '' and groupName != null">
group_name = #{groupName}
and group_name = #{groupName}
</if>
and create_dt >= #{startTime} and create_dt &lt;= #{endTime}
</where>
group by group_name, scene_name, idempotent_id) a
group by group_name, scene_name
order by total desc;
order by total desc
</select>
<select id="lineDispatchQuantity"
resultType="com.aizuda.easy.retry.server.web.model.response.DispatchQuantityResponseVO">
resultType="com.aizuda.easy.retry.server.web.model.response.DispatchQuantityResponseVO">
select
distinct(create_dt), count(*) total
from (
select group_name, scene_name, idempotent_id,
<choose>
<when test="type == 'day'">
DATE_FORMAT(create_dt,'%H')
@ -79,25 +70,21 @@
DATE_FORMAT(create_dt,'%Y-%m-%d')
</otherwise>
</choose>
as create_dt, count(*)
as createDt, count(*) as total
from retry_task_log
<where>
<if test="groupName != '' and groupName != null">
group_name = #{groupName}
</if>
<if test="retryStatus!=null ">
and retry_status = #{retryStatus}
and retry_status = #{retryStatus}
</if>
and create_dt >= #{startTime} and create_dt &lt;= #{endTime}
</where>
group by group_name, scene_name, idempotent_id, create_dt) a
group by create_dt
order by total desc;
group by createDt
order by total desc
</select>
<delete id="deleteByPrimaryKey" parameterType="java.lang.Long">
delete
from retry_task_log
where id = #{id,jdbcType=BIGINT}
</delete>
</mapper>

14
easy-retry-server/src/main/resources/mapper/RetryTaskLogMessageMapper.xml Normal file
View File

@ -0,0 +1,14 @@
<?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.aizuda.easy.retry.server.persistence.mybatis.mapper.RetryTaskLogMessageMapper">
<!-- 通用查询映射结果 -->
<resultMap id="BaseResultMap" type="com.aizuda.easy.retry.server.persistence.mybatis.po.RetryTaskLogMessage">
<id column="id" property="id" />
<result column="group_name" property="groupName" />
<result column="unique_id" property="uniqueId" />
<result column="create_dt" property="createDt" />
<result column="message" property="message" />
</resultMap>
</mapper>

6
easy-retry-server/src/main/resources/spring-configuration-metadata.json
View File

@ -8,11 +8,11 @@
],
"properties": [
{
"name": "easy-retry.server.last-days",
"type": "java.lang.Integer",
"name": "easy-retry.callback",
"type": "jcom.aizuda.easy.retry.server.config.SystemProperties.Callback",
"defaultValue": "30",
"description": "服务端的地址，若服务端集群部署则此处配置域名",
"sourceType": "com.aizuda.easy.retry.server.config.SystemProperties"
}
]
}
}