关于Java实现多CSV文件逐个复制至tag.csv并触发Oracle存储过程的技术咨询

我先理清楚你的核心需求：逐个处理指定目录下的上传CSV文件，每个文件先覆盖写入到目标位置的tag.csv，然后单独触发一次Shell脚本执行Oracle存储过程，循环直到所有文件处理完毕。下面我从Java实现逻辑、关键代码片段、避坑注意事项这几个方面给你具体方案：

一、核心流程梳理 #

整个流程的关键是保证「单文件→覆盖tag.csv→触发脚本→归档文件」的原子性，避免重复处理或漏处理：

• 轮询指定上传目录，获取待处理的CSV文件列表
• 遍历每个文件：
  1. 将当前CSV内容完整覆盖到tag.csv（确保脚本每次只读取当前文件的数据）
  2. 调用Shell脚本，让其读取tag.csv并执行Oracle存储过程
  3. 将处理完成的文件移动到归档目录，防止重复处理
• 循环轮询，直到上传目录无待处理文件

二、Java代码实现示例 #

用Java NIO处理文件操作效率更高，同时加入错误处理和归档逻辑：

import java.io.IOException;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.ArrayList;
import java.util.List;

public class CsvBatchProcessor {
    // 可根据实际场景修改路径参数
    private static final Path UPLOAD_DIR = Paths.get("/opt/upload/csv");
    private static final Path TAG_CSV = Paths.get("/opt/script/tag.csv");
    private static final Path ARCHIVE_DIR = Paths.get("/opt/upload/archive");
    private static final String SHELL_SCRIPT = "/opt/script/run_procedure.sh";

    public static void main(String[] args) throws IOException, InterruptedException {
        // 初始化归档目录
        if (!Files.exists(ARCHIVE_DIR)) {
            Files.createDirectories(ARCHIVE_DIR);
        }

        // 持续轮询处理
        while (true) {
            List<Path> pendingFiles = getPendingCsvFiles();
            if (pendingFiles.isEmpty()) {
                System.out.println("暂无待处理文件，10秒后重试...");
                Thread.sleep(10000); // 轮询间隔可按需调整
                continue;
            }

            // 逐个处理文件
            for (Path file : pendingFiles) {
                try {
                    // 1. 覆盖写入tag.csv
                    copyToTagCsv(file);
                    System.out.println("[成功] 已将文件 " + file.getFileName() + " 同步到tag.csv");

                    // 2. 触发Shell脚本
                    executeShellScript();
                    System.out.println("[成功] 已触发脚本处理文件 " + file.getFileName());

                    // 3. 归档文件（加时间戳避免重名）
                    archiveProcessedFile(file);
                    System.out.println("[成功] 文件 " + file.getFileName() + " 已归档");
                } catch (Exception e) {
                    System.err.println("[失败] 处理文件 " + file.getFileName() + " 出错：" + e.getMessage());
                    // 可选：将失败文件移到错误目录，方便后续排查
                    // moveToErrorDir(file);
                }
            }
        }
    }

    // 获取上传目录下的所有CSV文件（不遍历子目录）
    private static List<Path> getPendingCsvFiles() throws IOException {
        List<Path> csvFiles = new ArrayList<>();
        Files.walkFileTree(UPLOAD_DIR, new SimpleFileVisitor<>() {
            @Override
            public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
                if (file.toString().toLowerCase().endsWith(".csv")) {
                    csvFiles.add(file);
                }
                return FileVisitResult.CONTINUE;
            }

            @Override
            public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) {
                return dir.equals(UPLOAD_DIR) ? FileVisitResult.CONTINUE : FileVisitResult.SKIP_SUBTREE;
            }
        });
        return csvFiles;
    }

    // 覆盖复制文件到tag.csv
    private static void copyToTagCsv(Path source) throws IOException {
        Files.copy(source, TAG_CSV, StandardCopyOption.REPLACE_EXISTING);
    }

    // 执行Shell脚本并检查退出码
    private static void executeShellScript() throws IOException, InterruptedException {
        ProcessBuilder pb = new ProcessBuilder("/bin/bash", SHELL_SCRIPT);
        // 可选：设置脚本工作目录或传递参数
        // pb.directory(new File("/opt/script"));
        Process process = pb.start();

        int exitCode = process.waitFor();
        if (exitCode != 0) {
            throw new RuntimeException("脚本执行失败，退出码：" + exitCode);
        }
    }

    // 将处理完成的文件归档（添加时间戳避免重名）
    private static void archiveProcessedFile(Path source) throws IOException {
        String fileName = source.getFileName().toString();
        String timestamp = String.valueOf(System.currentTimeMillis());
        String archivedName = fileName.replace(".csv", "_" + timestamp + ".csv");
        Path target = ARCHIVE_DIR.resolve(archivedName);

        Files.move(source, target, StandardCopyOption.REPLACE_EXISTING);
    }
}

三、配套Shell脚本示例（run_procedure.sh） #

脚本需要读取tag.csv并调用Oracle存储过程，这里给出基础模板（需提前配置Oracle客户端环境）：

#!/bin/bash

# Oracle连接配置
ORACLE_USER="your_db_user"
ORACLE_PWD="your_db_pwd"
ORACLE_TNS="your_tns_name"
TAG_CSV_PATH="/opt/script/tag.csv"

# 调用存储过程（假设存储过程会读取tag.csv数据，可根据实际调整逻辑）
sqlplus -s ${ORACLE_USER}/${ORACLE_PWD}@${ORACLE_TNS} << EOF
SET SERVEROUTPUT ON;
BEGIN
    -- 替换为你的存储过程名称及参数
    PROCESS_TAG_DATA('${TAG_CSV_PATH}');
    COMMIT;
END;
/
EXIT;
EOF

# 检查脚本执行结果
if [ $? -eq 0 ]; then
    echo "存储过程执行成功"
else
    echo "存储过程执行失败"
    exit 1
fi

四、关键避坑点 #

• 文件完整性校验：如果文件是实时上传的，直接处理可能读到不完整数据。可以通过「等待文件最后修改时间超过5秒」或「上传程序先写临时文件，完成后重命名为.csv」的方式，只处理已上传完成的文件。
• 权限问题：确保Java程序拥有上传目录、tag.csv路径、归档目录的读写权限，以及执行Shell脚本的权限。
• 异常重试：对于临时网络故障或Oracle连接失败的情况，可以给关键步骤（比如脚本执行）添加重试逻辑，避免单次失败导致文件处理中断。
• 性能优化：如果文件数量极大，可考虑用线程池并行处理，但要注意tag.csv的写入互斥（需加锁避免多个线程同时写入）。

内容的提问来源于stack exchange，提问作者shamini sathyamoorthy