flowable各节点初始化本地文件夹

2025-12-05 14:14:22 +08:00
parent fd43685c8a
commit ec22441662
12 changed files with 225 additions and 42 deletions
--- a/flowable/src/main/java/com/sdm/flowable/constants/FlowableConfig.java
+++ b/flowable/src/main/java/com/sdm/flowable/constants/FlowableConfig.java
@@ -29,4 +29,9 @@ public interface FlowableConfig {
     */
    String ASYNC_TASK_SUFFIX = "_wait";
    String WAIT_USER_SUFFIX = "_waitUser";
+
+    /**
+     * 流程的节点本地文件夹基础路径
+     */
+    String FLOWABLE_SIMULATION_BASEDIR ="/home/simulation/";
 }
--- a/flowable/src/main/java/com/sdm/flowable/delegate/UniversalDelegate.java
+++ b/flowable/src/main/java/com/sdm/flowable/delegate/UniversalDelegate.java
@@ -1,7 +1,12 @@
 package com.sdm.flowable.delegate;

+import com.alibaba.fastjson2.JSONObject;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sdm.common.common.SdmResponse;
 import com.sdm.common.entity.flowable.executeConfig.BaseExecuteConfig;
+import com.sdm.common.entity.req.data.GetFileBaseInfoReq;
+import com.sdm.common.entity.resp.data.FileMetadataInfoResp;
+import com.sdm.common.feign.inter.data.IDataFeignClient;
 import com.sdm.flowable.constants.FlowableConfig;
 import com.sdm.flowable.delegate.handler.ExecutionHandler;
 import com.sdm.flowable.dto.req.AsyncCallbackRequest;
@@ -15,6 +20,9 @@ import org.flowable.engine.delegate.JavaDelegate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;

+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.Collections;
 import java.util.Map;

@@ -39,6 +47,9 @@ public class UniversalDelegate implements JavaDelegate {
    @Autowired
    private RuntimeService runtimeService;

+    @Autowired
+    private IDataFeignClient dataFeignClient;
+
    @Override
    public void execute(DelegateExecution execution) {
        try {
@@ -55,6 +66,17 @@ public class UniversalDelegate implements JavaDelegate {

            log.info("节点执行参数, 流程实例ID: {}, 节点ID: {}, 参数: {}", procInstId, nodeId, params);

+            // 3、创建本地文件夹，用于后续节点计算直接从本地读取，不需要再从minio中获取数据
+            Long currentNodeOutputDirId = (Long)params.getOrDefault("outputDirId", null);
+            GetFileBaseInfoReq getFileBaseInfoReq = new GetFileBaseInfoReq();
+            getFileBaseInfoReq.setFileId(currentNodeOutputDirId);
+            SdmResponse<FileMetadataInfoResp> fileBaseInfoResp = dataFeignClient.getFileBaseInfo(getFileBaseInfoReq);
+            if(!fileBaseInfoResp.isSuccess()||fileBaseInfoResp.getData()==null){
+                throw new RuntimeException("当前节点未查询到输入文件夹");
+            }
+            String objectKey = fileBaseInfoResp.getData().getObjectKey();
+            createLocalDir(objectKey);
+
            // 检查是否有扩展元素配置
            if (execution.getCurrentFlowElement().getExtensionElements() != null &&
                    execution.getCurrentFlowElement().getExtensionElements().get(FlowableConfig.EXECUTECONFIG) != null) {
@@ -92,6 +114,29 @@ public class UniversalDelegate implements JavaDelegate {
        }
    }

+    /**
+     * 基于objectkey创建本地文件夹
+     * @param objectKey minio的文件对象路径
+     */
+    private void createLocalDir(String objectKey) {
+        String simulationBaseDir = FlowableConfig.FLOWABLE_SIMULATION_BASEDIR;
+        //  构建本地基础路径（直接使用 simulationBaseDir + objectKey）
+        Path localBaseDir = Paths.get(simulationBaseDir).toAbsolutePath().normalize();
+        Path fullLocalBase = localBaseDir.resolve(objectKey).normalize();
+
+        // 安全校验：确保 fullLocalBase 确实在 basePath 下
+        if (!fullLocalBase.startsWith(localBaseDir)) {
+            throw new RuntimeException("非法文件夹路径，可能包含路径穿越: " + objectKey);
+        }
+
+        try {
+            Files.createDirectories(fullLocalBase);
+        } catch (Exception e) {
+            throw new RuntimeException("无法创建本地目录: " + fullLocalBase, e);
+        }
+    }
+
+
    /**
     * 处理任务执行失败的情况
     * @param execution 当前执行上下文
--- a/flowable/src/main/java/com/sdm/flowable/delegate/handler/HpcHandler.java
+++ b/flowable/src/main/java/com/sdm/flowable/delegate/handler/HpcHandler.java
@@ -5,10 +5,12 @@ import com.sdm.common.common.SdmResponse;
 import com.sdm.common.entity.flowable.executeConfig.HPCExecuteConfig;
 import com.sdm.common.entity.req.data.GetFileBaseInfoReq;
 import com.sdm.common.entity.req.pbs.SubmitHpcTaskRemoteReq;
+import com.sdm.common.entity.resp.data.FileMetadataInfoResp;
 import com.sdm.common.feign.inter.data.IDataFeignClient;
 import com.sdm.common.feign.inter.pbs.ITaskFeignClient;
 import com.sdm.common.log.CoreLogger;
 import com.sdm.common.utils.FilesUtil;
+import com.sdm.flowable.constants.FlowableConfig;
 import com.sdm.flowable.entity.ProcessNodeParam;
 import com.sdm.flowable.service.IAsyncTaskRecordService;
 import com.sdm.flowable.service.IProcessNodeParamService;
@@ -42,10 +44,6 @@ public class HpcHandler implements ExecutionHandler<Map<String, Object>,HPCExecu
    @Autowired
    private IDataFeignClient dataFeignClient;

-//    @Value("${flowable.simulationBaseDir:F:\\flowable\\}")
-    @Value("${flowable.simulationBaseDir:}")
-    private String simulationBaseDir;
-
    /*
    * params:业务参数
    * config：框架属性
@@ -94,6 +92,7 @@ public class HpcHandler implements ExecutionHandler<Map<String, Object>,HPCExecu

    private void dealHpcFile(SubmitHpcTaskRemoteReq submitHpcTaskRemoteReq, String beforeNodeId, String masterFileRegularStr,
                             String inputFilesRegularStr,String processDefinitionId,String processInstanceId) {
+        String simulationBaseDir = FlowableConfig.FLOWABLE_SIMULATION_BASEDIR;
        // 查询前节点的工作目录---》本地磁盘对应目录
        ProcessNodeParam processNodeParam = processNodeParamService.lambdaQuery().
                eq(ProcessNodeParam::getNodeId, beforeNodeId).
@@ -103,18 +102,16 @@ public class HpcHandler implements ExecutionHandler<Map<String, Object>,HPCExecu
        String paramJson = processNodeParam.getParamJson();
        JSONObject paramJsonObject = JSONObject.parseObject(paramJson);
        // outputDirId
-        Integer outputDirId = paramJsonObject.getInteger("outputDirId");
+        Long outputDirId = paramJsonObject.getLong("outputDirId");
        // 查data表
        GetFileBaseInfoReq getFileBaseInfoReq = new GetFileBaseInfoReq();
        getFileBaseInfoReq.setFileId(outputDirId);
-        SdmResponse fileBaseInfoResp = dataFeignClient.getFileBaseInfo(getFileBaseInfoReq);
+        SdmResponse<FileMetadataInfoResp> fileBaseInfoResp = dataFeignClient.getFileBaseInfo(getFileBaseInfoReq);
        if(!fileBaseInfoResp.isSuccess()||fileBaseInfoResp.getData()==null){
            throw new RuntimeException("上一节点信息查询失败");
        }
-        Object data = fileBaseInfoResp.getData();
-        // 直接强转为 Map（推荐指定泛型，避免后续取值强转）
-        Map<String, Object> dataMap = (Map<String, Object>) data;
-        String objectKey = dataMap.get("objectKey") == null ? "": dataMap.get("objectKey").toString();
+        FileMetadataInfoResp fileMetadataInfoResp = fileBaseInfoResp.getData();
+        String objectKey = fileMetadataInfoResp.getObjectKey();
        // 本地文件路径 taskLocalBaseDir
        String localSaveDir = simulationBaseDir + objectKey;
        CoreLogger.info("beforeNode localSaveDir:{}",localSaveDir);
--- a/flowable/src/main/resources/application-dev.yml
+++ b/flowable/src/main/resources/application-dev.yml
@@ -33,7 +33,4 @@ mybatis-plus:
  global-config:
    db-config:
      id-type: auto
-# 仿真任务本地nas盘基础目录
-flowable:
-  simulationBaseDir: /home/simulation/