super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(调度): 双节点下 collect-data/publish/image-video 定时调度加 Redis 分布式锁互斥——双活原来各自扫描执行同一批任务，重复派发/重复终态判定

Browse Source
This commit is contained in:
黄自达
2026-09-11 16:09:46 +08:00
parent 8dc03df95b
commit c2915036d6
3 changed files with 155 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java
+83 -63
View File
@@ -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<FileTaskEntity> tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.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<FileTaskEntity> tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.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<TaskChunkEntity> chunks, List<TaskResultItemEntity> items) {
if (chunks != null) {
Set<String> 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<TaskResultItemEntity> items) {
if (items == null) {
return;
}
Set<String> 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<CollectDataSummaryRowDto> summaries = new ArrayList<>();
}
@Transactional
// 不用 @Transactional（同类自调用代理失效）：用 TransactionTemplate 显式包 DB 段，
// 事务提交后再做远程删除（慢 IO），避免大任务删除时长事务占用连接池。
public void deleteTask(Long taskId, Long userId) {
List<String> chunkPayloadPointers = new ArrayList<>();
List<String> resultPayloadJsons = new ArrayList<>();
transactionTemplate.executeWithoutResult(status ->
deleteTaskRows(taskId, userId, chunkPayloadPointers, resultPayloadJsons));
// 事务已提交，这里执行远程删除
Set<String> 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<String> chunkPayloadPointers, List<String> resultPayloadJsons) {
FileTaskEntity task = requireTask(taskId, userId);
taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
.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<TaskResultItemEntity>()
.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<CollectDataItemEntity>()
.eq(CollectDataItemEntity::getTaskId, task.getId()));
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
@@ -1271,37 +1281,47 @@ public class CollectDataService {
taskResultItemMapper.delete(new LambdaQueryWrapper<TaskResultItemEntity>()
.eq(TaskResultItemEntity::getTaskId, task.getId())
.in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE));
deleteTransientTaskPayloads(
taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
.select(TaskChunkEntity::getPayloadJson)
.eq(TaskChunkEntity::getTaskId, task.getId())
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)),
taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
.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<String> payloadJsons, Set<String> 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<TaskResultItemEntity> resultItems = taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
.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<TaskResultItemEntity>()
.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<String> resultPayloadJsons = transactionTemplate.execute(status -> {
List<TaskResultItemEntity> resultItems = taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
.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<TaskResultItemEntity>()
.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) {
backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskService.java
+36 -17
View File
@@ -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<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>()
.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<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>()
.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<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>()
.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<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>()
.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)
backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java
+36 -24
View File
@@ -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<FileTaskEntity> staleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.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<FileTaskEntity> staleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.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());
}
}
}
}