工作流中调用工作流节点

DirectoryV3.xml

@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?> 
+ <trees>
+     <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/dispatch/WorkflowExecutorActor.java" title="工作流执行"/>
+     <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/dispatch/JobTaskPrepareActor.java" title="调度任务准备阶段"/>
+     <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/workflow/AbstractWorkflowExecutor.java" title="处理工作流中各种节点类型"/>
+     <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/workflow/CallbackWorkflowExecutor.java" title="回调节点"/>
+     <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/workflow/DecisionWorkflowExecutor.java" title="条件节点"/>
+     <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/workflow/JobTaskWorkflowExecutor.java" title="任务节点"/>
+     <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/workflow/WorkflowWorkflowExecutor.java" title="工作流节点"/>
+     <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/workflow" title="工作流节点"/>
+     <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/job/ClusterJobExecutor.java" title="集群任务"/>
+     <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/job/BroadcastTaskJobExecutor.java" title="广播任务"/>
+     <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/job/MapJobExecutor.java" title="Map任务"/>
+    <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/job/MapReduceJobExecutor.java" title="MapReduce任务"/>
+    <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/job/RequestClientActor.java" title="调用客户端执行任务"/>
+    <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/job/ShardingJobExecutor.java" title="分片任务"/>
+    <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/job" title="任务执行"/>
+    <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/timer/WorkflowTimeoutCheckTask.java" title="工作流任务超时检查"/>
+    <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/callback/AbstractClientCallbackHandler.java" title="客户端执行任务完成回调"/>
+    <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/callback/ClusterClientCallbackHandler.java" title="集群任务执行结果回调"/>
+    <tree path="/snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/dispatch/JobExecutorResultActor.java" title="任务执行结果处理"/>
+</trees>

snail-job-common/snail-job-common-core/src/main/java/com/aizuda/snailjob/common/core/enums/JobOperationReasonEnum.java

@@ -81,11 +81,26 @@ public enum JobOperationReasonEnum {
      * 工作流决策未通过
      */
     WORKFLOW_DECISION_FAILED(14, "Judgment not passed"),
+    /**
+     * 手动调用
+     */
+    MANUAL_CALL(15, "Manual call"),
+    /**
+     * 由工作流中被调用
+     */
+    WORKFLOW_CALLED(16, "Called by workflow"),
 
 
     ;
 
+    /**
+     * 原因
+     */
     private final int reason;
+
+    /**
+     * 描述
+     */
     private final String desc;
 
     /**

snail-job-common/snail-job-common-core/src/main/java/com/aizuda/snailjob/common/core/enums/JobTaskBatchStatusEnum.java

@@ -51,7 +51,7 @@ public enum JobTaskBatchStatusEnum {
     public static final List<Integer> NOT_COMPLETE = Arrays.asList(WAITING.status, RUNNING.status);
 
     /**
-     * 任务完成 状态 包含 SUCCESS, FAIL, STOP, CANCEL
+     * 任务完成 状态 包含 SUCCESS 3, FAIL 4, STOP 5, CANCEL 6
      */
     public static final List<Integer> COMPLETED = Arrays.asList(SUCCESS.status, FAIL.status, STOP.status, CANCEL.status);
 

snail-job-common/snail-job-common-core/src/main/java/com/aizuda/snailjob/common/core/enums/StatusEnum.java

@@ -13,7 +13,13 @@ import lombok.Getter;
 @Getter
 public enum StatusEnum {
 
+    /**
+     * 0、关闭、1、开启
+     */
     NO(0),
+    /**
+     * 0、关闭、1、开启
+     */
     YES(1);
 
     private final Integer status;

snail-job-server/snail-job-server-common/src/main/java/com/aizuda/snailjob/server/common/pekko/ActorGenerator.java

@@ -261,7 +261,7 @@ public class ActorGenerator {
 
     /**
      * Job任务执行阶段actor
-     *
+     * @see com.aizuda.snailjob.server.job.task.support.dispatch.JobExecutorActor
      * @return actor 引用
      */
     public static ActorRef jobTaskExecutorActor() {
@@ -288,7 +288,7 @@ public class ActorGenerator {
 
     /**
      * Job任务执行结果actor
-     *
+     * @see com.aizuda.snailjob.server.job.task.support.dispatch.JobExecutorResultActor
      * @return actor 引用
      */
     public static ActorRef jobTaskExecutorResultActor() {

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/callback/AbstractClientCallbackHandler.java

@@ -28,6 +28,8 @@ import java.time.Duration;
 import java.util.Objects;
 
 /**
+ * 客户端执行任务完成回调处理
+ *
  * @author opensnail
  * @date 2023-10-03 23:12:33
  * @since 2.4.0
@@ -128,6 +130,9 @@ public abstract class AbstractClientCallbackHandler implements ClientCallbackHan
 
     }
 
+    /**
+     * 判定是否需要重试
+     */
     private boolean isNeedRetry(ClientCallbackContext context) {
 
         JobTask jobTask = jobTaskMapper.selectById(context.getTaskId());

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/callback/ClientCallbackFactory.java

@@ -18,6 +18,9 @@ public class ClientCallbackFactory {
         CACHE.put(taskInstanceType, callbackHandler);
     }
 
+    /**
+     * 根据任务类型获取任务执行器
+     */
     public static ClientCallbackHandler getClientCallback(Integer type) {
         return CACHE.get(JobTaskTypeEnum.valueOf(type));
     }

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/callback/ClusterClientCallbackHandler.java

@@ -15,6 +15,8 @@ import org.springframework.stereotype.Component;
 import java.util.Objects;
 
 /**
+ * 集群任务执行结果回调处理
+ *
  * @author opensnail
  * @date 2023-10-03 23:12:12
  * @since 2.4.0
@@ -44,6 +46,10 @@ public class ClusterClientCallbackHandler extends AbstractClientCallbackHandler
         return ClientInfoUtils.generate(serverNode);
     }
 
+    /**
+     * 调用Job任务执行结果actor
+     * @see com.aizuda.snailjob.server.job.task.support.dispatch.JobExecutorResultActor
+     */
     @Override
     protected void doCallback(ClientCallbackContext context) {
 

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/dispatch/JobExecutorActor.java

@@ -110,6 +110,7 @@ public class JobExecutorActor extends AbstractActor {
         Job job = jobMapper.selectOne(queryWrapper.eq(Job::getId, taskExecute.getJobId()));
         int taskStatus = JobTaskBatchStatusEnum.RUNNING.getStatus();
         try {
+            // 操作原因
             int operationReason = JobOperationReasonEnum.NONE.getReason();
             if (Objects.isNull(job)) {
                 taskStatus = JobTaskBatchStatusEnum.CANCEL.getStatus();

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/dispatch/JobExecutorResultActor.java

@@ -22,6 +22,8 @@ import org.springframework.stereotype.Component;
 import java.util.Objects;
 
 /**
+ * 任务执行结果处理
+ *
  * @author opensnail
  * @date 2023-10-05 17:16:35
  * @since 2.4.0
@@ -74,6 +76,9 @@ public class JobExecutorResultActor extends AbstractActor {
 
     }
 
+    /**
+     * 尝试完成任务
+     */
     private void tryCompleteAndStop(JobExecutorResultDTO jobExecutorResultDTO) {
         CompleteJobBatchDTO completeJobBatchDTO = JobTaskConverter.INSTANCE.toCompleteJobBatchDTO(jobExecutorResultDTO);
         jobTaskBatchHandler.handleResult(completeJobBatchDTO);

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/dispatch/JobTaskPrepareActor.java

@@ -53,7 +53,7 @@ public class JobTaskPrepareActor extends AbstractActor {
     private void doPrepare(JobTaskPrepareDTO prepare) {
         LambdaQueryWrapper<JobTaskBatch> queryWrapper = new LambdaQueryWrapper<JobTaskBatch>()
                 .eq(JobTaskBatch::getJobId, prepare.getJobId())
-                .in(JobTaskBatch::getTaskBatchStatus, NOT_COMPLETE);
+                .in(JobTaskBatch::getTaskBatchStatus, NOT_COMPLETE);//未完成状态
 
         JobTaskExecutorSceneEnum jobTaskExecutorSceneEnum = JobTaskExecutorSceneEnum.get(
                 prepare.getTaskExecutorScene());

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/dispatch/WorkflowExecutorActor.java

@@ -47,6 +47,8 @@ import java.util.stream.Collectors;
 import static com.aizuda.snailjob.common.core.enums.JobOperationReasonEnum.WORKFLOW_SUCCESSOR_SKIP_EXECUTION;
 
 /**
+ * 工作流执行器
+ *
  * @author: xiaowoniu
  * @date : 2023-12-22 10:34
  * @since : 2.6.0
@@ -255,6 +257,9 @@ public class WorkflowExecutorActor extends AbstractActor {
         }
     }
 
+    /**
+     * 前置任务是否已完成
+     */
     private boolean arePredecessorsComplete(final WorkflowNodeTaskExecuteDTO taskExecute, Set<Long> predecessors,
                                             Map<Long, List<JobTaskBatch>> jobTaskBatchMap, WorkflowNode waitExecWorkflowNode,
                                             Map<Long, WorkflowNode> workflowNodeMap) {
@@ -303,6 +308,12 @@ public class WorkflowExecutorActor extends AbstractActor {
         return Boolean.TRUE;
     }
 
+    /**
+     * 更新工作流任务批次
+     *
+     * @param taskStatus      任务批次状态
+     * @param operationReason 操作原因
+     */
     private void handlerTaskBatch(WorkflowNodeTaskExecuteDTO taskExecute, int taskStatus, int operationReason) {
 
         WorkflowTaskBatch jobTaskBatch = new WorkflowTaskBatch();

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/executor/workflow/WorkflowWorkflowExecutor.java

@@ -1,12 +1,19 @@
 package com.aizuda.snailjob.server.job.task.support.executor.workflow;
 
+import com.aizuda.snailjob.common.core.constant.SystemConstants;
 import com.aizuda.snailjob.common.core.enums.JobOperationReasonEnum;
 import com.aizuda.snailjob.common.core.enums.JobTaskBatchStatusEnum;
 import com.aizuda.snailjob.common.core.enums.JobTaskStatusEnum;
 import com.aizuda.snailjob.common.core.enums.WorkflowNodeTypeEnum;
+import com.aizuda.snailjob.common.log.SnailJobLog;
+import com.aizuda.snailjob.server.common.dto.JobLogMetaDTO;
+import com.aizuda.snailjob.template.datasource.persistence.po.JobTask;
+import com.aizuda.snailjob.template.datasource.persistence.po.JobTaskBatch;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;
 
+import static com.aizuda.snailjob.common.core.enums.JobOperationReasonEnum.WORKFLOW_SUCCESSOR_SKIP_EXECUTION;
+
 @Component
 @RequiredArgsConstructor
 public class WorkflowWorkflowExecutor extends AbstractWorkflowExecutor {
@@ -23,6 +30,27 @@ public class WorkflowWorkflowExecutor extends AbstractWorkflowExecutor {
 
     @Override
     protected void afterExecute(WorkflowExecutorContext context) {
+        JobTaskBatch jobTaskBatch = generateJobTaskBatch(context);
+
+        JobTask jobTask = generateJobTask(context, jobTaskBatch);
+
+        JobLogMetaDTO jobLogMetaDTO = new JobLogMetaDTO();
+        jobLogMetaDTO.setNamespaceId(context.getNamespaceId());
+        jobLogMetaDTO.setGroupName(context.getGroupName());
+        jobLogMetaDTO.setTaskBatchId(jobTaskBatch.getId());
+        jobLogMetaDTO.setJobId(SystemConstants.WORKFLOW_JOB_ID);
+        jobLogMetaDTO.setTaskId(jobTask.getId());
+        if (jobTaskBatch.getTaskBatchStatus() == JobTaskStatusEnum.SUCCESS.getStatus()) {
+            SnailJobLog.REMOTE.info("Node [{}] workflow success.\nworkflow params: {} \nworkflow result: [{}] <|>{}<|>", context.getWorkflowNodeId(), context.getWfContext(), context.getEvaluationResult(), jobLogMetaDTO);
+        } else if (jobTaskBatch.getTaskBatchStatus() == JobTaskStatusEnum.CANCEL.getStatus()) {
+            if (WORKFLOW_SUCCESSOR_SKIP_EXECUTION.contains(context.getParentOperationReason())) {
+                SnailJobLog.REMOTE.warn("Node [{}] cancels workflow. Cancellation reason: Current task does not require processing <|>{}<|>", context.getWorkflowNodeId(), jobLogMetaDTO);
+            } else {
+                SnailJobLog.REMOTE.warn("Node [{}] cancels workflow. Cancellation reason: Task status is closed <|>{}<|>", context.getWorkflowNodeId(), jobLogMetaDTO);
+            }
+        } else {
+            SnailJobLog.REMOTE.error("Node [{}] fail to workflow.\nReason: {} <|>{}<|>", context.getWorkflowNodeId(), context.getLogMessage(), jobLogMetaDTO);
+        }
     }
 
     @Override

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/handler/JobTaskBatchHandler.java

@@ -54,6 +54,9 @@ public class JobTaskBatchHandler {
     private final GroupConfigMapper groupConfigMapper;
     private final List<JobExecutorResultHandler> resultHandlerList;
 
+    /**
+     * 处理任务批次结果
+     */
     @Transactional
     public boolean handleResult(CompleteJobBatchDTO completeJobBatchDTO) {
         Assert.notNull(completeJobBatchDTO.getTaskType(), ()-> new SnailJobServerException("taskType can not be null"));

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/handler/WorkflowBatchHandler.java

@@ -103,6 +103,9 @@ public class WorkflowBatchHandler {
         return complete(workflowTaskBatchId, null);
     }
 
+    /**
+     * 工作流执行完成
+     */
     public boolean complete(Long workflowTaskBatchId, WorkflowTaskBatch workflowTaskBatch) {
         workflowTaskBatch = Optional.ofNullable(workflowTaskBatch)
                 .orElseGet(() -> workflowTaskBatchMapper.selectById(workflowTaskBatchId));
@@ -175,6 +178,9 @@ public class WorkflowBatchHandler {
 
     }
 
+    /**
+     * 修改工作流任务批次状态
+     */
     private void handlerTaskBatch(Long workflowTaskBatchId, int taskStatus, int operationReason) {
 
         WorkflowTaskBatch jobTaskBatch = new WorkflowTaskBatch();
@@ -312,6 +318,9 @@ public class WorkflowBatchHandler {
         }
     }
 
+    /**
+     * 开启下一个工作流节点
+     */
     public void openNextNode(WorkflowNodeTaskExecuteDTO taskExecuteDTO) {
         if (Objects.isNull(taskExecuteDTO.getParentId()) || Objects.isNull(taskExecuteDTO.getWorkflowTaskBatchId()) || Long.valueOf(0).equals(taskExecuteDTO.getWorkflowTaskBatchId())) {
             return;
@@ -330,12 +339,16 @@ public class WorkflowBatchHandler {
         }
     }
 
+    /**
+     * 通知工作流执行器
+     * @see com.aizuda.snailjob.server.job.task.support.dispatch.WorkflowExecutorActor
+     */
     private void tellWorkflowTaskExecutor(WorkflowNodeTaskExecuteDTO taskExecuteDTO) {
         try {
             ActorRef actorRef = ActorGenerator.workflowTaskExecutorActor();
             actorRef.tell(taskExecuteDTO, actorRef);
         } catch (Exception e) {
-            SnailJobLog.LOCAL.error("Task scheduling execution failed", e);
+            SnailJobLog.LOCAL.error("Task scheduling execution failed", e); //任务调度执行失败
         }
     }
 

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/prepare/job/TerminalJobPrepareHandler.java

@@ -29,6 +29,9 @@ public class TerminalJobPrepareHandler extends AbstractJobPrepareHandler {
     @Autowired
     private JobTaskBatchGenerator jobTaskBatchGenerator;
 
+    /**
+     * 任务完成状态
+     */
     @Override
     public boolean matches(Integer status) {
         return COMPLETED.contains(status);

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/prepare/job/WaitJobPrepareHandler.java

@@ -13,7 +13,7 @@ import java.text.MessageFormat;
 import java.time.Duration;
 
 /**
- * 处理处于{@link JobTaskBatchStatusEnum::WAIT}状态的任务
+ * 处理处于{@link JobTaskBatchStatusEnum::WAIT}待处理状态的任务
  *
  * @author opensnail
  * @date 2023-10-05 18:29:22

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/request/ReportDispatchResultPostHttpRequestHandler.java

@@ -23,6 +23,8 @@ import java.util.Optional;
 import static com.aizuda.snailjob.common.core.constant.SystemConstants.HTTP_PATH.REPORT_JOB_DISPATCH_RESULT;
 
 /**
+ * 客户端执行完成后上报结果
+ *
  * @author opensnail
  * @date 2023-09-30 23:01:58
  * @since 2.4.0

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/result/job/AbstractJobExecutorResultHandler.java

@@ -84,7 +84,7 @@ public abstract class AbstractJobExecutorResultHandler implements JobExecutorRes
             doHandleSuccess(context);
         }
 
-        // 开启下一个工作流节点
+        // 开启下一个工作流节点（如果需要）
         openNextWorkflowNode(context);
 
         boolean res = updateStatus(context, taskBatchStatus);
@@ -104,6 +104,9 @@ public abstract class AbstractJobExecutorResultHandler implements JobExecutorRes
         workflowBatchHandler.openNextNode(taskExecuteDTO);
     }
 
+    /**
+     * 更新任务批次状态
+     */
     protected boolean updateStatus(final JobExecutorResultContext context, final Integer taskBatchStatus) {
         JobTaskBatch jobTaskBatch = new JobTaskBatch();
         jobTaskBatch.setId(context.getTaskBatchId());
@@ -135,6 +138,9 @@ public abstract class AbstractJobExecutorResultHandler implements JobExecutorRes
         instanceInterrupt.stop(stopJobContext);
     }
 
+    /**
+     * 成功（除MapReduce任务外啥也没干）
+     */
     protected abstract void doHandleSuccess(final JobExecutorResultContext context);
 
     protected abstract void doHandleStop(final JobExecutorResultContext context);

snail-job-server/snail-job-server-job-task/src/main/java/com/aizuda/snailjob/server/job/task/support/timer/WorkflowTimeoutCheckTask.java

@@ -18,6 +18,7 @@ import java.text.MessageFormat;
 import java.util.Objects;
 
 /**
+ * 工作流任务超时检查任务
  * @author opensnail
  * @date 2024-05-20 22:25:12
  * @since sj_1.0.0