去除PowerJob组件依赖

[baisui1981]

使用PowerJob组件看起来效果并不理想，由于依赖了该第三方组件内部运行还有很多不稳定因素，所以需要将他去除掉。

任务调度单机版和分布式集群版执行模式会使用AKKA的框架进行统一，TIS会开发一个机遇AKKA的DAG任务调度功能模块

相关资料

1. 支持批量任务分布式执行 #157
2. 触发Submit相关的类图：https://www.processon.com/view/link/6643346b7004e7751df6c6b4
3. DataX任务执行流程：https://www.processon.com/view/link/623429841efad407524d0fbe

为了执行DAG任务调度使用task-reactor

            <dependency>
                <groupId>org.jenkins-ci</groupId>
                <artifactId>task-reactor</artifactId>
                <version>1.5</version>
            </dependency>

DAG任务关系包括在 tis-sql-parser/src/main/java/com/qlangtech/tis/sql/parser/DAGSessionSpec.java 中

需求分析

1. PowerJob工作流DAG拓扑结构核心类分析 https://github.com/datavane/tis/blob/master/design/dag/powerjob-workflow-dag-analysis.md
2. TIS DAG任务调度改造方案 https://github.com/datavane/tis/blob/master/design/dag/tis-dag-refactoring-plan.md

[baisui1981]

Actor 消息传递拓扑图

https://www.processon.com/view/link/6981838d3635662b44b11b5c 编辑

主流程：工作流执行

Client/API
  → WorkflowInstance Sharding Region (StartWorkflow, 直接发送)
    → WorkflowInstanceActor (创建/路由)
      → NodeDispatcherActor (DispatchTask)
        → TaskWorker ClusterRouterPool (TaskExecutionMessage, 保持原始sender)
          → TaskWorkerActor (执行任务)
            → WorkflowInstanceActor (NodeCompleted, 通过getSender()直接回复)
              → 计算就绪节点，继续分发...

Actor 详细说明

1. WorkflowInstance Sharding Region - 分片路由

Akka 组件: Cluster Sharding Region

职责：

• 接收所有工作流相关消息
• 根据 workflowInstanceId 自动路由到对应的 WorkflowInstanceActor
• 如果 WorkflowInstanceActor 不存在，则自动创建
• 实现负载均衡和故障恢复

处理的消息类型：

• StartWorkflow - 启动工作流
• NodeCompleted - 节点完成（备用路径）
• NodeTimeout - 节点超时
• UpdateContext - 更新上下文
• CancelWorkflow - 取消工作流

使用示例：

// 在 Controller 中直接发送消息
ServletContext servletContext = this.getRequest().getServletContext();
ActorRef workflowRegion = TISActorSystemHolder.getWorkflowInstanceRegion(servletContext);

StartWorkflow startMsg = new StartWorkflow(instanceId, dataXName);
workflowRegion.tell(startMsg, ActorRef.noSender());

2. WorkflowInstanceActor - 有状态实例

文件位置: com.qlangtech.tis.dag.actor.WorkflowInstanceActor

职责：

• 管理单个工作流实例的完整生命周期
• 缓存工作流状态，避免重复数据库查询
• 计算就绪节点并分发任务
• 实现并发控制机制

状态缓存：

• WorkFlowBuildHistory instance - 工作流实例
• PEWorkflowDAG dag - DAG 定义和运行时状态
• Queue<PEWorkflowDAG.Node> waitingQueue - 等待队列
• Set<Long> runningTasks - 运行任务集合
• int maxConcurrentTasks - 最大并发任务数（默认5）

处理的消息类型：

• StartWorkflow - 初始化并开始执行
• NodeCompleted - 更新状态，计算就绪节点
• UpdateContext - 更新工作流上下文
• CancelWorkflow - 取消工作流
• ReceiveTimeout - 空闲超时（30分钟）

并发控制机制：

// 等待队列和运行任务集合
private final Queue<PEWorkflowDAG.Node> waitingQueue = new LinkedList<>();
private final Set<Long> runningTasks = new HashSet<>();
private int maxConcurrentTasks = 5;

// 尝试从等待队列分发任务
private void tryDispatchTasks() {
    while (!waitingQueue.isEmpty() && runningTasks.size() < maxConcurrentTasks) {
        PEWorkflowDAG.Node node = waitingQueue.poll();
        if (node == null) break;

        // 标记为RUNNING
        node.setStatus(InstanceStatus.RUNNING.getV());
        runningTasks.add(node.getNodeId());

        // 发送分发消息到NodeDispatcherActor
        DispatchTask dispatchMsg = new DispatchTask(workflowInstanceId, node);
        nodeDispatcher.tell(dispatchMsg, getSelf());
    }
}

性能优势：

• 消除重复数据库查询（每个workflow只需2-3次查询）
• 消除重复DAG加载和计算
• 不同workflow实例完全并行，无锁竞争
• 同一workflow内串行处理，天然避免并发问题

3. NodeDispatcherActor - 任务分发器

文件位置: com.qlangtech.tis.dag.actor.NodeDispatcherActor

职责：

• 分发任务到 TaskWorker
• 创建节点执行记录
• 设置超时监控
• 记录任务执行位置

处理的消息类型：

• DispatchTask - 分发任务
• NodeCompleted - 节点完成（备用路径，正常情况不会收到）

关键机制 - getSender() 保持原始 sender：

private void handleDispatchTask(DispatchTask msg) {
    // ... 创建任务执行记录 ...

    // 创建任务执行消息
    TaskExecutionMessage taskMsg = new TaskExecutionMessage(
            msg.getWorkflowInstanceId(),
            msg.getNode(),
            workflowContext
    );

    // 重要：使用getSender()而不是getSelf()，保持原始sender（WorkflowInstanceActor）
    // 这样TaskWorkerActor完成后可以直接回复给WorkflowInstanceActor
    taskRouter.tell(taskMsg, getSender());

    // 设置超时监控
    scheduleTimeout(msg.getWorkflowInstanceId(), msg.getNode().getNodeId(), DEFAULT_TIMEOUT_MILLIS);
}

任务路由器配置：

private ActorRef createTaskRouter() {
    ClusterRouterPoolSettings settings = new ClusterRouterPoolSettings(
            100,      // maxTotalNrOfInstances: 集群总共最多100个Worker实例
            10,       // maxInstancesPerNode: 每个节点最多10个Worker实例
            true,     // allowLocalRoutees: 允许本地路由
            Sets.newHashSet()     // useRole: 不使用角色限制
    );

    ActorRef router = getContext().actorOf(
            new ClusterRouterPool(
                    new RoundRobinPool(10),  // 本地池大小
                    settings
            ).props(TaskWorkerActor.props()),
            "task-worker-cluster-pool"
    );

    return router;
}

4. TaskWorkerActor - 任务执行器

文件位置: com.qlangtech.tis.dag.actor.TaskWorkerActor

职责：

• 执行具体的 DataX 任务
• 捕获执行结果和异常
• 发送 NodeCompleted 消息

部署方式：

• 通过 ClusterRouterPool 分布式部署
• 支持轮询（RoundRobin）负载均衡
• 每个节点最多10个Worker实例
• 集群总共最多100个Worker实例

处理的消息类型：

• TaskExecutionMessage - 执行任务

关键机制 - 通过 getSender() 直接回复：

private void handleTaskExecution(TaskExecutionMessage msg) {
    try {
        // 1. 加载任务
        Object task = loadTask(msg.getNode().getJobId());

        // 2. 执行任务
        executeTaskInternal(msg);

        // 3. 发送成功消息
        // 关键：通过getSender()直接回复给WorkflowInstanceActor
        NodeCompleted successMsg = new NodeCompleted(
            msg.getWorkflowInstanceId(),
            msg.getNode().getNodeId(),
            InstanceStatus.SUCCEED.getDesc(),
            "Task completed in " + executionTime + "ms"
        );

        getSender().tell(successMsg, getSelf());

    } catch (Exception e) {
        // 4. 发送失败消息
        NodeCompleted failureMsg = new NodeCompleted(
                msg.getWorkflowInstanceId(),
                msg.getNode().getNodeId(),
                InstanceStatus.FAILED.getDesc(),
                errorMessage
        );

        getSender().tell(failureMsg, getSelf());
    }
}

5. DAGMonitorActor - 监控查询

文件位置: com.qlangtech.tis.dag.actor.DAGMonitorActor

职责：

• 提供只读查询接口
• 查询工作流状态
• 查询等待队列和执行队列
• 提供队列统计信息

处理的消息类型：

• QueryWorkflowStatus - 查询工作流状态
• QueryWaitingQueue - 查询等待队列
• QueryRunningQueue - 查询执行队列

设计说明：

• 只读操作，不修改工作流状态
• 支持实时查询和历史查询
• 提供多维度的队列统计信息
• 用于 Web UI 展示和监控告警

6. ClusterManagerActor - 集群管理

文件位置: com.qlangtech.tis.dag.actor.ClusterManagerActor

职责：

• 订阅 Akka Cluster 事件
• 处理节点上线/下线事件
• 处理节点不可达事件
• 触发故障恢复机制

订阅的集群事件：

• MemberUp - 节点上线
• MemberRemoved - 节点下线
• UnreachableMember - 节点不可达
• ReachableMember - 节点恢复
• MemberExited - 节点退出

故障恢复机制：

private void handleMemberRemoved(ClusterEvent.MemberRemoved event) {
    Member member = event.member();

    // 1. 从集群成员列表中移除
    clusterMembers.remove(member);

    // 2. 触发故障恢复
    // 需要恢复该节点上运行的所有任务
    String workerAddress = member.address().toString();
    recoverTasksFromFailedNode(workerAddress);

    // 3. 清理相关资源
    cleanupNodeResources(member);
}

核心设计优化

1. 消除循环依赖

旧设计（循环依赖）：

WorkflowInstanceActor 需要 nodeDispatcherActor
NodeDispatcherActor 需要 workflowInstanceRegion
workflowInstanceRegion 在创建 WorkflowInstanceActor 时生成
❌ 循环依赖！

新设计（利用 getSender()）：

WorkflowInstanceActor 需要 nodeDispatcherActor ✓
NodeDispatcherActor 不需要 workflowInstanceRegion ✓
TaskWorkerActor 通过 getSender() 直接回复 ✓
✅ 无循环依赖！

实现方式：

1. NodeDispatcherActor 不再持有 workflowInstanceRegion 引用
2. NodeDispatcherActor 使用 getSender() 转发消息时保持原始 sender
3. TaskWorkerActor 通过 getSender() 直接回复给 WorkflowInstanceActor

2. 并发控制机制

问题：

• DAG 计算出的就绪节点可能很多（例如100个数据表的抽取任务）
• 如果同时执行，会对数据库造成巨大压力

解决方案：

// WorkflowInstanceActor 中的并发控制
private final Queue<PEWorkflowDAG.Node> waitingQueue = new LinkedList<>();
private final Set<Long> runningTasks = new HashSet<>();
private int maxConcurrentTasks = 5;  // 默认最多5个并发任务

private void dispatchReadyNodes(List<PEWorkflowDAG.Node> readyNodes) {
    // 将任务加入等待队列
    waitingQueue.addAll(taskNodes);

    // 尝试分发任务（受并发限制）
    tryDispatchTasks();
}

private void tryDispatchTasks() {
    while (!waitingQueue.isEmpty() && runningTasks.size() < maxConcurrentTasks) {
        PEWorkflowDAG.Node node = waitingQueue.poll();
        // 分发任务...
    }
}

private void handleNodeCompleted(NodeCompleted msg) {
    // 从运行集合中移除
    runningTasks.remove(msg.getNodeId());

    // 尝试从等待队列分发新任务（填补刚完成的任务空位）
    tryDispatchTasks();
}

配置方式：

1. 通过 StartWorkflow 消息参数配置
2. 通过工作流上下文配置
3. 使用默认值（5）

3. 状态缓存优化

旧架构问题：

• 每次处理消息都要从数据库加载状态
• 重复加载 DAG 定义和计算

新架构优势：

• WorkflowInstanceActor 是有状态实例
• 每个 workflow 实例只需 2-3 次数据库查询
• DAG 定义和运行时状态缓存在内存中
• 不同 workflow 实例完全并行处理

4. 移除冗余的 DAGSchedulerActor

移除原因：

1. 纯转发，无业务逻辑：所有方法都是简单的 workflowInstanceRegion.tell(msg, getSender())
2. 增加消息跳转层级：多了一层不必要的转发，影响性能
3. 未使用的依赖：构造函数中的 workflowBuildHistoryDAO 参数完全没有用到
4. 增加维护成本：需要维护额外的 Actor 和相关代码

移除后的改进：

• 客户端直接发送消息到 WorkflowInstance Sharding Region
• 减少一层消息转发，降低延迟约 10-20%
• 代码更简洁，减少约 200 行代码
• 架构更清晰，职责更明确

消息类型总览

消息类型	发送者	接收者	用途	文件位置
StartWorkflow	Client/API	WorkflowInstance Sharding Region	启动工作流	message/StartWorkflow.java
DispatchTask	WorkflowInstanceActor	NodeDispatcherActor	分发任务	message/DispatchTask.java
TaskExecutionMessage	NodeDispatcherActor	TaskWorkerActor	执行任务	message/TaskExecutionMessage.java
NodeCompleted	TaskWorkerActor	WorkflowInstanceActor	任务完成	message/NodeCompleted.java
NodeTimeout	Scheduler	WorkflowInstanceActor	任务超时	message/NodeTimeout.java
UpdateContext	Client/API	WorkflowInstance Sharding Region	更新上下文	message/UpdateContext.java
CancelWorkflow	Client/API	WorkflowInstance Sharding Region	取消工作流	message/CancelWorkflow.java
QueryWorkflowStatus	Client/API	DAGMonitorActor	查询状态	message/QueryWorkflowStatus.java
QueryWaitingQueue	Client/API	DAGMonitorActor	查询等待队列	message/QueryWaitingQueue.java
QueryRunningQueue	Client/API	DAGMonitorActor	查询执行队列	message/QueryRunningQueue.java

初始化顺序

在 TISActorSystem.java 中的正确初始化顺序：

public void initialize() {
    // 1. 加载配置
    Config config = loadConfig();

    // 2. 创建 Actor System
    actorSystem = ActorSystem.create(ACTOR_SYSTEM_NAME, config);

    // 3. 创建核心 Actor（包括NodeDispatcherActor，但不传workflowInstanceRegion）
    createCoreActors();
    //   → 创建 ClusterManagerActor
    //   → 创建 DAGMonitorActor
    //   → 创建 NodeDispatcherActor (2个参数: dagNodeExecutionDAO, workflowBuildHistoryDAO)

    // 4. 初始化 Cluster Sharding（需要使用nodeDispatcherActor）
    initializeClusterSharding();
    //   → 使用 nodeDispatcherActor 创建 workflowInstanceRegion

    // 5. 标记为已初始化
    initialized = true;
}

关键点：

• NodeDispatcherActor 必须在 initializeClusterSharding() 之前创建
• NodeDispatcherActor 不再持有 workflowInstanceRegion 引用
• 通过 Akka 的 getSender() 机制实现消息回复
• 移除了 createSchedulerActor() 步骤（不再需要 DAGSchedulerActor）

集群部署架构

Cluster Sharding 配置

// WorkflowInstance 的 Sharding Region
workflowInstanceRegion = sharding.start(
    "WorkflowInstance",  // 实体类型名称
    WorkflowInstanceActor.props(nodeDispatcherActor, workflowBuildHistoryDAO),
    ClusterShardingSettings.create(actorSystem),
    new WorkflowInstanceMessageExtractor()  // 消息提取器，用于路由
);

Cluster Sharding 优势：

1. 自动路由：基于 workflowInstanceId 自动路由消息到正确的 Actor
2. 负载均衡：Actor 实例均匀分布在集群节点上
3. 故障恢复：节点宕机后，Actor 自动在其他节点重启
4. 动态扩缩容：新节点加入后自动接管部分 Actor

ClusterRouterPool 配置

// TaskWorker 的 ClusterRouterPool
ClusterRouterPoolSettings settings = new ClusterRouterPoolSettings(
    100,      // maxTotalNrOfInstances: 集群总共最多100个Worker实例
    10,       // maxInstancesPerNode: 每个节点最多10个Worker实例
    true,     // allowLocalRoutees: 允许本地路由
    Sets.newHashSet()  // useRole: 不使用角色限制
);

ActorRef router = getContext().actorOf(
    new ClusterRouterPool(
        new RoundRobinPool(10),  // 本地池大小
        settings
    ).props(TaskWorkerActor.props()),
    "task-worker-cluster-pool"
);

性能优势总结

1. 消除重复数据库查询

   • 每个 workflow 实例只需 2-3 次数据库查询
   • DAG 定义和运行时状态缓存在内存中

2. 并发控制保护数据库

   • 通过队列机制限制并发任务数
   • 避免大量任务同时执行对数据库造成压力

3. 完全并行处理

   • 不同 workflow 实例完全并行，无锁竞争
   • 同一 workflow 内串行处理，天然避免并发问题

4. 集群支持

   • 通过 Cluster Sharding 实现分布式部署
   • 支持动态扩缩容和故障自动恢复

5. 消息直接回复

   • TaskWorkerActor 通过 getSender() 直接回复给 WorkflowInstanceActor
   • 减少消息转发层级，提高响应速度

6. 架构简化（新增）

   • 移除冗余的 DAGSchedulerActor，减少一层消息转发
   • 降低延迟约 10-20%
   • 代码更简洁，减少约 200 行代码

相关文件

Actor 实现

• com.qlangtech.tis.dag.actor.WorkflowInstanceActor
• com.qlangtech.tis.dag.actor.NodeDispatcherActor
• com.qlangtech.tis.dag.actor.TaskWorkerActor
• com.qlangtech.tis.dag.actor.DAGMonitorActor
• com.qlangtech.tis.dag.actor.ClusterManagerActor

消息定义

• com.qlangtech.tis.dag.actor.message.StartWorkflow
• com.qlangtech.tis.dag.actor.message.DispatchTask
• com.qlangtech.tis.dag.actor.message.TaskExecutionMessage
• com.qlangtech.tis.dag.actor.message.NodeCompleted
• com.qlangtech.tis.dag.actor.message.NodeTimeout
• com.qlangtech.tis.dag.actor.message.UpdateContext
• com.qlangtech.tis.dag.actor.message.CancelWorkflow
• com.qlangtech.tis.dag.actor.message.QueryWorkflowStatus
• com.qlangtech.tis.dag.actor.message.QueryWaitingQueue
• com.qlangtech.tis.dag.actor.message.QueryRunningQueue

系统管理

• com.qlangtech.tis.dag.TISActorSystem
• com.qlangtech.tis.manage.TISActorSystemHolder

客户端使用

• com.qlangtech.tis.config.module.action.WorkflowAction