diff --git a/backend/src/main/java/com/senmesh/component/file/config/FileProperties.java b/backend/src/main/java/com/senmesh/component/file/config/FileProperties.java index f5f2286..af250bd 100644 --- a/backend/src/main/java/com/senmesh/component/file/config/FileProperties.java +++ b/backend/src/main/java/com/senmesh/component/file/config/FileProperties.java @@ -32,6 +32,25 @@ public class FileProperties { /** 解包风险校验开关 **/ private Boolean unpackRiskCheck = false; + + /** 目录监听与解包调度参数 **/ + private final Monitor monitor = new Monitor(); + } + + @Data + @NoArgsConstructor + public static class Monitor { + /** 单轮处理时间预算（秒）。超预算即结束本轮，剩余留给下一轮。按个数限流不合理：几十 KB 与几百 MB 的包工作量差三个数量级 **/ + private Integer roundBudgetSeconds = 4; + + /** 单轮兜底个数上限，防止极端场景下单轮过长 **/ + private Integer roundMaxFiles = 500; + + /** 单个文件大小上限（MB）。解包时整个 entry 读进堆，upload 再拷一份、MD5 再一份，峰值约 2-3 倍文件大小，超限直接标 .error 而不是解包到 OOM **/ + private Integer maxUnpackSizeMb = 512; + + /** 校验不通过或解包异常，且文件 mtime 静止超过该秒数，判定为坏包并标 .error 停止重试。区分「光闸还在搬」与「搬完了但是坏包」 **/ + private Integer staleAfterSeconds = 300; } @Data diff --git a/backend/src/main/java/com/senmesh/sfile/service/impl/TransferFileMonitorServiceImpl.java b/backend/src/main/java/com/senmesh/sfile/service/impl/TransferFileMonitorServiceImpl.java index 80f9a8f..72e40ec 100644 --- a/backend/src/main/java/com/senmesh/sfile/service/impl/TransferFileMonitorServiceImpl.java +++ b/backend/src/main/java/com/senmesh/sfile/service/impl/TransferFileMonitorServiceImpl.java @@ -36,7 +36,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.nio.file.StandardCopyOption; -import java.nio.file.attribute.FileTime; +import java.util.ArrayList; import java.util.Comparator; import java.util.Enumeration; import java.util.HashMap; @@ -57,6 +57,25 @@ public class TransferFileMonitorServiceImpl implements TransferFileMonitorServic /** 接收目录多为 bind mount，单次 syscall 成本可达毫秒级，小缓冲会把耗时放大数十倍 */ private static final int TRANSFER_BUFFER_SIZE = 64 * 1024; + /** 目录剩余待处理包超过该数即告警，是容量规划的直接依据 */ + private static final int BACKLOG_WARN_THRESHOLD = 50; + + private static final Comparator MTIME_ASC = new Comparator() { + @Override + public int compare(Path a, Path b) { + return Long.compare(mtimeMillis(a), mtimeMillis(b)); + } + }; + + private enum ProcessOutcome { + /** 解包入库成功，包已删除 */ + SUCCESS, + /** 永久性失败，已改名 .error 停止重试 */ + ERROR, + /** 本轮不处理（光闸还在搬、校验未通过但未静止够久），留给下一轮 */ + SKIPPED + } + @Autowired private FileProperties fileProperties; @@ -77,8 +96,9 @@ public class TransferFileMonitorServiceImpl implements TransferFileMonitorServic @Override public void scanAndProcess() { - String transferReceiveDir = fileProperties.getTransfer().getReceiveDir(); - Path receiveDir = Paths.get(transferReceiveDir); + FileProperties.Transfer transfer = fileProperties.getTransfer(); + FileProperties.Monitor monitor = transfer.getMonitor(); + Path receiveDir = Paths.get(transfer.getReceiveDir()); try { if (!Files.exists(receiveDir)) { @@ -87,47 +107,57 @@ public class TransferFileMonitorServiceImpl implements TransferFileMonitorServic return; } - // 1. 扫描 .zip 文件（解包逻辑） - List zipFiles = Files.list(receiveDir) + long roundStart = System.currentTimeMillis(); + long budgetMillis = monitor.getRoundBudgetSeconds() * 1000L; + int maxFiles = monitor.getRoundMaxFiles(); + + List candidates = Files.list(receiveDir) .filter(Files::isRegularFile) - .filter(p -> p.toString().endsWith(".zip")) - .filter(p -> !p.toString().endsWith(".tmp")) - .sorted(Comparator.comparing(p -> { - try { - return Files.getLastModifiedTime(p); - } catch (IOException e) { - return FileTime.fromMillis(0); - } - })) - .limit(10) .collect(Collectors.toList()); - if (!zipFiles.isEmpty()) { - log.info("发现 {} 个待处理的传输文件", zipFiles.size()); - for (Path zipFile : zipFiles) { - processZipFile(zipFile); + // 1. .finish 只做一次状态更新，极轻：优先处理且不限个数，让发送端尽早看到「已完成」 + List finishFiles = filterBySuffix(candidates, ".finish"); + int finishDone = 0; + for (Path finishFile : finishFiles) { + if (processFinishFile(finishFile)) { + finishDone++; } } - // 2. 扫描 .finish 文件（完成标记逻辑） - List finishFiles = Files.list(receiveDir) - .filter(Files::isRegularFile) - .filter(p -> p.toString().endsWith(".finish")) - .sorted(Comparator.comparing(p -> { - try { - return Files.getLastModifiedTime(p); - } catch (IOException e) { - return FileTime.fromMillis(0); - } - })) - .limit(10) - .collect(Collectors.toList()); - - if (!finishFiles.isEmpty()) { - log.info("发现 {} 个完成标记文件", finishFiles.size()); - for (Path finishFile : finishFiles) { - processFinishFile(finishFile); + // 2. .zip 受时间预算与个数上限约束，按 mtime 升序保证先到先处理 + List zipFiles = filterBySuffix(candidates, ".zip"); + int success = 0; + int error = 0; + int skipped = 0; + int attempted = 0; + for (Path zipFile : zipFiles) { + if (attempted >= maxFiles || System.currentTimeMillis() - roundStart > budgetMillis) { + break; } + attempted++; + switch (processZipFile(zipFile)) { + case SUCCESS: + success++; + break; + case ERROR: + error++; + break; + default: + skipped++; + break; + } + } + + int remaining = zipFiles.size() - attempted; + int parked = filterBySuffix(candidates, ".error").size(); + if (attempted > 0 || finishDone > 0 || remaining > 0) { + log.info("传输监听本轮结束: 完成标记={}/{}, 传输包处理={}(成功={},坏包={},未就绪={}), 耗时={}ms, 预算={}ms, 目录剩余待处理={}, 已搁置坏包={}", + finishDone, finishFiles.size(), attempted, success, error, skipped, + System.currentTimeMillis() - roundStart, budgetMillis, remaining, parked); + } + if (remaining > BACKLOG_WARN_THRESHOLD) { + log.warn("传输接收目录积压 {} 个待处理包，超过告警阈值 {}，可调大 file.transfer.monitor.round-budget-seconds", + remaining, BACKLOG_WARN_THRESHOLD); } } catch (Exception e) { @@ -135,44 +165,110 @@ public class TransferFileMonitorServiceImpl implements TransferFileMonitorServic } } + private static long mtimeMillis(Path path) { + try { + return Files.getLastModifiedTime(path).toMillis(); + } catch (IOException e) { + return 0L; + } + } + + private List filterBySuffix(List files, String suffix) { + List matched = new ArrayList<>(); + for (Path p : files) { + if (p.toString().endsWith(suffix)) { + matched.add(p); + } + } + matched.sort(MTIME_ASC); + return matched; + } + /** * 处理单个 ZIP 文件（解包逻辑） */ - private void processZipFile(Path zipFile) { + private ProcessOutcome processZipFile(Path zipFile) { String fileName = zipFile.getFileName().toString(); + FileProperties.Monitor monitor = fileProperties.getTransfer().getMonitor(); try { // 1. 校验 ZIP 包内文件大小是否匹配，并返回元数据 TransferMetadataDTO metadata = verifyFileMD5(zipFile); if (metadata == null) { - // 校验不通过，跳过 - return; + // 返回 null 有两重含义：光闸还在搬（下一轮就好了）与包本身是坏的（永远不会好）。 + // 用 mtime 是否静止区分，否则 5 秒轮询会为坏包无限白跑 + if (isStale(zipFile, monitor.getStaleAfterSeconds())) { + renameToError(zipFile, "校验持续不通过且文件已静止超过 " + monitor.getStaleAfterSeconds() + " 秒，判定为坏包"); + return ProcessOutcome.ERROR; + } + return ProcessOutcome.SKIPPED; + } + + // 2. 单个文件大小上限：解包把整个 entry 读进堆，upload 再拷一份、MD5 再一份，峰值约 2-3 倍 + long largestEntry = largestEntrySize(zipFile); + long limitBytes = monitor.getMaxUnpackSizeMb() * 1024L * 1024L; + if (largestEntry > limitBytes) { + renameToError(zipFile, "包内最大单文件 " + (largestEntry / 1048576L) + "MB 超过解包上限 " + + monitor.getMaxUnpackSizeMb() + "MB，解包会撑爆堆内存"); + return ProcessOutcome.ERROR; } if (!resolveTargetUser(zipFile, metadata)) { - return; + // resolveTargetUser 内部已改名 .error：用户映射失败是永久性的 + return ProcessOutcome.ERROR; } log.info("文件大小校验通过，开始解包: {}", fileName); - // 2. 解包并创建用户文件（传入已解析的元数据，直接创建为正常状态） + // 3. 解包并创建用户文件（传入已解析的元数据，直接创建为正常状态） ExpandUserFileDTO result = unpackTransferFileInternal(zipFile, metadata); - // 3. MeshSafe文件安全检测(异步送检,不影响解包流程) + // 4. MeshSafe文件安全检测(异步送检,不影响解包流程) submitSecurityScan(result); - // 4. 创建 .finish 文件到发送目录，告诉发送方处理完成 + // 5. 创建 .finish 文件到发送目录，告诉发送方处理完成 createFinishFile(fileName); - // 5. 删除已处理的 ZIP 文件 + // 6. 删除已处理的 ZIP 文件 Files.delete(zipFile); log.info("处理完成，已删除 ZIP 文件: {}", fileName); + return ProcessOutcome.SUCCESS; } catch (Exception e) { - log.error("处理文件失败，跳过: {}", fileName, e); + log.error("处理文件失败: {}", fileName, e); + // 短暂的基础设施抖动（数据库/Redis 不可用）应保留重试，只有持续失败超过静止阈值才搁置， + // 否则一个每轮都抛异常的坏包会永久占用时间预算 + if (isStale(zipFile, monitor.getStaleAfterSeconds())) { + renameToError(zipFile, "解包持续异常且文件已静止超过 " + monitor.getStaleAfterSeconds() + " 秒: " + e.getMessage()); + return ProcessOutcome.ERROR; + } + return ProcessOutcome.SKIPPED; } } + /** 只读 ZIP 中央目录取包内最大单文件的原始大小，不解压内容 */ + private long largestEntrySize(Path zipFile) throws IOException { + long largest = 0L; + try (ZipFile zf = new ZipFile(zipFile.toFile())) { + Enumeration entries = zf.entries(); + while (entries.hasMoreElements()) { + ZipEntry entry = entries.nextElement(); + if (!entry.isDirectory() && entry.getSize() > largest) { + largest = entry.getSize(); + } + } + } + return largest; + } + + /** 文件 mtime 静止超过阈值即认为光闸已搬完，此后的失败都是永久性的 */ + private boolean isStale(Path file, int staleAfterSeconds) { + if (!Files.exists(file)) { + return false; + } + return System.currentTimeMillis() - mtimeMillis(file) > staleAfterSeconds * 1000L; + } + /** * MeshSafe文件安全检测:解包产生的每个文件异步送检,失败不影响解包流程 */ @@ -233,10 +329,10 @@ public class TransferFileMonitorServiceImpl implements TransferFileMonitorServic Path errorFile = zipFile.resolveSibling(fileName + ".error"); try { Files.move(zipFile, errorFile, StandardCopyOption.REPLACE_EXISTING); - log.error("传输文件停止解包并改名为 .error: file={}, errorFile={}, reason={}", + log.error("传输文件停止处理并改名为 .error: file={}, errorFile={}, reason={}", fileName, errorFile.getFileName(), reason); } catch (Exception e) { - log.error("传输文件停止解包但改名 .error 失败: file={}, reason={}", fileName, reason, e); + log.error("传输文件停止处理但改名 .error 失败: file={}, reason={}", fileName, reason, e); } } @@ -361,7 +457,9 @@ public class TransferFileMonitorServiceImpl implements TransferFileMonitorServic return metadata; } catch (Exception e) { - log.error("文件大小校验失败: {}", fileName, e); + // 光闸搬运未完成时 ZIP 中央目录还不存在，打不开是正常状态而非错误； + // 真正的坏包由调用方在文件静止超过阈值后改名 .error 并记 ERROR，此处不重复刷日志 + log.debug("文件大小校验失败（可能还在传输中）: {}", fileName, e); return null; } } @@ -562,8 +660,9 @@ public class TransferFileMonitorServiceImpl implements TransferFileMonitorServic * 处理 .finish 文件（发送方接收到对方的完成标记） * * @param finishFile .finish 文件路径（例如：123_456.zip.finish） + * @return 是否处理成功（成功会删除该标记文件） */ - private void processFinishFile(Path finishFile) { + private boolean processFinishFile(Path finishFile) { String fileName = finishFile.getFileName().toString(); log.info("处理完成标记文件: {}", fileName); @@ -574,12 +673,20 @@ public class TransferFileMonitorServiceImpl implements TransferFileMonitorServic String[] parts = zipFileName.replace(".zip", "").split("_"); if (parts.length != 2) { - log.warn("无法解析完成标记文件名，跳过: {}", fileName); - return; + // 文件名不合规是永久性的，改名 .error 停止重试，否则每轮都白跑一次并刷日志 + renameToError(finishFile, "完成标记文件名无法解析出 userId_folderId"); + return false; } - Long userId = Long.parseLong(parts[0]); - Long folderId = Long.parseLong(parts[1]); + Long userId; + Long folderId; + try { + userId = Long.parseLong(parts[0]); + folderId = Long.parseLong(parts[1]); + } catch (NumberFormatException e) { + renameToError(finishFile, "完成标记文件名的 userId/folderId 不是数字: " + e.getMessage()); + return false; + } log.info("解析完成标记文件: folderId={}, userId={}", folderId, userId); @@ -595,9 +702,12 @@ public class TransferFileMonitorServiceImpl implements TransferFileMonitorServic // 3. 删除 .finish 文件 Files.delete(finishFile); log.info("处理完成，已删除完成标记文件: {}", fileName); + return true; } catch (Exception e) { + // 数据库/Redis 抖动属短暂故障，保留文件下一轮重试 log.error("处理完成标记文件失败: {}", fileName, e); + return false; } } } diff --git a/deploy/config/application-iw.yml b/deploy/config/application-iw.yml index a4a292e..ac7827f 100644 --- a/deploy/config/application-iw.yml +++ b/deploy/config/application-iw.yml @@ -78,6 +78,14 @@ file: send-dir: /data/transfer/send receive-dir: /data/transfer/receive unpack-risk-check: false + monitor: + # 单轮时间预算（秒），与 5 秒监听周期配套；个数不代表工作量，时间才代表 + round-budget-seconds: 4 + round-max-files: 500 + # 包内单个文件超过该大小（MB）直接标 .error，避免整文件读进堆导致 OOM + max-unpack-size-mb: 512 + # 校验不过或解包异常且文件静止超过该秒数即判坏包，标 .error 停止重试 + stale-after-seconds: 300 dify: risk: diff --git a/deploy/config/application-ow.yml b/deploy/config/application-ow.yml index 82132f4..ad1a217 100644 --- a/deploy/config/application-ow.yml +++ b/deploy/config/application-ow.yml @@ -78,6 +78,14 @@ file: send-dir: /data/transfer/send receive-dir: /data/transfer/receive unpack-risk-check: false + monitor: + # 单轮时间预算（秒），与 5 秒监听周期配套；个数不代表工作量，时间才代表 + round-budget-seconds: 4 + round-max-files: 500 + # 包内单个文件超过该大小（MB）直接标 .error，避免整文件读进堆导致 OOM + max-unpack-size-mb: 512 + # 校验不过或解包异常且文件静止超过该秒数即判坏包，标 .error 停止重试 + stale-after-seconds: 300 dify: risk: diff --git a/deploy/offline/config/application.yml b/deploy/offline/config/application.yml index 2790fa8..b40682b 100644 --- a/deploy/offline/config/application.yml +++ b/deploy/offline/config/application.yml @@ -74,6 +74,19 @@ file: # location / send-dir / receive-dir / monitor.cron 均由环境变量注入 # 解包时是否再做一次内容风险识别（依赖下面的 dify，未部署 dify 时保持 false） unpack-risk-check: false + monitor: + # 单轮处理时间预算（秒）。超预算即结束本轮，剩余留给下一轮。 + # 与监听周期配套：周期 5 秒则预算留 4 秒，保证两轮不互相挤压。 + # 目录持续积压（日志「目录剩余待处理」超过 50）时调大此值 + round-budget-seconds: 4 + # 单轮兜底个数上限，防止极端场景单轮过长 + round-max-files: 500 + # 包内单个文件大小上限（MB）。解包时整个文件读进堆，峰值约 2-3 倍文件大小， + # 超限直接标 .error 而不是解包到 OOM。按 APP_JAVA_OPTS 的 -Xmx 调整：-Xmx3g 对应 512 + max-unpack-size-mb: 512 + # 校验不通过或解包异常，且文件静止超过该秒数即判定为坏包，标 .error 停止重试。 + # 用于区分「光闸还在搬」与「搬完了但是坏包」，必须大于光闸搬运单个包的最长时间 + stale-after-seconds: 300 # 内容风险识别（Dify）。未部署 Dify 时留空并保持 file.transfer.unpack-risk-check=false dify: