diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java index 4f8e7086..bd177057 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java @@ -58,6 +58,7 @@ import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; import com.nanri.aiimage.modules.task.service.TaskFileJobService; import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; import lombok.RequiredArgsConstructor; +import com.nanri.aiimage.common.service.DistributedJobLockService; import lombok.extern.slf4j.Slf4j; import org.apache.poi.ss.usermodel.Cell; import org.apache.poi.ss.usermodel.DataFormatter; @@ -147,6 +148,7 @@ public class CollectDataService { private final OssStorageService ossStorageService; private final ObjectMapper objectMapper; private final TransactionTemplate transactionTemplate; + private final DistributedJobLockService distributedJobLockService; /** ASIN 去重 + 无效品牌批量集合查询器：两段式查询合并为一次往返，语义与旧实现等价。 */ private final CollectDataBatchQuery collectDataBatchQuery; @@ -483,16 +485,25 @@ public class CollectDataService { @Scheduled(cron = "${aiimage.collect-data.stale-check-cron:*/30 * * * * *}") public void finalizeStaleTasks() { - long timeoutMinutes = Math.max(1L, staleTimeoutMinutes); - LocalDateTime threshold = LocalDateTime.now().minusMinutes(timeoutMinutes); - List tasks = fileTaskMapper.selectList(new LambdaQueryWrapper() - .eq(FileTaskEntity::getModuleType, MODULE_TYPE) - .eq(FileTaskEntity::getStatus, STATUS_RUNNING) - .lt(FileTaskEntity::getUpdatedAt, threshold) - .orderByAsc(FileTaskEntity::getUpdatedAt) - .last("limit 200")); - for (FileTaskEntity task : tasks) { - finalizeStaleTask(task.getId(), threshold, timeoutMinutes); + // 双节点互斥：Redis 不可用时 fail-open（仅记日志），兜底靠下方 per-task DB 锁，不会双写终态 + DistributedJobLockService.LockHandle jobLock = + distributedJobLockService.tryLock("collect-data:stale-check", java.time.Duration.ofMinutes(2)); + if (jobLock == null) { + log.info("[collect-data] stale-check skipped, another instance holds the distributed lock"); + return; + } + try (jobLock) { + long timeoutMinutes = Math.max(1L, staleTimeoutMinutes); + LocalDateTime threshold = LocalDateTime.now().minusMinutes(timeoutMinutes); + List tasks = fileTaskMapper.selectList(new LambdaQueryWrapper() + .eq(FileTaskEntity::getModuleType, MODULE_TYPE) + .eq(FileTaskEntity::getStatus, STATUS_RUNNING) + .lt(FileTaskEntity::getUpdatedAt, threshold) + .orderByAsc(FileTaskEntity::getUpdatedAt) + .last("limit 200")); + for (FileTaskEntity task : tasks) { + finalizeStaleTask(task.getId(), threshold, timeoutMinutes); + } } } @@ -1201,36 +1212,6 @@ public class CollectDataService { return safe.isBlank() ? "collect-data" : safe; } - private void deleteTransientTaskPayloads(List chunks, List items) { - if (chunks != null) { - Set deleted = new HashSet<>(); - for (TaskChunkEntity chunk : chunks) { - // 多 chunk 共享同一对象（deterministic key 残留场景）按值去重只删一次。 - if (chunk.getPayloadJson() != null && deleted.add(chunk.getPayloadJson())) { - transientPayloadStorageService.deletePayloadIfPresent(chunk.getPayloadJson()); - } - } - } - deleteResultItemPayloads(items); - } - - /** 删除结果明细 payload：chunk 级引用按对象去重后各删一次，旧格式逐行删。 */ - private void deleteResultItemPayloads(List items) { - if (items == null) { - return; - } - Set deletedPointers = new HashSet<>(); - for (TaskResultItemEntity item : items) { - CollectDataResultDetailCodec.ChunkRef ref = resultDetailCodec.parseRef(item.getPayloadJson()); - if (ref != null) { - if (deletedPointers.add(ref.pointer())) { - transientPayloadStorageService.deletePayloadIfPresent(ref.pointer()); - } - } else { - transientPayloadStorageService.deletePayloadIfPresent(item.getPayloadJson()); - } - } - } private static class CollectDataStats { private int totalRows; @@ -1252,9 +1233,38 @@ public class CollectDataService { private List summaries = new ArrayList<>(); } - @Transactional + // 不用 @Transactional（同类自调用代理失效）：用 TransactionTemplate 显式包 DB 段， + // 事务提交后再做远程删除（慢 IO），避免大任务删除时长事务占用连接池。 public void deleteTask(Long taskId, Long userId) { + List chunkPayloadPointers = new ArrayList<>(); + List resultPayloadJsons = new ArrayList<>(); + transactionTemplate.executeWithoutResult(status -> + deleteTaskRows(taskId, userId, chunkPayloadPointers, resultPayloadJsons)); + // 事务已提交，这里执行远程删除 + Set deleted = new HashSet<>(); + for (String pointer : chunkPayloadPointers) { + if (deleted.add(pointer)) { + transientPayloadStorageService.deletePayloadIfPresent(pointer); + } + } + deletePayloadJsonsAndPointers(resultPayloadJsons, deleted); + } + + private void deleteTaskRows(Long taskId, Long userId, List chunkPayloadPointers, List resultPayloadJsons) { FileTaskEntity task = requireTask(taskId, userId); + taskChunkMapper.selectList(new LambdaQueryWrapper() + .select(TaskChunkEntity::getPayloadJson) + .eq(TaskChunkEntity::getTaskId, task.getId()) + .eq(TaskChunkEntity::getModuleType, MODULE_TYPE)).stream() + .map(TaskChunkEntity::getPayloadJson) + .filter(Objects::nonNull) + .forEach(chunkPayloadPointers::add); + taskResultItemMapper.selectList(new LambdaQueryWrapper() + .select(TaskResultItemEntity::getPayloadJson) + .eq(TaskResultItemEntity::getTaskId, task.getId()) + .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)).stream() + .map(TaskResultItemEntity::getPayloadJson) + .forEach(resultPayloadJsons::add); collectDataItemMapper.delete(new LambdaQueryWrapper() .eq(CollectDataItemEntity::getTaskId, task.getId())); fileResultMapper.delete(new LambdaQueryWrapper() @@ -1271,37 +1281,47 @@ public class CollectDataService { taskResultItemMapper.delete(new LambdaQueryWrapper() .eq(TaskResultItemEntity::getTaskId, task.getId()) .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)); - deleteTransientTaskPayloads( - taskChunkMapper.selectList(new LambdaQueryWrapper() - .select(TaskChunkEntity::getPayloadJson) - .eq(TaskChunkEntity::getTaskId, task.getId()) - .eq(TaskChunkEntity::getModuleType, MODULE_TYPE)), - taskResultItemMapper.selectList(new LambdaQueryWrapper() - .select(TaskResultItemEntity::getPayloadJson) - .eq(TaskResultItemEntity::getTaskId, task.getId()) - .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE))); taskFileJobService.deleteTaskJobs(task.getId(), MODULE_TYPE); fileTaskMapper.deleteById(task.getId()); } + private void deletePayloadJsonsAndPointers(List payloadJsons, Set deletedPointers) { + for (String payloadJson : payloadJsons) { + if (payloadJson == null) { + continue; + } + CollectDataResultDetailCodec.ChunkRef ref = resultDetailCodec.parseRef(payloadJson); + if (ref != null) { + if (deletedPointers.add(ref.pointer())) { + transientPayloadStorageService.deletePayloadIfPresent(ref.pointer()); + } + } else { + transientPayloadStorageService.deletePayloadIfPresent(payloadJson); + } + } + } + public void deleteHistory(Long resultId, Long userId) { FileResultEntity row = fileResultMapper.selectById(resultId); if (row == null || !MODULE_TYPE.equals(row.getModuleType()) || !Objects.equals(userId, row.getUserId())) { throw new BusinessException("记录不存在"); } - List resultItems = taskResultItemMapper.selectList(new LambdaQueryWrapper() - .select(TaskResultItemEntity::getPayloadJson) - .eq(TaskResultItemEntity::getTaskId, row.getTaskId()) - .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE) - .eq(TaskResultItemEntity::getResultId, row.getId())); - taskResultItemMapper.delete(new LambdaQueryWrapper() - .eq(TaskResultItemEntity::getTaskId, row.getTaskId()) - .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE) - .eq(TaskResultItemEntity::getResultId, row.getId())); - // 与 deleteTask 一致：先删 DB 行再物理删对象，保证行删除与对象删除一致。 - deleteResultItemPayloads(resultItems); - taskFileJobService.deleteResultJobs(row.getTaskId(), MODULE_TYPE, row.getId()); - fileResultMapper.deleteById(resultId); + List resultPayloadJsons = transactionTemplate.execute(status -> { + List resultItems = taskResultItemMapper.selectList(new LambdaQueryWrapper() + .select(TaskResultItemEntity::getPayloadJson) + .eq(TaskResultItemEntity::getTaskId, row.getTaskId()) + .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE) + .eq(TaskResultItemEntity::getResultId, row.getId())); + taskResultItemMapper.delete(new LambdaQueryWrapper() + .eq(TaskResultItemEntity::getTaskId, row.getTaskId()) + .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE) + .eq(TaskResultItemEntity::getResultId, row.getId())); + taskFileJobService.deleteResultJobs(row.getTaskId(), MODULE_TYPE, row.getId()); + fileResultMapper.deleteById(resultId); + return resultItems.stream().map(TaskResultItemEntity::getPayloadJson).toList(); + }); + // 事务已提交，再物理删对象，保证行删除与对象删除一致。 + deletePayloadJsonsAndPointers(resultPayloadJsons, new HashSet<>()); } public CollectDataCountryPreferenceVo getCountryPreference(Long userId) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskService.java index 7ce5147a..be6c04c8 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskService.java @@ -16,6 +16,7 @@ import com.nanri.aiimage.modules.imagevideo.model.dto.ImageVideoWorkflowResultRe import com.nanri.aiimage.modules.imagevideo.model.dto.ImageVideoWorkflowRunRequest; import com.nanri.aiimage.modules.imagevideo.model.entity.ImageVideoAsyncTaskEntity; import com.nanri.aiimage.modules.imagevideo.model.vo.ImageVideoAsyncTaskVo; +import com.nanri.aiimage.common.service.DistributedJobLockService; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.context.event.ApplicationReadyEvent; import org.springframework.beans.factory.annotation.Qualifier; @@ -66,6 +67,7 @@ public class ImageVideoAsyncTaskService { private final ObjectMapper objectMapper; private final TaskExecutor taskQueueExecutor; private final InstanceMetadata instanceMetadata; + private final com.nanri.aiimage.common.service.DistributedJobLockService distributedJobLockService; public ImageVideoAsyncTaskService( ImageVideoAsyncTaskMapper taskMapper, @@ -74,7 +76,8 @@ public class ImageVideoAsyncTaskService { ImageVideoArchiveService archiveService, ObjectMapper objectMapper, @Qualifier("taskQueueExecutor") TaskExecutor taskQueueExecutor, - InstanceMetadata instanceMetadata) { + InstanceMetadata instanceMetadata, + com.nanri.aiimage.common.service.DistributedJobLockService distributedJobLockService) { this.taskMapper = taskMapper; this.cozeService = cozeService; this.workflowConfigService = workflowConfigService; @@ -82,6 +85,7 @@ public class ImageVideoAsyncTaskService { this.objectMapper = objectMapper; this.taskQueueExecutor = taskQueueExecutor; this.instanceMetadata = instanceMetadata; + this.distributedJobLockService = distributedJobLockService; } public ImageVideoAsyncTaskVo submitDouyinCopy(DouyinCopyRequest request) { @@ -130,26 +134,41 @@ public class ImageVideoAsyncTaskService { @Scheduled(fixedDelayString = "${aiimage.image-video.async-task-dispatch-delay-ms:1000}") public void dispatchPendingTasks() { - List tasks = taskMapper.selectList(new LambdaQueryWrapper() - .eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.PENDING.name()) - .and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId()) - .or().isNull(ImageVideoAsyncTaskEntity::getOwnerInstanceId) - .or().eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, "")) - .orderByAsc(ImageVideoAsyncTaskEntity::getId) - .last("LIMIT " + DISPATCH_BATCH_SIZE)); - tasks.forEach(task -> taskQueueExecutor.execute(() -> executeTask(task.getId()))); + // 双节点互斥：只影响扫描入队动作（CAS claim 已兜底不重复执行），省掉另一节点的每秒空扫 + DistributedJobLockService.LockHandle jobLock = + distributedJobLockService.tryLock("image-video:dispatch", java.time.Duration.ofSeconds(5)); + if (jobLock == null) { + return; + } + try (jobLock) { + List tasks = taskMapper.selectList(new LambdaQueryWrapper() + .eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.PENDING.name()) + .and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId()) + .or().isNull(ImageVideoAsyncTaskEntity::getOwnerInstanceId) + .or().eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, "")) + .orderByAsc(ImageVideoAsyncTaskEntity::getId) + .last("LIMIT " + DISPATCH_BATCH_SIZE)); + tasks.forEach(task -> taskQueueExecutor.execute(() -> executeTask(task.getId()))); + } } @Scheduled(fixedDelayString = "${aiimage.image-video.async-task-poll-delay-ms:5000}") public void pollWaitingTasks() { - List tasks = taskMapper.selectList(new LambdaQueryWrapper() - .eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.WAITING.name()) - .and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId()) - .or().isNull(ImageVideoAsyncTaskEntity::getOwnerInstanceId) - .or().eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, "")) - .orderByAsc(ImageVideoAsyncTaskEntity::getUpdatedAt) - .last("LIMIT " + POLL_BATCH_SIZE)); - tasks.forEach(task -> taskQueueExecutor.execute(() -> pollTask(task.getId()))); + DistributedJobLockService.LockHandle jobLock = + distributedJobLockService.tryLock("image-video:poll", java.time.Duration.ofSeconds(30)); + if (jobLock == null) { + return; + } + try (jobLock) { + List tasks = taskMapper.selectList(new LambdaQueryWrapper() + .eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.WAITING.name()) + .and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId()) + .or().isNull(ImageVideoAsyncTaskEntity::getOwnerInstanceId) + .or().eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, "")) + .orderByAsc(ImageVideoAsyncTaskEntity::getUpdatedAt) + .last("LIMIT " + POLL_BATCH_SIZE)); + tasks.forEach(task -> taskQueueExecutor.execute(() -> pollTask(task.getId()))); + } } @EventListener(ApplicationReadyEvent.class) diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java index b33ef376..e819e3b9 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java @@ -50,6 +50,7 @@ import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopIndexService; import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import com.nanri.aiimage.common.service.DistributedJobLockService; import org.springframework.beans.factory.annotation.Value; import org.springframework.dao.DuplicateKeyException; import org.springframework.scheduling.annotation.Scheduled; @@ -108,6 +109,7 @@ public class PublishTaskService { private final OssStorageService ossStorageService; private final ObjectMapper objectMapper; private final TransactionTemplate transactionTemplate; + private final DistributedJobLockService distributedJobLockService; private final InstanceMetadata instanceMetadata; private final TaskProgressLightAssembler taskProgressLightAssembler; @@ -420,32 +422,42 @@ public class PublishTaskService { @Scheduled(fixedDelayString = "${aiimage.publish.stale-scan-delay-ms:60000}") public void failStaleTasks() { - LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(5, staleTimeoutMinutes)); - List staleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() - .eq(FileTaskEntity::getModuleType, MODULE_TYPE) - .eq(FileTaskEntity::getStatus, STATUS_RUNNING) - .and(owner -> owner - .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) IS NULL") - .or() - .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = ''") - .or() - .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = {0}", currentInstanceId())) - .lt(FileTaskEntity::getUpdatedAt, threshold) - .orderByAsc(FileTaskEntity::getUpdatedAt) - .last("limit 100")); - for (FileTaskEntity candidate : staleTasks) { - if (taskFileJobService.countUnfinishedAssembleJobs(candidate.getId(), MODULE_TYPE) > 0L) { - continue; - } - try (TaskDistributedLockService.LockHandle lock = - taskDistributedLockService.acquire(MODULE_TYPE, candidate.getId(), 0L)) { - if (lock == null) { + // 双节点互斥：避免 ownerInstanceId 为空的 stale 行被两节点同时扫出双写错误信息； + // Redis 不可用时 fail-open，兜底靠下方 per-task 锁 + DistributedJobLockService.LockHandle jobLock = + distributedJobLockService.tryLock("publish:stale-check", java.time.Duration.ofMinutes(2)); + if (jobLock == null) { + log.info("[publish] stale-check skipped, another instance holds the distributed lock"); + return; + } + try (jobLock) { + LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(5, staleTimeoutMinutes)); + List staleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() + .eq(FileTaskEntity::getModuleType, MODULE_TYPE) + .eq(FileTaskEntity::getStatus, STATUS_RUNNING) + .and(owner -> owner + .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) IS NULL") + .or() + .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = ''") + .or() + .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = {0}", currentInstanceId())) + .lt(FileTaskEntity::getUpdatedAt, threshold) + .orderByAsc(FileTaskEntity::getUpdatedAt) + .last("limit 100")); + for (FileTaskEntity candidate : staleTasks) { + if (taskFileJobService.countUnfinishedAssembleJobs(candidate.getId(), MODULE_TYPE) > 0L) { continue; } - transactionTemplate.executeWithoutResult(status -> failStaleTaskLocked(candidate.getId(), threshold)); - } catch (Exception ex) { - log.warn("[publish] stale task cleanup failed taskId={} msg={}", - candidate.getId(), ex.getMessage()); + try (TaskDistributedLockService.LockHandle lock = + taskDistributedLockService.acquire(MODULE_TYPE, candidate.getId(), 0L)) { + if (lock == null) { + continue; + } + transactionTemplate.executeWithoutResult(status -> failStaleTaskLocked(candidate.getId(), threshold)); + } catch (Exception ex) { + log.warn("[publish] stale task cleanup failed taskId={} msg={}", + candidate.getId(), ex.getMessage()); + } } } }