modfiy for powerjob batch sync

Author: baisui1981

tis-datax/executor/powerjob-worker-samples/docker-build.sh

@@ -1,4 +1,4 @@
-tis_version="4.2.0"
+tis_version="4.3.0-SNAPSHOT"
 docker rmi registry.cn-hangzhou.aliyuncs.com/tis/tis-datax-executor:$tis_version
 docker rmi tis/tis-datax-executor:$tis_version
 docker build . -t tis/tis-datax-executor:$tis_version

tis-datax/executor/powerjob-worker-samples/learning/PowerJob中与DAG调度触发相关的核心代码.md

@@ -0,0 +1,170 @@
+在 PowerJob 中，与 DAG 调度触发相关的核心代码主要集中在 **服务端（powerjob-server）** 模块，以下是关键的核心类及其职责：
+
+---
+
+### 1. **工作流调度引擎入口**
+**`WorkflowInstanceManager`**
+- 路径：`tech.powerjob.server.core.workflow.algorithm.WorkflowInstanceManager`
+- 核心职责：
+    - 工作流实例的生命周期管理（创建、启动、停止）
+    - 触发工作流实例的初始调度
+- 关键方法：
+  ```java
+  // 创建并启动工作流实例
+  public void createWorkflowInstance(WorkflowInstanceCreateReq req) {
+      // 1. 持久化工作流实例
+      // 2. 提交到调度队列 (workflowReadyQueue)
+  }
+  ```
+
+---
+
+### 2. **DAG 调度核心处理器**
+**`WorkflowExecutor`**
+- 路径：`tech.powerjob.server.core.workflow.algorithm.WorkflowExecutor`
+- 核心职责：
+    - **DAG 状态机驱动**：推进工作流实例的状态流转
+    - **任务节点调度**：根据 DAG 依赖关系触发任务执行
+    - **容错处理**：失败重试、超时处理等
+- 关键逻辑：
+  ```java
+  public void executeWorkflowInstance(Long wfInstanceId) {
+      // 1. 加载DAG拓扑结构
+      WorkflowDAG dag = workflowService.fetchWorkflowDAG(wfInstanceId);
+      
+      // 2. 计算可运行节点
+      List<Node> runnableNodes = DAGUtils.calculateRunnableNodes(dag);
+      
+      // 3. 提交任务到任务调度器
+      for (Node node : runnableNodes) {
+          taskDispatcher.dispatch(node.toTaskInstance(), node.getNodeParams());
+      }
+  }
+  ```
+
+---
+
+### 3. **DAG 图遍历与状态计算**
+**`DAGTraverser`**
+- 路径：`tech.powerjob.server.core.workflow.algorithm.DAGTraverser`
+- 核心职责：
+    - **依赖解析**：基于 DAG 图计算可执行节点
+    - **状态传播**：处理节点完成/失败后的状态传递
+- 关键算法：
+  ```java
+  public static List<Node> calculateRunnableNodes(WorkflowDAG dag) {
+      // 实现拓扑排序算法
+      return topologicalSort(dag)
+          .stream()
+          .filter(node -> isRunnable(node, dag)) // 检查前置依赖是否完成
+          .collect(Collectors.toList());
+  }
+  ```
+
+---
+
+### 4. **任务触发执行器**
+**`TaskDispatcher`**
+- 路径：`tech.powerjob.server.core.scheduler.TaskDispatcher`
+- 核心职责：
+    - 将 DAG 节点转化为具体任务实例
+    - 提交任务到工作线程池执行
+- 关键代码：
+  ```java
+  public void dispatch(TaskInstance task, Map<String, String> params) {
+      // 1. 创建任务实例记录
+      taskInstanceService.save(task);
+      
+      // 2. 提交到执行器
+      workerClusterManager.dispatchTask(
+          task.getInstanceId(), 
+          task.getTaskId(), 
+          params
+      );
+  }
+  ```
+
+---
+
+### 5. **回调处理器（状态推进引擎）**
+**`WorkflowInstanceInfoService`**
+- 路径：`tech.powerjob.server.core.service.WorkflowInstanceInfoService`
+- 核心职责：
+    - 监听**任务完成事件**
+    - 触发工作流状态更新和后续节点调度
+- 关键逻辑：
+  ```java
+  @EventListener
+  public void onTaskInstanceStatusChange(TaskInstanceStatusChangeEvent event) {
+      if (event.getStatus() == TaskResult.COMPLETED || event.getStatus() == TaskResult.FAILED) {
+          // 更新DAG节点状态
+          workflowService.updateNodeStatus(event.getTaskId(), event.getStatus());
+          
+          // 重新调度工作流实例
+          workflowInstanceManager.resumeWorkflowInstance(
+              event.getWfInstanceId()
+          );
+      }
+  }
+  ```
+
+---
+
+### 6. **调度器入口（定时触发）**
+**`WorkflowScheduleService`**
+- 路径：`tech.powerjob.server.core.scheduler.WorkflowScheduleService`
+- 核心职责：
+    - 扫描需要调度的**定时工作流**
+    - 触发工作流实例创建
+- 调度逻辑：
+  ```java
+  @Scheduled(fixedRate = 10_000)
+  public void scheduleTimedWorkflow() {
+      List<WorkflowInfo> timedWorkflows = workflowService.fetchTimedWorkflow();
+      for (WorkflowInfo wf : timedWorkflows) {
+          if (shouldSchedule(wf)) {
+              workflowInstanceManager.createWorkflowInstance(
+                  new WorkflowInstanceCreateReq(wf.getId())
+              );
+          }
+      }
+  }
+  ```
+
+---
+
+### 核心执行流程
+```mermaid
+sequenceDiagram
+    participant ScheduleService
+    participant InstanceManager
+    participant WorkflowExecutor
+    participant DAGTraverser
+    participant TaskDispatcher
+    
+    ScheduleService->>InstanceManager: 定时触发创建实例
+    InstanceManager->>WorkflowExecutor: 提交实例
+    WorkflowExecutor->>DAGTraverser: 请求可运行节点
+    DAGTraverser-->>WorkflowExecutor: 返回节点列表
+    WorkflowExecutor->>TaskDispatcher: 分发任务
+    TaskDispatcher->>Worker: 执行任务
+    Worker-->>InstanceManager: 任务完成事件
+    InstanceManager->>WorkflowExecutor: 触发状态更新
+    loop 直到DAG完成
+        WorkflowExecutor->>DAGTraverser: 重新计算节点
+        DAGTraverser-->>WorkflowExecutor: 新可运行节点
+        WorkflowExecutor->>TaskDispatcher: 分发新任务
+    end
+```
+
+---
+
+### 关键设计特点
+1. **事件驱动架构**：通过任务完成事件驱动 DAG 状态流转
+2. **拓扑排序算法**：使用 Kahn 算法实现高效依赖解析
+3. **异步执行模型**：所有调度操作异步化，避免阻塞主线程
+4. **状态机管理**：明确定义工作流实例的 6 种状态（WAITING/RUNNING/SUCCEED/FAILED/STOPPED）
+
+建议重点关注以下包路径的源码：  
+`powerjob-server/src/main/java/tech/powerjob/server/core/workflow/algorithm/`  
+`powerjob-server/src/main/java/tech/powerjob/server/core/scheduler/`

tis-datax/executor/powerjob-worker-samples/learning/能帮我列出powerjob内部与akka相关的核心代码.md

@@ -0,0 +1,177 @@
+好的，我为您梳理了 PowerJob 中与 Akka 相关的核心代码实现部分。Akka 主要用于 PowerJob 的**分布式通信层**，特别是在服务端(Server)和客户端(Worker)之间的高性能网络通信中。以下是关键代码模块分析：
+
+---
+
+### 核心模块路径
+```bash
+powerjob-remote/akka-remote/src/main/java/tech/powerjob/remote/framework
+```
+
+---
+
+### 核心实现类
+
+#### 1. Akka 初始化与配置
+**`AkkaRemoteEngine.java`**
+- 位置：`tech.powerjob.remote.framework.engine.impl.AkkaRemoteEngine`
+- 核心职责：
+    - 初始化 Akka ActorSystem
+    - 加载 Akka 配置（`application.conf`）
+    - 创建根 Actor `ActorSystemHolder`
+- 关键代码：
+  ```java
+  ActorSystem actorSystem = ActorSystem.create("powerjob-akka-system", akkaConfig);
+  actorSystem.actorOf(Props.create(ActorSystemHolder.class, actorHandler), "holder");
+  ```
+
+#### 2. Actor 系统容器
+**`ActorSystemHolder.java`**
+- 位置：`tech.powerjob.remote.framework.actor.ActorSystemHolder`
+- 核心职责：
+    - 管理所有业务 Actor 的生命周期
+    - 作为所有 Actor 的根监督者
+    - 处理 Actor 的创建/销毁
+- 关键方法：
+  ```java
+  private void initHandlers() {
+      // 动态创建处理不同请求类型的Actor
+      handlers.forEach(handlerInfo -> {
+          String actorName = handlerInfo.getPath();
+          getContext().actorOf(Props.create(HandlerActor.class, handlerInfo), actorName);
+      });
+  }
+  ```
+
+#### 3. 请求处理 Actor
+**`HandlerActor.java`**
+- 位置：`tech.powerjob.remote.framework.actor.HandlerActor`
+- 核心职责：
+    - 实际处理网络请求的 Akka Actor
+    - 路由请求到对应的业务处理器
+    - 实现异步非阻塞处理
+- 核心逻辑：
+  ```java
+  @Override
+  public Receive createReceive() {
+      return receiveBuilder()
+          .match(RemoteReq.class, this::onReceiveRemoteReq)
+          .build();
+  }
+  
+  private void onReceiveRemoteReq(RemoteReq remoteReq) {
+      // 1. 反序列化请求
+      // 2. 路由到对应业务处理器
+      // 3. 异步处理并返回响应
+  }
+  ```
+
+#### 4. 远程消息协议
+**`RemoteProtocol.java`**
+- 位置：`tech.powerjob.remote.framework.actor.HandlerActor`
+- 核心类：
+    - `RemoteReq`：封装请求数据
+    - `RemoteResp`：封装响应数据
+- 数据结构：
+  ```java
+  public class RemoteReq implements Serializable {
+      private String path;       // 请求路径
+      private byte[] payload;   // 二进制负载
+      private Map<String, String> metadata; // 元数据
+  }
+  ```
+
+#### 5. 序列化扩展
+**`PowerAkkaSerializer.java`**
+- 位置：`tech.powerjob.remote.framework.serializer.PowerAkkaSerializer`
+- 核心职责：
+    - 实现 Akka 的 `Serializer` 接口
+    - 使用 Hessian2 进行高效二进制序列化
+- 关键实现：
+  ```java
+  @Override
+  public byte[] toBinary(Object object) {
+      return SerializerUtils.hes.serialize(object);
+  }
+  
+  @Override
+  public Object fromBinary(byte[] bytes, Class<?> manifest) {
+      return SerializerUtils.hes.deserialize(bytes);
+  }
+  ```
+
+---
+
+### 配置资源文件
+**`application.conf`**
+- 路径：`src/main/resources/application.conf`
+- 核心配置：
+  ```conf
+  akka {
+    actor {
+      provider = cluster
+      serializers {
+        powerjob = "tech.powerjob.remote.framework.serializer.PowerAkkaSerializer"
+      }
+      serialization-bindings {
+        "java.io.Serializable" = powerjob
+      }
+    }
+    remote.artery {
+      transport = tcp
+      canonical.hostname = "127.0.0.1"
+      canonical.port = 10000
+    }
+  }
+  ```
+
+---
+
+### 执行流程
+1. **初始化阶段**  
+   `AkkaRemoteEngine` → 创建 `ActorSystem` → 初始化 `ActorSystemHolder`
+
+2. **请求接收阶段**  
+   网络请求 → 序列化为 `RemoteReq` → 路由到 `HandlerActor`
+
+3. **业务处理阶段**  
+   `HandlerActor` → 解析请求 → 调用业务处理器 → 生成 `RemoteResp`
+
+4. **响应返回阶段**  
+   序列化响应 → 通过 Akka 网络层返回 → 客户端反序列化
+
+---
+
+### 设计特点
+1. **高性能传输**  
+   使用 Artery TCP 传输层，支持百万级 QPS
+2. **零拷贝优化**  
+   通过 ByteBuffer 实现二进制零拷贝传输
+3. **弹性扩展**  
+   基于 Actor 模型的横向扩展能力
+4. **协议无关性**  
+   业务处理器无需感知 Akka 实现细节
+
+---
+
+### 示例调用栈
+```mermaid
+sequenceDiagram
+    participant Client
+    participant ServerActor
+    participant HandlerActor
+    participant BusinessHandler
+    
+    Client->>ServerActor: 发送RemoteReq
+    ServerActor->>HandlerActor: 转发请求
+    HandlerActor->>BusinessHandler: 调用process()
+    BusinessHandler-->>HandlerActor: 返回结果
+    HandlerActor->>ServerActor: 封装RemoteResp
+    ServerActor->>Client: 返回响应
+```
+
+建议结合源码查看以下关键包路径：  
+`tech.powerjob.remote.framework.actor`  
+`tech.powerjob.remote.framework.engine.impl`  
+`tech.powerjob.remote.framework.serializer`
+
+这些模块共同构成了 PowerJob 基于 Akka 的高性能分布式通信基础架构。