完成品牌删除模块
This commit is contained in:
BIN
backend-java/aiimage-backend.log.2026-03-28.0.gz
Normal file
BIN
backend-java/aiimage-backend.log.2026-03-28.0.gz
Normal file
Binary file not shown.
4449
backend-java/aiimage-backend.log.2026-03-28.0258380216510400.tmp
Normal file
4449
backend-java/aiimage-backend.log.2026-03-28.0258380216510400.tmp
Normal file
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,16 @@
|
||||
package com.nanri.aiimage.config;
|
||||
|
||||
import lombok.Data;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
@Data
|
||||
@ConfigurationProperties(prefix = "aiimage.delete-brand-progress")
|
||||
public class DeleteBrandProgressProperties {
|
||||
private long heartbeatTimeoutMinutes = 15;
|
||||
private String staleCheckCron = "0 */2 * * * *";
|
||||
|
||||
/**
|
||||
* 定时补偿 finalize:用于“分片其实已齐,但最后一次提交断开导致没触发 finalize”的场景。
|
||||
*/
|
||||
private String finalizeCheckCron = "30 */2 * * * *";
|
||||
}
|
||||
@@ -4,6 +4,6 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
@Configuration
|
||||
@EnableConfigurationProperties({OssProperties.class, StorageProperties.class, BrandProgressProperties.class, ZiniaoProperties.class, ModuleCleanupProperties.class})
|
||||
@EnableConfigurationProperties({OssProperties.class, StorageProperties.class, BrandProgressProperties.class, DeleteBrandProgressProperties.class, ZiniaoProperties.class, ModuleCleanupProperties.class})
|
||||
public class PropertiesConfig {
|
||||
}
|
||||
|
||||
@@ -3,8 +3,10 @@ package com.nanri.aiimage.modules.deletebrand.controller;
|
||||
import com.nanri.aiimage.common.api.ApiResponse;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandRunRequest;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandSubmitResultRequest;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandTaskBatchRequest;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandHistoryVo;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandRunVo;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandTaskBatchVo;
|
||||
import com.nanri.aiimage.modules.deletebrand.service.DeleteBrandRunService;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import io.swagger.v3.oas.annotations.Parameter;
|
||||
@@ -75,6 +77,12 @@ public class DeleteBrandRunController {
|
||||
return ApiResponse.success(deleteBrandRunService.getTaskDetail(taskId));
|
||||
}
|
||||
|
||||
@PostMapping("/tasks/batch")
|
||||
@Operation(summary = "批量获取删除品牌任务详情", description = "合并多个 taskId 的详情查询,减少轮询期对 DB/Redis 的压力。")
|
||||
public ApiResponse<DeleteBrandTaskBatchVo> getTasksBatch(@Valid @RequestBody DeleteBrandTaskBatchRequest request) {
|
||||
return ApiResponse.success(deleteBrandRunService.getTaskDetails(request.getTaskIds()));
|
||||
}
|
||||
|
||||
@PostMapping("/tasks/{taskId}/result")
|
||||
@Operation(summary = "提交删除品牌处理结果", description = "插件/前端分片回传处理结果;后端累计分片并推进进度,全部完成后再执行最终组装与落库。")
|
||||
public ApiResponse<Void> submitResult(
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
package com.nanri.aiimage.modules.deletebrand.model.dto;
|
||||
|
||||
import io.swagger.v3.oas.annotations.media.Schema;
|
||||
import jakarta.validation.constraints.NotEmpty;
|
||||
import lombok.Data;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
@Data
|
||||
@Schema(description = "删除品牌任务批量详情请求")
|
||||
public class DeleteBrandTaskBatchRequest {
|
||||
|
||||
@NotEmpty(message = "taskIds 不能为空")
|
||||
@Schema(description = "任务ID列表")
|
||||
private List<Long> taskIds = new ArrayList<>();
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
package com.nanri.aiimage.modules.deletebrand.model.vo;
|
||||
|
||||
import io.swagger.v3.oas.annotations.media.Schema;
|
||||
import lombok.Data;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
@Data
|
||||
@Schema(description = "删除品牌任务批量详情响应")
|
||||
public class DeleteBrandTaskBatchVo {
|
||||
|
||||
@Schema(description = "任务详情列表,按请求顺序返回")
|
||||
private List<DeleteBrandTaskDetailVo> items = new ArrayList<>();
|
||||
|
||||
@Schema(description = "不存在/无权限的 taskId")
|
||||
private List<Long> missingTaskIds = new ArrayList<>();
|
||||
}
|
||||
@@ -7,6 +7,7 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.nanri.aiimage.common.exception.BusinessException;
|
||||
import com.nanri.aiimage.config.DeleteBrandProgressProperties;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.cache.DeleteBrandParsedFileCacheDto;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandCountryResultItemDto;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandProcessedCountryDto;
|
||||
@@ -66,6 +67,7 @@ public class DeleteBrandRunService {
|
||||
private final com.nanri.aiimage.modules.ziniao.service.ZiniaoAuthService ziniaoAuthService;
|
||||
private final OssStorageService ossStorageService;
|
||||
private final ObjectMapper objectMapper;
|
||||
private final DeleteBrandProgressProperties deleteBrandProgressProperties;
|
||||
|
||||
public DeleteBrandRunVo run(DeleteBrandRunRequest request) {
|
||||
if (request.getUserId() == null || request.getUserId() <= 0) {
|
||||
@@ -206,27 +208,72 @@ public class DeleteBrandRunService {
|
||||
|
||||
public DeleteBrandHistoryVo listHistory(Long userId) {
|
||||
DeleteBrandHistoryVo vo = new DeleteBrandHistoryVo();
|
||||
vo.setItems(fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||
.eq(FileResultEntity::getUserId, userId)
|
||||
.orderByDesc(FileResultEntity::getCreatedAt)
|
||||
.last("limit 100"))
|
||||
.stream()
|
||||
.map(entity -> {
|
||||
DeleteBrandResultItemVo item = new DeleteBrandResultItemVo();
|
||||
item.setResultId(entity.getId());
|
||||
item.setTaskId(entity.getTaskId());
|
||||
item.setFileKey(entity.getSourceFileUrl());
|
||||
item.setSourceFilename(entity.getSourceFilename());
|
||||
item.setShopName(entity.getSourceFilename() == null ? null : FileUtil.mainName(entity.getSourceFilename()));
|
||||
item.setOutputFilename(entity.getResultFilename());
|
||||
item.setDownloadUrl(entity.getResultFileUrl() == null ? null : ossStorageService.getPublicUrl(entity.getResultFileUrl()));
|
||||
item.setTotalRows(entity.getRowCount());
|
||||
item.setSuccess(entity.getSuccess() != null && entity.getSuccess() == 1);
|
||||
item.setError(entity.getErrorMessage());
|
||||
return item;
|
||||
})
|
||||
.toList());
|
||||
List<FileResultEntity> entities = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||
.eq(FileResultEntity::getUserId, userId)
|
||||
.orderByDesc(FileResultEntity::getCreatedAt)
|
||||
.last("limit 100"));
|
||||
|
||||
// 补充从 task.resultJson 还原匹配信息/真实失败原因(历史表本身不存这些字段)
|
||||
Map<Long, List<DeleteBrandResultItemVo>> itemsByTaskId = new LinkedHashMap<>();
|
||||
List<Long> taskIds = entities.stream()
|
||||
.map(FileResultEntity::getTaskId)
|
||||
.filter(id -> id != null && id > 0)
|
||||
.distinct()
|
||||
.toList();
|
||||
if (!taskIds.isEmpty()) {
|
||||
List<FileTaskEntity> tasks = fileTaskMapper.selectBatchIds(taskIds);
|
||||
for (FileTaskEntity task : tasks) {
|
||||
if (task == null || task.getResultJson() == null || task.getResultJson().isBlank()) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
List<DeleteBrandResultItemVo> taskItems = objectMapper.readValue(task.getResultJson(), new TypeReference<List<DeleteBrandResultItemVo>>() {
|
||||
});
|
||||
itemsByTaskId.put(task.getId(), taskItems);
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
vo.setItems(entities.stream().map(entity -> {
|
||||
DeleteBrandResultItemVo item = new DeleteBrandResultItemVo();
|
||||
item.setResultId(entity.getId());
|
||||
item.setTaskId(entity.getTaskId());
|
||||
item.setFileKey(entity.getSourceFileUrl());
|
||||
item.setSourceFilename(entity.getSourceFilename());
|
||||
item.setShopName(entity.getSourceFilename() == null ? null : FileUtil.mainName(entity.getSourceFilename()));
|
||||
item.setOutputFilename(entity.getResultFilename());
|
||||
item.setDownloadUrl(entity.getResultFileUrl() == null ? null : ossStorageService.getPublicUrl(entity.getResultFileUrl()));
|
||||
item.setTotalRows(entity.getRowCount());
|
||||
item.setSuccess(entity.getSuccess() != null && entity.getSuccess() == 1);
|
||||
item.setError(entity.getErrorMessage());
|
||||
|
||||
// 以 task.resultJson 中的项为准补全 matched/shopId/platform/openStoreUrl/错误信息
|
||||
if (entity.getTaskId() != null) {
|
||||
List<DeleteBrandResultItemVo> taskItems = itemsByTaskId.get(entity.getTaskId());
|
||||
if (taskItems != null && entity.getSourceFilename() != null && !entity.getSourceFilename().isBlank()) {
|
||||
for (DeleteBrandResultItemVo candidate : taskItems) {
|
||||
if (candidate == null) {
|
||||
continue;
|
||||
}
|
||||
if (entity.getSourceFilename().equals(candidate.getSourceFilename())) {
|
||||
item.setMatched(candidate.isMatched());
|
||||
item.setShopId(candidate.getShopId());
|
||||
item.setPlatform(candidate.getPlatform());
|
||||
item.setOpenStoreUrl(candidate.getOpenStoreUrl());
|
||||
// 历史表可能只有通用 errorMessage;优先返回任务当时的真实 error
|
||||
if ((item.getError() == null || item.getError().isBlank()) && candidate.getError() != null && !candidate.getError().isBlank()) {
|
||||
item.setError(candidate.getError());
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return item;
|
||||
}).toList());
|
||||
|
||||
return vo;
|
||||
}
|
||||
|
||||
@@ -391,7 +438,46 @@ public class DeleteBrandRunService {
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
|
||||
throw new BusinessException("任务不存在");
|
||||
}
|
||||
return buildTaskDetail(task);
|
||||
}
|
||||
|
||||
public com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandTaskBatchVo getTaskDetails(java.util.List<Long> taskIds) {
|
||||
com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandTaskBatchVo vo = new com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandTaskBatchVo();
|
||||
if (taskIds == null || taskIds.isEmpty()) {
|
||||
return vo;
|
||||
}
|
||||
|
||||
java.util.List<Long> normalizedTaskIds = taskIds.stream()
|
||||
.filter(id -> id != null && id > 0)
|
||||
.distinct()
|
||||
.limit(50)
|
||||
.toList();
|
||||
if (normalizedTaskIds.isEmpty()) {
|
||||
return vo;
|
||||
}
|
||||
|
||||
java.util.Map<Long, FileTaskEntity> taskById = fileTaskMapper.selectBatchIds(normalizedTaskIds).stream()
|
||||
.filter(task -> task != null && MODULE_TYPE.equals(task.getModuleType()))
|
||||
.collect(java.util.stream.Collectors.toMap(FileTaskEntity::getId, task -> task, (left, right) -> left, java.util.LinkedHashMap::new));
|
||||
|
||||
java.util.Map<Long, java.util.Map<Object, Object>> progressByTaskId = deleteBrandTaskCacheService.getProgressBatch(normalizedTaskIds);
|
||||
|
||||
for (Long taskId : normalizedTaskIds) {
|
||||
FileTaskEntity task = taskById.get(taskId);
|
||||
if (task == null) {
|
||||
vo.getMissingTaskIds().add(taskId);
|
||||
continue;
|
||||
}
|
||||
vo.getItems().add(buildTaskDetail(task, progressByTaskId.get(taskId)));
|
||||
}
|
||||
return vo;
|
||||
}
|
||||
|
||||
private DeleteBrandTaskDetailVo buildTaskDetail(FileTaskEntity task) {
|
||||
return buildTaskDetail(task, null);
|
||||
}
|
||||
|
||||
private DeleteBrandTaskDetailVo buildTaskDetail(FileTaskEntity task, java.util.Map<Object, Object> progress) {
|
||||
DeleteBrandTaskItemVo taskVo = new DeleteBrandTaskItemVo();
|
||||
taskVo.setId(task.getId());
|
||||
taskVo.setTaskNo(task.getTaskNo());
|
||||
@@ -403,10 +489,16 @@ public class DeleteBrandRunService {
|
||||
taskVo.setCreatedAt(task.getCreatedAt());
|
||||
taskVo.setUpdatedAt(task.getUpdatedAt());
|
||||
taskVo.setFinishedAt(task.getFinishedAt());
|
||||
taskVo.setDownloadUrl(resolveTaskDownloadUrl(task.getId()));
|
||||
taskVo.setDownloadFilename(resolveTaskDownloadFilename(task.getId()));
|
||||
// 轮询期避免每次都查 biz_file_result(会放大 DB 压力);仅任务成功后再提供下载信息
|
||||
if ("SUCCESS".equals(task.getStatus())) {
|
||||
taskVo.setDownloadUrl(resolveTaskDownloadUrl(task.getId()));
|
||||
taskVo.setDownloadFilename(resolveTaskDownloadFilename(task.getId()));
|
||||
} else {
|
||||
taskVo.setDownloadUrl(null);
|
||||
taskVo.setDownloadFilename(null);
|
||||
}
|
||||
|
||||
DeleteBrandLineProgressVo progressVo = buildLineProgress(taskId);
|
||||
DeleteBrandLineProgressVo progressVo = buildLineProgress(task.getId(), progress);
|
||||
|
||||
DeleteBrandTaskDetailVo detail = new DeleteBrandTaskDetailVo();
|
||||
detail.setTask(taskVo);
|
||||
@@ -415,23 +507,27 @@ public class DeleteBrandRunService {
|
||||
}
|
||||
|
||||
private DeleteBrandLineProgressVo buildLineProgress(Long taskId) {
|
||||
Map<Object, Object> progress = deleteBrandTaskCacheService.getProgress(taskId);
|
||||
return buildLineProgress(taskId, null);
|
||||
}
|
||||
|
||||
private DeleteBrandLineProgressVo buildLineProgress(Long taskId, Map<Object, Object> progress) {
|
||||
Map<Object, Object> resolvedProgress = progress != null ? progress : deleteBrandTaskCacheService.getProgress(taskId);
|
||||
DeleteBrandLineProgressVo vo = new DeleteBrandLineProgressVo();
|
||||
if (progress == null || progress.isEmpty()) {
|
||||
if (resolvedProgress == null || resolvedProgress.isEmpty()) {
|
||||
vo.setHas_progress(false);
|
||||
vo.setInfo(null);
|
||||
return vo;
|
||||
}
|
||||
DeleteBrandLineProgressInfoVo info = new DeleteBrandLineProgressInfoVo();
|
||||
info.setFile_index(parseInt(progress.get("file_index")));
|
||||
info.setFile_total(parseInt(progress.get("file_total")));
|
||||
info.setFile_name(stringValue(progress.get("file_name")));
|
||||
info.setCurrent_line(parseInt(progress.get("current_line")));
|
||||
info.setTotal_lines(parseInt(progress.get("total_lines")));
|
||||
info.setCurrent_country(stringValue(progress.get("current_country")));
|
||||
info.setCurrent_asin(stringValue(progress.get("current_asin")));
|
||||
info.setFinished_files(parseInt(progress.get("finished_files")));
|
||||
info.setPhase(stringValue(progress.get("phase")));
|
||||
info.setFile_index(parseInt(resolvedProgress.get("file_index")));
|
||||
info.setFile_total(parseInt(resolvedProgress.get("file_total")));
|
||||
info.setFile_name(stringValue(resolvedProgress.get("file_name")));
|
||||
info.setCurrent_line(parseInt(resolvedProgress.get("current_line")));
|
||||
info.setTotal_lines(parseInt(resolvedProgress.get("total_lines")));
|
||||
info.setCurrent_country(stringValue(resolvedProgress.get("current_country")));
|
||||
info.setCurrent_asin(stringValue(resolvedProgress.get("current_asin")));
|
||||
info.setFinished_files(parseInt(resolvedProgress.get("finished_files")));
|
||||
info.setPhase(stringValue(resolvedProgress.get("phase")));
|
||||
vo.setHas_progress(true);
|
||||
vo.setInfo(info);
|
||||
return vo;
|
||||
@@ -524,6 +620,9 @@ public class DeleteBrandRunService {
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
|
||||
throw new BusinessException("任务不存在");
|
||||
}
|
||||
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
|
||||
throw new BusinessException("任务已结束,拒绝继续提交结果");
|
||||
}
|
||||
|
||||
Map<String, DeleteBrandParsedFileCacheDto> parsedPayload = deleteBrandTaskCacheService.getParsedPayload(taskId,
|
||||
new TypeReference<Map<String, DeleteBrandParsedFileCacheDto>>() {
|
||||
@@ -548,9 +647,17 @@ public class DeleteBrandRunService {
|
||||
throw new BusinessException("文件标识不匹配: " + fileIdentity);
|
||||
}
|
||||
|
||||
// 更早的 duplicate fast-path:尽量在较重的校验/进度写入之前直接短路
|
||||
if (deleteBrandTaskCacheService.hasSameResultChunk(taskId, fileIdentity, fileDto.getChunkIndex(), fileDto)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
validateChunk(fileDto, parsedFile);
|
||||
validateResultFileAgainstParsed(fileDto, parsedFile);
|
||||
deleteBrandTaskCacheService.mergeResultChunk(taskId, fileIdentity, fileDto.getChunkIndex(), fileDto);
|
||||
boolean stored = deleteBrandTaskCacheService.mergeResultChunk(taskId, fileIdentity, fileDto.getChunkIndex(), fileDto);
|
||||
if (!stored) {
|
||||
continue;
|
||||
}
|
||||
|
||||
Map<String, String> progress = new LinkedHashMap<>();
|
||||
progress.put("phase", DeleteBrandTaskCacheService.PHASE_CRAWLING);
|
||||
@@ -566,19 +673,52 @@ public class DeleteBrandRunService {
|
||||
deleteBrandTaskCacheService.saveProgress(taskId, progress);
|
||||
}
|
||||
|
||||
tryFinalizeTask(taskId, false);
|
||||
}
|
||||
|
||||
public void tryFinalizeTask(Long taskId, boolean fromCompensation) {
|
||||
if (taskId == null || taskId <= 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
FileTaskEntity task = fileTaskMapper.selectById(taskId);
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
|
||||
return;
|
||||
}
|
||||
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
|
||||
return;
|
||||
}
|
||||
|
||||
Map<String, DeleteBrandParsedFileCacheDto> parsedPayload = deleteBrandTaskCacheService.getParsedPayload(taskId,
|
||||
new TypeReference<Map<String, DeleteBrandParsedFileCacheDto>>() {
|
||||
});
|
||||
if (parsedPayload == null || parsedPayload.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
|
||||
Map<Object, Object> progressSnapshot = deleteBrandTaskCacheService.getProgress(taskId);
|
||||
int expectedFiles = parsedPayload.size();
|
||||
Integer finishedFilesHint = parseInt(progressSnapshot == null ? null : progressSnapshot.get("finished_files"));
|
||||
if (finishedFilesHint != null && finishedFilesHint < expectedFiles) {
|
||||
return;
|
||||
}
|
||||
|
||||
Map<String, List<DeleteBrandResultFileDto>> mergedByFile = loadMergedChunks(taskId);
|
||||
int finishedFiles = countCompletedFiles(mergedByFile);
|
||||
|
||||
Map<String, String> progress = new LinkedHashMap<>();
|
||||
progress.put("finished_files", String.valueOf(finishedFiles));
|
||||
progress.put("updated_at", String.valueOf(System.currentTimeMillis()));
|
||||
if (fromCompensation) {
|
||||
progress.put("last_heartbeat_at", String.valueOf(System.currentTimeMillis()));
|
||||
}
|
||||
deleteBrandTaskCacheService.saveProgress(taskId, progress);
|
||||
|
||||
task.setSuccessFileCount(finishedFiles);
|
||||
task.setUpdatedAt(LocalDateTime.now());
|
||||
fileTaskMapper.updateById(task);
|
||||
|
||||
if (finishedFiles < parsedPayload.size()) {
|
||||
if (finishedFiles < expectedFiles) {
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
package com.nanri.aiimage.modules.deletebrand.service;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import com.nanri.aiimage.config.DeleteBrandProgressProperties;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.time.LocalDateTime;
|
||||
import java.time.ZoneId;
|
||||
import java.util.Map;
|
||||
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class DeleteBrandStaleTaskService {
|
||||
|
||||
private static final String MODULE_TYPE = "DELETE_BRAND";
|
||||
|
||||
private final FileTaskMapper fileTaskMapper;
|
||||
private final DeleteBrandTaskCacheService deleteBrandTaskCacheService;
|
||||
private final DeleteBrandRunService deleteBrandRunService;
|
||||
private final DeleteBrandProgressProperties deleteBrandProgressProperties;
|
||||
|
||||
@Scheduled(cron = "${aiimage.delete-brand-progress.stale-check-cron:0 */2 * * * *}")
|
||||
public void failStaleRunningTasks() {
|
||||
LocalDateTime threshold = LocalDateTime.now().minusMinutes(deleteBrandProgressProperties.getHeartbeatTimeoutMinutes());
|
||||
|
||||
// 先按 DB updatedAt 粗筛,避免全表扫描
|
||||
java.util.List<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
|
||||
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
|
||||
.eq(FileTaskEntity::getStatus, "RUNNING")
|
||||
.lt(FileTaskEntity::getUpdatedAt, threshold)
|
||||
.last("limit 200"));
|
||||
|
||||
for (FileTaskEntity task : runningTasks) {
|
||||
Map<Object, Object> progress = deleteBrandTaskCacheService.getProgress(task.getId());
|
||||
if (progress == null || progress.isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
long lastHeartbeatAt = 0L;
|
||||
try {
|
||||
lastHeartbeatAt = Long.parseLong(String.valueOf(progress.getOrDefault("last_heartbeat_at", "0")));
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
if (lastHeartbeatAt <= 0) {
|
||||
continue;
|
||||
}
|
||||
|
||||
LocalDateTime lastHeartbeat = LocalDateTime.ofInstant(Instant.ofEpochMilli(lastHeartbeatAt), ZoneId.systemDefault());
|
||||
if (lastHeartbeat.isAfter(threshold)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
int updated = fileTaskMapper.update(null, new com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper<FileTaskEntity>()
|
||||
.eq(FileTaskEntity::getId, task.getId())
|
||||
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
|
||||
.eq(FileTaskEntity::getStatus, "RUNNING")
|
||||
.set(FileTaskEntity::getStatus, "FAILED")
|
||||
.set(FileTaskEntity::getErrorMessage, "结果回传长时间无响应,任务已自动失败")
|
||||
.set(FileTaskEntity::getUpdatedAt, LocalDateTime.now())
|
||||
.set(FileTaskEntity::getFinishedAt, LocalDateTime.now()));
|
||||
|
||||
if (updated > 0) {
|
||||
deleteBrandTaskCacheService.saveProgress(task.getId(), java.util.Map.of(
|
||||
"phase", DeleteBrandTaskCacheService.PHASE_FAILED,
|
||||
"updated_at", String.valueOf(System.currentTimeMillis())
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Scheduled(cron = "${aiimage.delete-brand-progress.finalize-check-cron:30 */2 * * * *}")
|
||||
public void finalizeCompletedRunningTasks() {
|
||||
java.util.List<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
|
||||
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
|
||||
.eq(FileTaskEntity::getStatus, "RUNNING")
|
||||
.orderByAsc(FileTaskEntity::getUpdatedAt)
|
||||
.last("limit 100"));
|
||||
|
||||
for (FileTaskEntity task : runningTasks) {
|
||||
try {
|
||||
deleteBrandRunService.tryFinalizeTask(task.getId(), true);
|
||||
} catch (Exception ignored) {
|
||||
// 定时补偿不影响主流程;下次调度或人工重试时再处理
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -8,6 +8,7 @@ import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
@@ -19,25 +20,45 @@ public class DeleteBrandTaskCacheService {
|
||||
public static final String PHASE_FAILED = "failed";
|
||||
|
||||
private static final long PAYLOAD_TTL_HOURS = 24;
|
||||
private static final long LOCAL_PARSED_PAYLOAD_CACHE_MILLIS = 3000;
|
||||
|
||||
private final StringRedisTemplate stringRedisTemplate;
|
||||
private final ObjectMapper objectMapper;
|
||||
private final ConcurrentHashMap<Long, LocalParsedPayloadCacheEntry> parsedPayloadLocalCache = new ConcurrentHashMap<>();
|
||||
|
||||
public void saveParsedPayload(Long taskId, Object payload) {
|
||||
try {
|
||||
stringRedisTemplate.opsForValue().set(buildPayloadKey(taskId), objectMapper.writeValueAsString(payload), Duration.ofHours(PAYLOAD_TTL_HOURS));
|
||||
parsedPayloadLocalCache.put(taskId, new LocalParsedPayloadCacheEntry(System.currentTimeMillis(), payload));
|
||||
} catch (Exception ex) {
|
||||
throw new BusinessException("暂存删除品牌原始数据失败");
|
||||
}
|
||||
}
|
||||
|
||||
public <T> T getParsedPayload(Long taskId, TypeReference<T> typeReference) {
|
||||
if (taskId == null || taskId <= 0) {
|
||||
return null;
|
||||
}
|
||||
|
||||
long now = System.currentTimeMillis();
|
||||
LocalParsedPayloadCacheEntry cached = parsedPayloadLocalCache.get(taskId);
|
||||
if (cached != null && cached.isFresh(now) && cached.value() != null) {
|
||||
try {
|
||||
return objectMapper.convertValue(cached.value(), typeReference);
|
||||
} catch (Exception ignored) {
|
||||
// fallthrough to redis
|
||||
}
|
||||
}
|
||||
|
||||
String raw = stringRedisTemplate.opsForValue().get(buildPayloadKey(taskId));
|
||||
if (raw == null || raw.isBlank()) {
|
||||
parsedPayloadLocalCache.remove(taskId);
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
return objectMapper.readValue(raw, typeReference);
|
||||
T parsed = objectMapper.readValue(raw, typeReference);
|
||||
parsedPayloadLocalCache.put(taskId, new LocalParsedPayloadCacheEntry(now, parsed));
|
||||
return parsed;
|
||||
} catch (Exception ex) {
|
||||
throw new BusinessException("读取删除品牌原始数据失败");
|
||||
}
|
||||
@@ -53,7 +74,67 @@ public class DeleteBrandTaskCacheService {
|
||||
return stringRedisTemplate.opsForHash().entries(buildProgressKey(taskId));
|
||||
}
|
||||
|
||||
public void mergeResultChunk(Long taskId, String fileIdentity, Integer chunkIndex, Object chunk) {
|
||||
public java.util.Map<Long, java.util.Map<Object, Object>> getProgressBatch(java.util.List<Long> taskIds) {
|
||||
java.util.Map<Long, java.util.Map<Object, Object>> result = new java.util.LinkedHashMap<>();
|
||||
if (taskIds == null || taskIds.isEmpty()) {
|
||||
return result;
|
||||
}
|
||||
|
||||
java.util.List<Long> normalizedTaskIds = taskIds.stream()
|
||||
.filter(taskId -> taskId != null && taskId > 0)
|
||||
.distinct()
|
||||
.toList();
|
||||
if (normalizedTaskIds.isEmpty()) {
|
||||
return result;
|
||||
}
|
||||
|
||||
java.util.List<Object> pipelineResults = stringRedisTemplate.executePipelined((org.springframework.data.redis.core.RedisCallback<Object>) connection -> {
|
||||
org.springframework.data.redis.serializer.RedisSerializer<String> serializer = stringRedisTemplate.getStringSerializer();
|
||||
for (Long taskId : normalizedTaskIds) {
|
||||
connection.hGetAll(serializer.serialize(buildProgressKey(taskId)));
|
||||
}
|
||||
return null;
|
||||
});
|
||||
|
||||
for (int i = 0; i < normalizedTaskIds.size(); i++) {
|
||||
Long taskId = normalizedTaskIds.get(i);
|
||||
Object raw = pipelineResults != null && i < pipelineResults.size() ? pipelineResults.get(i) : null;
|
||||
if (raw instanceof java.util.Map<?, ?> rawMap) {
|
||||
java.util.Map<Object, Object> converted = new java.util.LinkedHashMap<>();
|
||||
for (java.util.Map.Entry<?, ?> entry : rawMap.entrySet()) {
|
||||
converted.put(entry.getKey(), entry.getValue());
|
||||
}
|
||||
result.put(taskId, converted);
|
||||
} else {
|
||||
result.put(taskId, java.util.Collections.emptyMap());
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
public boolean hasSameResultChunk(Long taskId, String fileIdentity, Integer chunkIndex, Object chunk) {
|
||||
if (chunkIndex == null || chunkIndex <= 0) {
|
||||
throw new BusinessException("chunkIndex 不合法");
|
||||
}
|
||||
if (fileIdentity == null || fileIdentity.isBlank()) {
|
||||
throw new BusinessException("fileIdentity 不能为空");
|
||||
}
|
||||
String field = fileIdentity.trim() + "#" + chunkIndex;
|
||||
String chunkJson;
|
||||
try {
|
||||
chunkJson = objectMapper.writeValueAsString(chunk);
|
||||
} catch (Exception ex) {
|
||||
throw new BusinessException("读取删除品牌结果分片失败");
|
||||
}
|
||||
Object existing = stringRedisTemplate.opsForHash().get(buildResultChunksKey(taskId), field);
|
||||
if (chunkJson.equals(existing)) {
|
||||
stringRedisTemplate.expire(buildResultChunksKey(taskId), Duration.ofHours(PAYLOAD_TTL_HOURS));
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
public boolean mergeResultChunk(Long taskId, String fileIdentity, Integer chunkIndex, Object chunk) {
|
||||
String resultKey = buildResultChunksKey(taskId);
|
||||
if (chunkIndex == null || chunkIndex <= 0) {
|
||||
throw new BusinessException("chunkIndex 不合法");
|
||||
@@ -62,12 +143,20 @@ public class DeleteBrandTaskCacheService {
|
||||
throw new BusinessException("fileIdentity 不能为空");
|
||||
}
|
||||
String field = fileIdentity.trim() + "#" + chunkIndex;
|
||||
String chunkJson;
|
||||
try {
|
||||
stringRedisTemplate.opsForHash().put(resultKey, field, objectMapper.writeValueAsString(chunk));
|
||||
chunkJson = objectMapper.writeValueAsString(chunk);
|
||||
} catch (Exception ex) {
|
||||
throw new BusinessException("暂存删除品牌结果分片失败");
|
||||
}
|
||||
Object existing = stringRedisTemplate.opsForHash().get(resultKey, field);
|
||||
if (chunkJson.equals(existing)) {
|
||||
stringRedisTemplate.expire(resultKey, Duration.ofHours(PAYLOAD_TTL_HOURS));
|
||||
return false;
|
||||
}
|
||||
stringRedisTemplate.opsForHash().put(resultKey, field, chunkJson);
|
||||
stringRedisTemplate.expire(resultKey, Duration.ofHours(PAYLOAD_TTL_HOURS));
|
||||
return true;
|
||||
}
|
||||
|
||||
public java.util.Map<String, java.util.List<String>> groupResultChunkJsonByFile(Long taskId) {
|
||||
@@ -88,11 +177,18 @@ public class DeleteBrandTaskCacheService {
|
||||
}
|
||||
|
||||
public void delete(Long taskId) {
|
||||
parsedPayloadLocalCache.remove(taskId);
|
||||
stringRedisTemplate.delete(buildPayloadKey(taskId));
|
||||
stringRedisTemplate.delete(buildProgressKey(taskId));
|
||||
stringRedisTemplate.delete(buildResultChunksKey(taskId));
|
||||
}
|
||||
|
||||
private record LocalParsedPayloadCacheEntry(long cachedAtMillis, Object value) {
|
||||
private boolean isFresh(long now) {
|
||||
return now - cachedAtMillis <= LOCAL_PARSED_PAYLOAD_CACHE_MILLIS;
|
||||
}
|
||||
}
|
||||
|
||||
private String buildPayloadKey(Long taskId) {
|
||||
return "delete-brand:task:parsed-payload:" + taskId;
|
||||
}
|
||||
|
||||
@@ -15,6 +15,11 @@ spring:
|
||||
url: ${AIIMAGE_DB_URL:jdbc:mysql://127.0.0.1:3306/aiimage?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false}
|
||||
username: ${AIIMAGE_DB_USERNAME:root}
|
||||
password: ${AIIMAGE_DB_PASSWORD:change-me}
|
||||
hikari:
|
||||
maximum-pool-size: ${AIIMAGE_DB_POOL_MAX_SIZE:30}
|
||||
minimum-idle: ${AIIMAGE_DB_POOL_MIN_IDLE:5}
|
||||
connection-timeout: ${AIIMAGE_DB_POOL_CONNECTION_TIMEOUT_MS:30000}
|
||||
leak-detection-threshold: ${AIIMAGE_DB_POOL_LEAK_DETECT_MS:0}
|
||||
data:
|
||||
redis:
|
||||
host: ${AIIMAGE_REDIS_HOST:127.0.0.1}
|
||||
@@ -72,6 +77,10 @@ aiimage:
|
||||
failed-ttl-hours: ${AIIMAGE_BRAND_PROGRESS_FAILED_TTL_HOURS:2}
|
||||
heartbeat-timeout-minutes: ${AIIMAGE_BRAND_PROGRESS_HEARTBEAT_TIMEOUT_MINUTES:15}
|
||||
stale-check-cron: ${AIIMAGE_BRAND_PROGRESS_STALE_CHECK_CRON:0 */2 * * * *}
|
||||
delete-brand-progress:
|
||||
heartbeat-timeout-minutes: ${AIIMAGE_DELETE_BRAND_HEARTBEAT_TIMEOUT_MINUTES:15}
|
||||
stale-check-cron: ${AIIMAGE_DELETE_BRAND_STALE_CHECK_CRON:0 */2 * * * *}
|
||||
finalize-check-cron: ${AIIMAGE_DELETE_BRAND_FINALIZE_CHECK_CRON:30 */2 * * * *}
|
||||
module-cleanup:
|
||||
enabled: ${AIIMAGE_MODULE_CLEANUP_ENABLED:true}
|
||||
cron: ${AIIMAGE_MODULE_CLEANUP_CRON:0 0 0 * * *}
|
||||
|
||||
Reference in New Issue
Block a user