perf(传输): 单轮处理上限从「10 个」改为「4 秒时间预算」

个数不代表工作量:几十 KB 与几百 MB 的包差三个数量级,10 个小包 1 秒就跑完
(限制毫无必要),10 个大包能跑十几分钟(还会把下一轮挤掉)。改为时间预算后
单轮能处理多少就处理多少,超预算结束本轮、剩余留给下一轮,个数上限 500 兜底。
按 5.3 改造后的单包耗时估算,4 秒预算约可处理数十个小包,吞吐提升一个数量级。

同时解决三个会让预算被白耗的问题:
- 校验不通过原来只有一种处理(静默跳过、永不改名),但它有两重含义:光闸还在搬
  (下一轮就好)与包本身坏了(永远不会好)。用 mtime 是否静止超过 stale-after-seconds
  区分,坏包改名 .error 停止重试,否则 5 秒轮询会为它无限白跑
- 解包异常同样按静止阈值判定,短暂基础设施抖动仍保留重试
- .finish 文件名不合规原来只 log.warn 后保留文件,每轮重复解析并刷日志,改为标 .error

.finish 只做一次状态更新,改为优先处理且不受个数限制,让发送端尽早看到「已完成」。
每轮结束输出结构化日志(完成标记/成功/坏包/未就绪/耗时/预算/剩余/已搁置),
剩余超过 50 告警,作为容量规划依据。

按方案 5.3 注意事项补 max-unpack-size-mb(默认 512):解包把整个 entry 读进堆,
upload 再拷一份、MD5 再一份,峰值约 2-3 倍文件大小,-Xmx3g 下单文件过大会 OOM,
超限直接标 .error 而不是尝试解包。

verifyFileMD5 打不开 zip 时原来记 ERROR 带堆栈,但「光闸还在搬」本来就是这个状态,
5 秒轮询下会为每个在途文件刷 60 条 ERROR,降为 debug;真正的坏包由改名 .error 时记 ERROR。

未实施方案里的认领改名与并发解包:两侧都是单实例部署,Spring 的 cron 任务要等上一轮
跑完才排下一轮(已实测触发间隔精确 5.00 秒),不存在同实例重叠;认领机制只会引入
kill -9 后遗留 .processing 这类新故障模式,换不到吞吐。若将来改多实例共享同一摆渡目录,
再补认领与 Redisson 互斥。
This commit is contained in:
KodouWZF
2026-11-03 22:42:06 +08:00
parent 75ab51e13c
commit 06852c5d90
5 changed files with 212 additions and 54 deletions

View File

@@ -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

View File

@@ -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<Path> MTIME_ASC = new Comparator<Path>() {
@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,90 +107,166 @@ public class TransferFileMonitorServiceImpl implements TransferFileMonitorServic
return;
}
// 1. 扫描 .zip 文件(解包逻辑)
List<Path> zipFiles = Files.list(receiveDir)
long roundStart = System.currentTimeMillis();
long budgetMillis = monitor.getRoundBudgetSeconds() * 1000L;
int maxFiles = monitor.getRoundMaxFiles();
List<Path> 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);
}
}
// 2. 扫描 .finish 文件(完成标记逻辑)
List<Path> 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());
// 1. .finish 只做一次状态更新,极轻:优先处理且不限个数,让发送端尽早看到「已完成」
List<Path> finishFiles = filterBySuffix(candidates, ".finish");
int finishDone = 0;
for (Path finishFile : finishFiles) {
processFinishFile(finishFile);
if (processFinishFile(finishFile)) {
finishDone++;
}
}
// 2. .zip 受时间预算与个数上限约束,按 mtime 升序保证先到先处理
List<Path> 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) {
log.error("扫描传输目录失败", e);
}
}
private static long mtimeMillis(Path path) {
try {
return Files.getLastModifiedTime(path).toMillis();
} catch (IOException e) {
return 0L;
}
}
private List<Path> filterBySuffix(List<Path> files, String suffix) {
List<Path> 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<? extends ZipEntry> 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;
}
/**
@@ -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;
}
}
}

View File

@@ -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:

View File

@@ -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:

View File

@@ -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: