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

feat: 2.6.0

Browse Source 
1. 优化定时任务回调重试
2. 优化工作流前端显示问题
This commit is contained in:
byteblogs168 2024-01-13 22:53:09 +08:00
parent ee5dfa962c
commit c8a238b91c
20 changed files with 247 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
easy-retry-client/easy-retry-client-job-core/src/main/java/com/aizuda/easy/retry/client/job/core/client/JobEndPoint.java
View File

@ -11,6 +11,7 @@ import com.aizuda.easy.retry.client.model.request.DispatchJobRequest;
import com.aizuda.easy.retry.common.core.context.SpringContext;
import com.aizuda.easy.retry.common.core.model.JobContext;
import com.aizuda.easy.retry.common.core.model.Result;
import com.aizuda.easy.retry.common.log.EasyRetryLog;
import lombok.extern.slf4j.Slf4j;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.PostMapping;
@ -36,19 +37,26 @@ public class JobEndPoint {
JobContext jobContext = buildJobContext(dispatchJob);
JobExecutorInfo jobExecutorInfo = JobExecutorInfoCache.get(jobContext.getExecutorInfo());
if (Objects.isNull(jobExecutorInfo)) {
EasyRetryLog.REMOTE.error("执行器配置有误. executorInfo:[{}]", dispatchJob.getExecutorInfo());
return new Result<>("执行器配置有误", Boolean.FALSE);
}
// 选择执行器
Object executor = jobExecutorInfo.getExecutor();
IJobExecutor jobExecutor;
if (IJobExecutor.class.isAssignableFrom(executor.getClass())) {
jobExecutor = (AbstractJobExecutor) executor;
} else {
jobExecutor = SpringContext.getBeanByType(AnnotationJobExecutor.class);
try {
// 选择执行器
Object executor = jobExecutorInfo.getExecutor();
IJobExecutor jobExecutor;
if (IJobExecutor.class.isAssignableFrom(executor.getClass())) {
jobExecutor = (AbstractJobExecutor) executor;
} else {
jobExecutor = SpringContext.getBeanByType(AnnotationJobExecutor.class);
}
jobExecutor.jobExecute(jobContext);
} catch (Exception e) {
EasyRetryLog.REMOTE.error("客户端发生非预期异常. taskBatchId:[{}]", dispatchJob.getTaskBatchId());
throw e;
}
jobExecutor.jobExecute(jobContext);
return new Result<>(Boolean.TRUE);
}

10
easy-retry-client/easy-retry-client-job-core/src/main/java/com/aizuda/easy/retry/client/job/core/executor/JobExecutorFutureCallback.java
View File

@ -30,7 +30,7 @@ public class JobExecutorFutureCallback implements FutureCallback<ExecuteResult>
private static final JobNettyClient CLIENT = RequestBuilder.<JobNettyClient, NettyResult>newBuilder()
.client(JobNettyClient.class)
.callback(nettyResult -> LogUtils.info(log, "Data report successfully requestId:[{}]", nettyResult.getRequestId())).build();
.callback(nettyResult -> EasyRetryLog.LOCAL.info("Job execute result report successfully requestId:[{}]", nettyResult.getRequestId())).build();
private final JobContext jobContext;
@ -41,7 +41,7 @@ public class JobExecutorFutureCallback implements FutureCallback<ExecuteResult>
@Override
public void onSuccess(ExecuteResult result) {
// 上报执行成功
EasyRetryLog.LOCAL.warn("任务执行成功 taskBatchId:[{}] [{}]", jobContext.getTaskBatchId(), JsonUtil.toJsonString(result));
EasyRetryLog.REMOTE.info("任务执行成功 taskBatchId:[{}] [{}]", jobContext.getTaskBatchId(), JsonUtil.toJsonString(result));
if (Objects.isNull(result)) {
result = ExecuteResult.success();
@ -57,7 +57,7 @@ public class JobExecutorFutureCallback implements FutureCallback<ExecuteResult>
try {
CLIENT.dispatchResult(buildDispatchJobResultRequest(result, taskStatus));
} catch (Exception e) {
EasyRetryLog.LOCAL.error("执行结果上报异常.[{}]", jobContext.getTaskId(), e);
EasyRetryLog.REMOTE.error("执行结果上报异常.[{}]", jobContext.getTaskId(), e);
} finally {
stopThreadPool();
ThreadLocalLogUtil.removeContext();
@ -67,7 +67,7 @@ public class JobExecutorFutureCallback implements FutureCallback<ExecuteResult>
@Override
public void onFailure(final Throwable t) {
// 上报执行失败
log.error("任务执行失败 任务执行成功 taskBatchId:[{}]", jobContext.getTaskBatchId(), t);
EasyRetryLog.REMOTE.error("任务执行失败 taskBatchId:[{}]", jobContext.getTaskBatchId(), t);
try {
ExecuteResult failure = ExecuteResult.failure();
@ -81,7 +81,7 @@ public class JobExecutorFutureCallback implements FutureCallback<ExecuteResult>
buildDispatchJobResultRequest(failure, JobTaskStatusEnum.FAIL.getStatus())
);
} catch (Exception e) {
EasyRetryLog.LOCAL.error("执行结果上报异常.[{}]", jobContext.getTaskId(), e);
EasyRetryLog.REMOTE.error("执行结果上报异常.[{}]", jobContext.getTaskId(), e);
} finally {
stopThreadPool();
ThreadLocalLogUtil.removeContext();

2
easy-retry-common/easy-retry-common-server-api/src/main/java/com/aizuda/easy/retry/server/model/dto/CallbackParamsDTO.java
View File

@ -3,6 +3,8 @@ package com.aizuda.easy.retry.server.model.dto;
import lombok.Data;
/**
* 工作流回调节点参数模型
*
* @author: xiaowoniu
* @date : 2024-01-02
* @since : 2.6.0

14
easy-retry-server/easy-retry-server-common/src/main/java/com/aizuda/easy/retry/server/common/handler/ClientNodeAllocateHandler.java
View File

@ -1,13 +1,13 @@
package com.aizuda.easy.retry.server.common.handler;
import com.aizuda.easy.retry.common.core.log.LogUtils;
import com.aizuda.easy.retry.common.log.EasyRetryLog;
import com.aizuda.easy.retry.server.common.ClientLoadBalance;
import com.aizuda.easy.retry.server.common.allocate.client.ClientLoadBalanceManager;
import com.aizuda.easy.retry.server.common.allocate.client.ClientLoadBalanceManager.AllocationAlgorithmEnum;
import com.aizuda.easy.retry.server.common.dto.RegisterNodeInfo;
import com.aizuda.easy.retry.server.common.cache.CacheRegisterTable;
import com.aizuda.easy.retry.server.common.dto.RegisterNodeInfo;
import com.aizuda.easy.retry.template.datasource.access.AccessTemplate;
import lombok.extern.slf4j.Slf4j;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
@ -22,11 +22,9 @@ import java.util.stream.Stream;
* @date : 2023-01-10 14:18
*/
@Component
@Slf4j
@RequiredArgsConstructor
public class ClientNodeAllocateHandler {
@Autowired
protected AccessTemplate accessTemplate;
private final AccessTemplate accessTemplate;
/**
* 获取分配的节点
@ -39,7 +37,7 @@ public class ClientNodeAllocateHandler {
Set<RegisterNodeInfo> serverNodes = CacheRegisterTable.getServerNodeSet(groupName, namespaceId);
if (CollectionUtils.isEmpty(serverNodes)) {
LogUtils.warn(log, "client node is null. groupName:[{}]", groupName);
EasyRetryLog.LOCAL.warn("client node is null. groupName:[{}]", groupName);
return null;
}

2
easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/JobTaskConverter.java
View File

@ -63,7 +63,7 @@ public interface JobTaskConverter {
JobLogMessage toJobLogMessage(LogTaskDTO logTaskDTO);
JobLogDTO toJobLogDTO(JobExecutorContext context);
LogMetaDTO toJobLogDTO(ClientCallbackContext context);
LogMetaDTO toJobLogDTO(JobExecutorResultDTO resultDTO);

114
easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/callback/AbstractClientCallbackHandler.java
View File

@ -1,9 +1,28 @@
package com.aizuda.easy.retry.server.job.task.support.callback;
import akka.actor.ActorRef;
import cn.hutool.core.util.StrUtil;
import com.aizuda.easy.retry.common.core.enums.JobTaskStatusEnum;
import com.aizuda.easy.retry.common.log.EasyRetryLog;
import com.aizuda.easy.retry.server.common.akka.ActorGenerator;
import com.aizuda.easy.retry.server.common.dto.RegisterNodeInfo;
import com.aizuda.easy.retry.server.common.util.ClientInfoUtils;
import com.aizuda.easy.retry.server.job.task.dto.LogMetaDTO;
import com.aizuda.easy.retry.server.job.task.dto.RealJobExecutorDTO;
import com.aizuda.easy.retry.server.job.task.support.ClientCallbackHandler;
import com.aizuda.easy.retry.server.job.task.support.JobTaskConverter;
import com.aizuda.easy.retry.template.datasource.persistence.mapper.JobMapper;
import com.aizuda.easy.retry.template.datasource.persistence.mapper.JobTaskMapper;
import com.aizuda.easy.retry.template.datasource.persistence.po.Job;
import com.aizuda.easy.retry.template.datasource.persistence.po.JobTask;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.toolkit.SqlHelper;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import java.util.Objects;
/**
* @author www.byteblogs.com
* @date 2023-10-03 23:12:33
@ -11,12 +30,107 @@ import org.springframework.transaction.annotation.Transactional;
*/
public abstract class AbstractClientCallbackHandler implements ClientCallbackHandler, InitializingBean {
@Autowired
private JobTaskMapper jobTaskMapper;
@Autowired
private JobMapper jobMapper;
@Override
@Transactional
public void callback(ClientCallbackContext context) {
// 判定是否需要重试
boolean needRetry = isNeedRetry(context);
if (needRetry) {
// 更新重试次数
if (updateRetryCount(context)) {
Job job = context.getJob();
JobTask jobTask = context.getJobTask();
RealJobExecutorDTO realJobExecutor = JobTaskConverter.INSTANCE.toRealJobExecutorDTO(JobTaskConverter.INSTANCE.toJobExecutorContext(job), jobTask);
realJobExecutor.setClientId(ClientInfoUtils.clientId(context.getClientInfo()));
ActorRef actorRef = ActorGenerator.jobRealTaskExecutorActor();
actorRef.tell(realJobExecutor, actorRef);
LogMetaDTO logMetaDTO = JobTaskConverter.INSTANCE.toJobLogDTO(context);
EasyRetryLog.REMOTE.info("任务执行/调度失败执行重试. 重试次数:[{}] <|>{}<|>",
jobTask.getRetryCount() + 1, logMetaDTO);
return;
}
}
// 不需要重试执行回调
doCallback(context);
}
private boolean updateRetryCount(ClientCallbackContext context) {
JobTask updateJobTask = new JobTask();
updateJobTask.setRetryCount(1);
String newClient = chooseNewClient(context);
if (StrUtil.isNotBlank(newClient)) {
updateJobTask.setClientInfo(newClient);
// 覆盖老的的客户端信息
context.setClientInfo(newClient);
}
Job job = context.getJob();
return SqlHelper.retBool(jobTaskMapper.update(updateJobTask, Wrappers.<JobTask>lambdaUpdate()
.lt(JobTask::getRetryCount, job.getMaxRetryTimes())
.eq(JobTask::getId, context.getTaskId())
));
}
private boolean isNeedRetry(ClientCallbackContext context) {
if (context.getTaskStatus().equals(JobTaskStatusEnum.FAIL.getStatus())) {
JobTask jobTask = jobTaskMapper.selectById(context.getTaskId());
Job job = jobMapper.selectById(context.getJobId());
if (Objects.isNull(jobTask) || Objects.isNull(job)) {
return Boolean.FALSE;
}
if (jobTask.getRetryCount() < job.getMaxRetryTimes()) {
context.setClientInfo(jobTask.getClientInfo());
context.setJob(job);
context.setJobTask(jobTask);
return Boolean.TRUE;
}
}
return Boolean.FALSE;
}
protected String chooseNewClient(ClientCallbackContext context) {
return null;
}
private void failRetry(ClientCallbackContext context) {
if (context.getTaskStatus().equals(JobTaskStatusEnum.FAIL.getStatus())) {
JobTask jobTask = jobTaskMapper.selectById(context.getTaskId());
Job job = jobMapper.selectById(context.getJobId());
if (jobTask == null || job == null) {
return;
}
if (jobTask.getRetryCount() < job.getMaxRetryTimes()) {
// 更新重试次数
JobTask updateJobTask = new JobTask();
updateJobTask.setRetryCount(1);
boolean success = SqlHelper.retBool(jobTaskMapper.update(updateJobTask, Wrappers.<JobTask>lambdaUpdate()
.lt(JobTask::getRetryCount, job.getMaxRetryTimes())
.eq(JobTask::getId, context.getTaskId())
));
if (success) {
}
return;
}
}
}
protected abstract void doCallback(ClientCallbackContext context);
@Override

27
easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/callback/BroadcastClientCallbackHandler.java
View File

@ -2,8 +2,10 @@ package com.aizuda.easy.retry.server.job.task.support.callback;
import akka.actor.ActorRef;
import com.aizuda.easy.retry.common.core.enums.JobTaskStatusEnum;
import com.aizuda.easy.retry.common.log.EasyRetryLog;
import com.aizuda.easy.retry.server.common.akka.ActorGenerator;
import com.aizuda.easy.retry.server.common.util.ClientInfoUtils;
import com.aizuda.easy.retry.server.job.task.dto.LogMetaDTO;
import com.aizuda.easy.retry.server.job.task.dto.RealJobExecutorDTO;
import com.aizuda.easy.retry.server.job.task.support.JobTaskConverter;
import com.aizuda.easy.retry.server.job.task.dto.JobExecutorResultDTO;
@ -39,30 +41,7 @@ public class BroadcastClientCallbackHandler extends AbstractClientCallbackHandle
@Override
protected void doCallback(final ClientCallbackContext context) {
if (context.getTaskStatus().equals(JobTaskStatusEnum.FAIL.getStatus())) {
JobTask jobTask = jobTaskMapper.selectById(context.getTaskId());
Job job = jobMapper.selectById(context.getJobId());
if (jobTask == null || job == null) {
return;
}
if (jobTask.getRetryCount() < job.getMaxRetryTimes()) {
// 更新重试次数
JobTask updateJobTask = new JobTask();
updateJobTask.setRetryCount(1);
boolean success = SqlHelper.retBool(jobTaskMapper.update(updateJobTask, Wrappers.<JobTask>lambdaUpdate()
.lt(JobTask::getRetryCount, job.getMaxRetryTimes())
.eq(JobTask::getId, context.getTaskId())
));
if (success) {
RealJobExecutorDTO realJobExecutor = JobTaskConverter.INSTANCE.toRealJobExecutorDTO(JobTaskConverter.INSTANCE.toJobExecutorContext(job), jobTask);
realJobExecutor.setClientId(ClientInfoUtils.clientId(jobTask.getClientInfo()));
ActorRef actorRef = ActorGenerator.jobRealTaskExecutorActor();
actorRef.tell(realJobExecutor, actorRef);
// TODO 记录日志
}
return;
}
}
JobExecutorResultDTO jobExecutorResultDTO = JobTaskConverter.INSTANCE.toJobExecutorResultDTO(context);
jobExecutorResultDTO.setTaskId(context.getTaskId());
jobExecutorResultDTO.setMessage(context.getExecuteResult().getMessage());

8
easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/callback/ClientCallbackContext.java
View File

@ -1,6 +1,8 @@
package com.aizuda.easy.retry.server.job.task.support.callback;
import com.aizuda.easy.retry.client.model.ExecuteResult;
import com.aizuda.easy.retry.template.datasource.persistence.po.Job;
import com.aizuda.easy.retry.template.datasource.persistence.po.JobTask;
import lombok.Data;
/**
@ -35,5 +37,11 @@ public class ClientCallbackContext {
private ExecuteResult executeResult;
private String clientInfo;
private Job job;
private JobTask jobTask;
}

52
easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/callback/ClusterClientCallbackHandler.java
View File

@ -3,11 +3,14 @@ package com.aizuda.easy.retry.server.job.task.support.callback;
import akka.actor.ActorRef;
import com.aizuda.easy.retry.common.core.enums.JobTaskStatusEnum;
import com.aizuda.easy.retry.common.core.enums.JobTaskTypeEnum;
import com.aizuda.easy.retry.common.log.EasyRetryLog;
import com.aizuda.easy.retry.server.common.akka.ActorGenerator;
import com.aizuda.easy.retry.server.common.cache.CacheRegisterTable;
import com.aizuda.easy.retry.server.common.dto.RegisterNodeInfo;
import com.aizuda.easy.retry.server.common.handler.ClientNodeAllocateHandler;
import com.aizuda.easy.retry.server.common.util.ClientInfoUtils;
import com.aizuda.easy.retry.server.job.task.dto.JobExecutorResultDTO;
import com.aizuda.easy.retry.server.job.task.dto.LogMetaDTO;
import com.aizuda.easy.retry.server.job.task.dto.RealJobExecutorDTO;
import com.aizuda.easy.retry.server.job.task.support.JobTaskConverter;
import com.aizuda.easy.retry.template.datasource.persistence.mapper.JobMapper;
@ -21,6 +24,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Objects;
import java.util.Set;
/**
* @author www.byteblogs.com
@ -43,43 +47,23 @@ public class ClusterClientCallbackHandler extends AbstractClientCallbackHandler
return JobTaskTypeEnum.CLUSTER;
}
@Override
protected String chooseNewClient(ClientCallbackContext context) {
// 选择重试的节点
RegisterNodeInfo serverNode = clientNodeAllocateHandler.getServerNode(context.getJobId().toString(),
context.getGroupName(), context.getNamespaceId(), context.getJob().getRouteKey());
if (Objects.isNull(serverNode)) {
log.error("无可执行的客户端信息. jobId:[{}]", context.getJobId());
return null;
}
return ClientInfoUtils.generate(serverNode);
}
@Override
protected void doCallback(ClientCallbackContext context) {
if (context.getTaskStatus().equals(JobTaskStatusEnum.FAIL.getStatus())) {
JobTask jobTask = jobTaskMapper.selectById(context.getTaskId());
Job job = jobMapper.selectById(context.getJobId());
if (jobTask == null || job == null) {
return;
}
if (jobTask.getRetryCount() < job.getMaxRetryTimes()) {
// 选择重试的节点
RegisterNodeInfo serverNode = clientNodeAllocateHandler.getServerNode(context.getJobId().toString(),
context.getGroupName(), context.getNamespaceId(), job.getRouteKey());
if (Objects.isNull(serverNode)) {
log.error("无可执行的客户端信息. jobId:[{}]", context.getJobId());
return;
}
String newClient = ClientInfoUtils.generate(serverNode);
// 更新重试次数
JobTask updateJobTask = new JobTask();
updateJobTask.setClientInfo(newClient);
updateJobTask.setRetryCount(1);
boolean success = SqlHelper.retBool(jobTaskMapper.update(updateJobTask, Wrappers.<JobTask>lambdaUpdate()
.lt(JobTask::getRetryCount, job.getMaxRetryTimes())
.eq(JobTask::getId, context.getTaskId())
));
// 更新成功执行重试
if (success) {
RealJobExecutorDTO realJobExecutor = JobTaskConverter.INSTANCE.toRealJobExecutorDTO(JobTaskConverter.INSTANCE.toJobExecutorContext(job), jobTask);
realJobExecutor.setClientId(ClientInfoUtils.clientId(newClient));
ActorRef actorRef = ActorGenerator.jobRealTaskExecutorActor();
actorRef.tell(realJobExecutor, actorRef);
// TODO 记录日志
}
return;
}
}
JobExecutorResultDTO jobExecutorResultDTO = JobTaskConverter.INSTANCE.toJobExecutorResultDTO(context);
jobExecutorResultDTO.setTaskId(context.getTaskId());
jobExecutorResultDTO.setMessage(context.getExecuteResult().getMessage());

45
easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/callback/ShardingClientCallbackHandler.java
View File

@ -45,40 +45,6 @@ public class ShardingClientCallbackHandler extends AbstractClientCallbackHandler
@Override
protected void doCallback(final ClientCallbackContext context) {
if (context.getTaskStatus().equals(JobTaskStatusEnum.FAIL.getStatus())) {
JobTask jobTask = jobTaskMapper.selectById(context.getTaskId());
Job job = jobMapper.selectById(context.getJobId());
if (jobTask == null || job == null) {
return;
}
if (jobTask.getRetryCount() < job.getMaxRetryTimes()) {
Set<RegisterNodeInfo> nodes = CacheRegisterTable.getServerNodeSet(context.getGroupName(), context.getNamespaceId());
if (CollUtil.isEmpty(nodes)) {
log.error("无可执行的客户端信息. jobId:[{}]", context.getJobId());
return;
}
RegisterNodeInfo serverNode = RandomUtil.randomEle(nodes.toArray(new RegisterNodeInfo[0]));
String newClient = ClientInfoUtils.generate(serverNode);
// 更新重试次数
JobTask updateJobTask = new JobTask();
updateJobTask.setClientInfo(newClient);
updateJobTask.setRetryCount(1);
boolean success = SqlHelper.retBool(
jobTaskMapper.update(updateJobTask, Wrappers.<JobTask>lambdaUpdate()
.lt(JobTask::getRetryCount, job.getMaxRetryTimes())
.eq(JobTask::getId, context.getTaskId())
)
);
if (success) {
RealJobExecutorDTO realJobExecutor = JobTaskConverter.INSTANCE.toRealJobExecutorDTO(JobTaskConverter.INSTANCE.toJobExecutorContext(job), jobTask);
realJobExecutor.setClientId(ClientInfoUtils.clientId(newClient));
ActorRef actorRef = ActorGenerator.jobRealTaskExecutorActor();
actorRef.tell(realJobExecutor, actorRef);
// TODO 记录日志
}
return;
}
}
JobExecutorResultDTO jobExecutorResultDTO = JobTaskConverter.INSTANCE.toJobExecutorResultDTO(context);
jobExecutorResultDTO.setTaskId(context.getTaskId());
@ -90,4 +56,15 @@ public class ShardingClientCallbackHandler extends AbstractClientCallbackHandler
actorRef.tell(jobExecutorResultDTO, actorRef);
}
@Override
protected String chooseNewClient(ClientCallbackContext context) {
Set<RegisterNodeInfo> nodes = CacheRegisterTable.getServerNodeSet(context.getGroupName(), context.getNamespaceId());
if (CollUtil.isEmpty(nodes)) {
log.error("无可执行的客户端信息. jobId:[{}]", context.getJobId());
return null;
}
RegisterNodeInfo serverNode = RandomUtil.randomEle(nodes.toArray(new RegisterNodeInfo[0]));
return ClientInfoUtils.generate(serverNode);
}
}

8
easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/dispatch/JobExecutorResultActor.java
View File

@ -88,10 +88,10 @@ public class JobExecutorResultActor extends AbstractActor {
}
});
LogMetaDTO logMetaDTO = JobTaskConverter.INSTANCE.toJobLogDTO(result);
// 防止客户端日志还未上报完成，导致日志时序错误
logMetaDTO.setTimestamp(DateUtils.toEpochMilli(LocalDateTime.now().plusHours(1)));
EasyRetryLog.REMOTE.info("taskId:[{}] 任务执行成功. <|>{}<|>", logMetaDTO.getTaskId(), logMetaDTO);
// LogMetaDTO logMetaDTO = JobTaskConverter.INSTANCE.toJobLogDTO(result);
// // 防止客户端日志还未上报完成，导致日志时序错误
// logMetaDTO.setTimestamp(DateUtils.toEpochMilli(LocalDateTime.now().plusHours(1)));
// EasyRetryLog.REMOTE.info("taskId:[{}] 任务执行成功. <|>{}<|>", logMetaDTO.getTaskId(), logMetaDTO);
} catch (Exception e) {
LogUtils.error(log, " job executor result exception. [{}]", result, e);

3
easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/job/RequestClientActor.java
View File

@ -14,6 +14,7 @@ import com.aizuda.easy.retry.server.common.akka.ActorGenerator;
import com.aizuda.easy.retry.server.common.cache.CacheRegisterTable;
import com.aizuda.easy.retry.server.common.client.RequestBuilder;
import com.aizuda.easy.retry.server.common.dto.RegisterNodeInfo;
import com.aizuda.easy.retry.server.common.util.DateUtils;
import com.aizuda.easy.retry.server.job.task.client.JobRpcClient;
import com.aizuda.easy.retry.server.job.task.dto.JobExecutorResultDTO;
import com.aizuda.easy.retry.server.job.task.dto.JobLogDTO;
@ -69,10 +70,12 @@ public class RequestClientActor extends AbstractActor {
try {
// 构建请求客户端对象
Long timestamp = DateUtils.toNowMilli();
JobRpcClient rpcClient = buildRpcClient(registerNodeInfo, realJobExecutorDTO);
Result<Boolean> dispatch = rpcClient.dispatch(dispatchJobRequest);
if (dispatch.getStatus() == StatusEnum.YES.getStatus() && Objects.equals(dispatch.getData(), Boolean.TRUE)) {
LogMetaDTO logMetaDTO = JobTaskConverter.INSTANCE.toJobLogDTO(realJobExecutorDTO);
logMetaDTO.setTimestamp(timestamp);
EasyRetryLog.REMOTE.info("taskId:[{}] 任务调度成功. <|>{}<|>", logMetaDTO.getTaskId(), logMetaDTO);
} else {
// 客户端返回失败，则认为任务执行失败

4
easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/CallbackWorkflowExecutor.java
View File

@ -92,9 +92,7 @@ public class CallbackWorkflowExecutor extends AbstractWorkflowExecutor {
result = exchange.getBody();
log.info("回调结果. webHook:[{}]，参数: [{}]", decisionConfig.getWebhook(), result);
} catch (Exception e) {
// log.error("回调异常. webHook:[{}]，参数: [{}]", decisionConfig.getWebhook(), context.getTaskResult(), e);
EasyRetryLog.LOCAL.error("回调异常. webHook:[{}]，参数: [{}]", decisionConfig.getWebhook(), context.getTaskResult(), e);
EasyRetryLog.REMOTE.error("回调异常. webHook:[{}]，参数: [{}]", decisionConfig.getWebhook(), context.getTaskResult(), e);
taskBatchStatus = JobTaskBatchStatusEnum.FAIL.getStatus();
operationReason = JobOperationReasonEnum.WORKFLOW_CALLBACK_NODE_EXECUTOR_ERROR.getReason();
jobTaskStatus = JobTaskStatusEnum.FAIL.getStatus();

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

@ -27,6 +27,11 @@ public class JobBatchResponseVO {
*/
private String jobName;
/**
* 工作流节点名称
*/
private String nodeName;
/**
* 任务信息id
*/

33
easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/service/impl/JobBatchServiceImpl.java
View File

@ -102,25 +102,26 @@ public class JobBatchServiceImpl implements JobBatchService {
return null;
}
if (jobTaskBatch.getTaskType().equals(TaskTypeEnum.JOB.getType())) {
Job job = jobMapper.selectById(jobTaskBatch.getJobId());
return JobBatchResponseVOConverter.INSTANCE.toJobBatchResponseVO(jobTaskBatch, job);
}
Job job = jobMapper.selectById(jobTaskBatch.getJobId());
JobBatchResponseVO jobBatchResponseVO = JobBatchResponseVOConverter.INSTANCE.toJobBatchResponseVO(jobTaskBatch, job);
JobBatchResponseVO jobBatchResponseVO = JobBatchResponseVOConverter.INSTANCE.toJobBatchResponseVO(jobTaskBatch);
// 回调节点
if (SystemConstants.CALLBACK_JOB_ID.equals(jobTaskBatch.getJobId())) {
if (jobTaskBatch.getTaskType().equals(TaskTypeEnum.WORKFLOW.getType())) {
WorkflowNode workflowNode = workflowNodeMapper.selectById(jobTaskBatch.getWorkflowNodeId());
jobBatchResponseVO.setJobName(workflowNode.getNodeName());
jobBatchResponseVO.setCallback(JsonUtil.parseObject(workflowNode.getNodeInfo(), CallbackConfig.class));
}
jobBatchResponseVO.setNodeName(workflowNode.getNodeName());
// 条件节点
if (SystemConstants.DECISION_JOB_ID.equals(jobTaskBatch.getJobId())) {
WorkflowNode workflowNode = workflowNodeMapper.selectById(jobTaskBatch.getWorkflowNodeId());
jobBatchResponseVO.setJobName(workflowNode.getNodeName());
jobBatchResponseVO.setDecision(JsonUtil.parseObject(workflowNode.getNodeInfo(), DecisionConfig.class));
// 回调节点
if (SystemConstants.CALLBACK_JOB_ID.equals(jobTaskBatch.getJobId())) {
jobBatchResponseVO.setCallback(JsonUtil.parseObject(workflowNode.getNodeInfo(), CallbackConfig.class));
jobBatchResponseVO.setExecutionAt(jobTaskBatch.getCreateDt());
return jobBatchResponseVO;
}
// 条件节点
if (SystemConstants.DECISION_JOB_ID.equals(jobTaskBatch.getJobId())) {
jobBatchResponseVO.setDecision(JsonUtil.parseObject(workflowNode.getNodeInfo(), DecisionConfig.class));
jobBatchResponseVO.setExecutionAt(jobTaskBatch.getCreateDt());
return jobBatchResponseVO;
}
}
return jobBatchResponseVO;

29
easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/service/impl/JobLogServiceImpl.java
View File

@ -1,14 +1,18 @@
package com.aizuda.easy.retry.server.web.service.impl;
import com.aizuda.easy.retry.common.core.enums.JobTaskBatchStatusEnum;
import com.aizuda.easy.retry.common.core.util.JsonUtil;
import com.aizuda.easy.retry.server.web.model.request.JobLogQueryVO;
import com.aizuda.easy.retry.server.web.model.response.JobLogResponseVO;
import com.aizuda.easy.retry.server.web.service.JobLogService;
import com.aizuda.easy.retry.template.datasource.persistence.mapper.JobLogMessageMapper;
import com.aizuda.easy.retry.template.datasource.persistence.mapper.JobTaskBatchMapper;
import com.aizuda.easy.retry.template.datasource.persistence.po.JobLogMessage;
import com.aizuda.easy.retry.template.datasource.persistence.po.JobTaskBatch;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.PageDTO;
import com.google.common.collect.Lists;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
@ -23,10 +27,10 @@ import java.util.stream.Collectors;
* @since ：2.4.0
*/
@Service
@RequiredArgsConstructor
public class JobLogServiceImpl implements JobLogService {
@Autowired
private JobLogMessageMapper jobLogMessageMapper;
private final JobLogMessageMapper jobLogMessageMapper;
private final JobTaskBatchMapper jobTaskBatchMapper;
@Override
public JobLogResponseVO getJobLogPage(final JobLogQueryVO queryVO) {
@ -35,16 +39,25 @@ public class JobLogServiceImpl implements JobLogService {
LambdaQueryWrapper<JobLogMessage> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper
.select(JobLogMessage::getId, JobLogMessage::getLogNum)
.ge(JobLogMessage::getId, queryVO.getStartId())
.eq(JobLogMessage::getTaskId, queryVO.getTaskId());
.select(JobLogMessage::getId, JobLogMessage::getLogNum)
.ge(JobLogMessage::getId, queryVO.getStartId())
.eq(JobLogMessage::getTaskId, queryVO.getTaskId());
queryWrapper.orderByAsc(JobLogMessage::getRealTime).orderByDesc(JobLogMessage::getId);
PageDTO<JobLogMessage> selectPage = jobLogMessageMapper.selectPage(pageDTO, queryWrapper);
List<JobLogMessage> records = selectPage.getRecords();
if (CollectionUtils.isEmpty(records)) {
Long count = jobTaskBatchMapper.selectCount(
new LambdaQueryWrapper<JobTaskBatch>()
.in(JobTaskBatch::getTaskBatchStatus, JobTaskBatchStatusEnum.COMPLETED)
);
JobLogResponseVO jobLogResponseVO = new JobLogResponseVO();
jobLogResponseVO.setFinished(Boolean.TRUE);
if (count > 0) {
jobLogResponseVO.setFinished(Boolean.TRUE);
}
return jobLogResponseVO;
}
@ -70,7 +83,7 @@ public class JobLogServiceImpl implements JobLogService {
List<String> originalList = JsonUtil.parseObject(jobLogMessage.getMessage(), List.class);
int size = originalList.size() - fromIndex;
List<String> pageList = originalList.stream().skip(fromIndex).limit(queryVO.getSize())
.collect(Collectors.toList());
.collect(Collectors.toList());
if (messages.size() + size >= queryVO.getSize()) {
messages.addAll(pageList);

17
easy-retry-server/easy-retry-server-web/src/main/java/com/aizuda/easy/retry/server/web/service/impl/SystemUserServiceImpl.java
View File

@ -31,6 +31,7 @@ import com.aizuda.easy.retry.server.web.model.request.SystemUserQueryVO;
import com.aizuda.easy.retry.server.web.model.request.SystemUserRequestVO;
import com.aizuda.easy.retry.server.web.model.response.SystemUserResponseVO;
import com.google.common.collect.Lists;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@ -48,18 +49,14 @@ import java.util.stream.Collectors;
* @since 2022-03-05
*/
@Service
@RequiredArgsConstructor
public class SystemUserServiceImpl implements SystemUserService {
public static final long EXPIRE_TIME = 3600 * 24 * 1000;
public static final long EXPIRE_TIME = 3600 * 7 * 1000;
@Autowired
private SystemUserMapper systemUserMapper;
@Autowired
private SystemUserPermissionMapper systemUserPermissionMapper;
@Autowired
private NamespaceMapper namespaceMapper;
@Autowired
private SystemProperties systemProperties;
private final SystemUserMapper systemUserMapper;
private final SystemUserPermissionMapper systemUserPermissionMapper;
private final NamespaceMapper namespaceMapper;
private final SystemProperties systemProperties;
@Override
public SystemUserResponseVO login(SystemUserRequestVO requestVO) {

4
frontend/public/lib/assets/MlqLg9-Y.js → frontend/public/lib/assets/YpO3912B.js
View File

File diff suppressed because one or more lines are too long

2
frontend/public/lib/assets/vhU7tY2l.css → frontend/public/lib/assets/yQfA4Ykm.css
View File

File diff suppressed because one or more lines are too long

4
frontend/public/lib/index.html
View File

@ -5,8 +5,8 @@
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Easy Retry</title>
<script type="module" crossorigin src="./assets/MlqLg9-Y.js"></script>
<link rel="stylesheet" crossorigin href="./assets/vhU7tY2l.css">
<script type="module" crossorigin src="./assets/YpO3912B.js"></script>
<link rel="stylesheet" crossorigin href="./assets/yQfA4Ykm.css">
</head>
<body>