1

src/main/java/com/fdkankan/rmnas/config/ThreadPoolConfig.java

@@ -0,0 +1,45 @@
+package com.fdkankan.rmnas.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.concurrent.*;
+
+/**
+ * 线程池配置 - 用于 NAS/OSS 批量删除等 IO 密集型任务
+ */
+@Configuration
+public class ThreadPoolConfig {
+
+    /**
+     * IO 密集型任务线程池
+     * 线程数 = 2 * CPU核数，适合网络上传、文件删除等阻塞操作
+     */
+    @Bean("ioTaskExecutor")
+    public ExecutorService ioTaskExecutor() {
+        int cores = Runtime.getRuntime().availableProcessors();
+        int poolSize = Math.max(cores * 2, 10); // 至少 10 个线程
+
+        ThreadPoolExecutor executor = new ThreadPoolExecutor(
+                poolSize,
+                poolSize,
+                60L, TimeUnit.SECONDS,
+                new LinkedBlockingQueue<>(1000),
+                new ThreadFactory() {
+                    private int counter = 0;
+                    @Override
+                    public Thread newThread(Runnable r) {
+                        Thread t = new Thread(r, "io-task-" + counter++);
+                        t.setDaemon(true);
+                        return t;
+                    }
+                },
+                new ThreadPoolExecutor.CallerRunsPolicy() // 队列满时由调用线程执行，防止任务丢失
+        );
+
+        // 允许核心线程超时，空闲时释放资源
+        executor.allowCoreThreadTimeOut(true);
+
+        return executor;
+    }
+}

src/main/java/com/fdkankan/rmnas/task/TaskService.java

@@ -22,6 +22,8 @@ import org.springframework.stereotype.Service;
 import java.io.File;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicInteger;
 
 @Service
@@ -44,6 +46,8 @@ public class TaskService {
 
     @Autowired
     ITbSceneProService tbSceneProService;
+    @Autowired
+    private ExecutorService ioTaskExecutor;
 
     public void rmNas() throws InterruptedException {
         RequestBase param = new RequestBase(1,1000, DateUtils.getMonthAgo() +" 00:00:00");
@@ -113,30 +117,37 @@ public class TaskService {
     public void rmOssPath(List<CommonScene> tbSceneList) {
         List<TbOssCloud> newList = new ArrayList<>();
         AtomicInteger i = new AtomicInteger();
-        tbSceneList.parallelStream().forEach(tbScene ->{
-            i.getAndIncrement();
-            String ossPath = DataSourcePath.getOssPath(tbScene.getDataSource());
-            if(StringUtils.isNotBlank(ossPath) && fYunFileServiceInterface.fileExist(ossPath+"/data.fdage")){
-                log.info("将oss原始资源复制到冷存储{},{},{},{},{}",bucket,ossPath,coldBucket,bucket+"/"+ossPath,i);
-                fYunFileServiceInterface.copyFileBetweenBucket(bucket,ossPath,coldBucket,bucket+"/"+ossPath);
-                try {
-                    log.info("将oss原始资源删除：{},{},{}",bucket,ossPath,i);
-                    fYunFileServiceInterface.deleteFolder(bucket,ossPath);
-                }catch (Exception e){
-                    log.info("oss删除文件出错：{}，{}",bucket,ossPath);
-                    log.info("oss删除文件出错",e);
+
+        List<CompletableFuture<Void>> futures = new ArrayList<>();
+        for (CommonScene tbScene : tbSceneList) {
+            futures.add(CompletableFuture.runAsync(() -> {
+                int idx = i.getAndIncrement();
+                String ossPath = DataSourcePath.getOssPath(tbScene.getDataSource());
+                if (StringUtils.isNotBlank(ossPath) && fYunFileServiceInterface.fileExist(ossPath + "/data.fdage")) {
+                    log.info("将oss原始资源复制到冷存储{},{},{},{},{}", bucket, ossPath, coldBucket, bucket + "/" + ossPath, idx);
+                    fYunFileServiceInterface.copyFileBetweenBucket(bucket, ossPath, coldBucket, bucket + "/" + ossPath);
+                    try {
+                        log.info("将oss原始资源删除：{},{},{}", bucket, ossPath, idx);
+                        fYunFileServiceInterface.deleteFolder(bucket, ossPath);
+                    } catch (Exception e) {
+                        log.info("oss删除文件出错：{}，{}", bucket, ossPath);
+                        log.info("oss删除文件出错", e);
+                    }
                 }
-            }
-            TbOssCloud tbOssCloud = new TbOssCloud();
-            tbOssCloud.setNum(tbScene.getNum());
-            tbOssCloud.setBucket(bucket);
-            tbOssCloud.setCloudBucket(coldBucket);
-            tbOssCloud.setOssPath(ossPath);
-            tbOssCloud.setCloudOssPath(StringUtils.isBlank(ossPath)?null:bucket+"/"+ossPath);
-            tbOssCloudService.save(tbOssCloud);
-            newList.add(tbOssCloud);
-        });
-        log.info("本次执行oss资源冷存储数量：{}",newList.size());
+                TbOssCloud tbOssCloud = new TbOssCloud();
+                tbOssCloud.setNum(tbScene.getNum());
+                tbOssCloud.setBucket(bucket);
+                tbOssCloud.setCloudBucket(coldBucket);
+                tbOssCloud.setOssPath(ossPath);
+                tbOssCloud.setCloudOssPath(StringUtils.isBlank(ossPath) ? null : bucket + "/" + ossPath);
+                tbOssCloudService.save(tbOssCloud);
+                synchronized (newList) {
+                    newList.add(tbOssCloud);
+                }
+            }, ioTaskExecutor));
+        }
+        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+        log.info("本次执行oss资源冷存储数量：{}", newList.size());
 
     }
 
@@ -159,24 +170,25 @@ public class TaskService {
         if(sceneProList != null && !sceneProList.isEmpty()){
             log.info("执行删除v3nas计算结果数据：{}",sceneProList.size());
 
-            sceneProList.parallelStream().forEach(tbScenePro ->{
-                if(StringUtils.isBlank(tbScenePro.getDataSource())
-                        || !tbScenePro.getDataSource().contains("/")
-                        || tbScenePro.getDataSource().split("/").length <4){
-                    log.info("跳过删除：{}",tbScenePro.getDataSource());
-                    return;
-                }
-                log.info("上传目录：{},{}",tbScenePro.getDataSource(),"build_result/"+tbScenePro.getNum()+"/");
-                fYunFileServiceInterface.uploadFolder(tbScenePro.getDataSource(),"build_result/"+tbScenePro.getNum()+"/");
-
-                FileUtil.del(tbScenePro.getDataSource());
-                FileUtil.del(tbScenePro.getDataSource()+"_images");
-                FileUtil.del(tbScenePro.getDataSource()+"_obj2txt");
-                tbSceneProService.updateDelBuildResult(tbScenePro.getId(),2);
-
-
-            });
-
+            List<CompletableFuture<Void>> futures = new ArrayList<>();
+            for (TbScenePro tbScenePro : sceneProList) {
+                futures.add(CompletableFuture.runAsync(() -> {
+                    if (StringUtils.isBlank(tbScenePro.getDataSource())
+                            || !tbScenePro.getDataSource().contains("/")
+                            || tbScenePro.getDataSource().split("/").length < 4) {
+                        log.info("跳过删除：{}", tbScenePro.getDataSource());
+                        return;
+                    }
+                    log.info("上传目录：{},{}", tbScenePro.getDataSource(), "build_result/" + tbScenePro.getNum() + "/");
+                    fYunFileServiceInterface.uploadFolder(tbScenePro.getDataSource(), "build_result/" + tbScenePro.getNum() + "/");
+
+                    FileUtil.del(tbScenePro.getDataSource());
+                    FileUtil.del(tbScenePro.getDataSource() + "_images");
+                    FileUtil.del(tbScenePro.getDataSource() + "_obj2txt");
+                    tbSceneProService.updateDelBuildResult(tbScenePro.getId(), 2);
+                }, ioTaskExecutor));
+            }
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
         }
     }