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

task-63: 进度快照写入最小更新间隔节流（终态豁免+删除清理跟踪）

Browse Source
This commit is contained in:
黄自达
2026-08-30 18:28:15 +08:00
parent a9cee8e9a1
commit 9c9b8a6078
2 changed files with 239 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskProgressSnapshotService.java
+39
View File
@@ -7,19 +7,28 @@ import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.modules.task.mapper.TaskProgressSnapshotMapper;
import com.nanri.aiimage.modules.task.model.entity.TaskProgressSnapshotEntity;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@Service
@RequiredArgsConstructor
public class TaskProgressSnapshotService {
private static final Set<String> TERMINAL_STATUSES = Set.of("SUCCESS", "FAILED", "CANCELLED", "CANCELED");
private final TaskProgressSnapshotMapper taskProgressSnapshotMapper;
private final ObjectMapper objectMapper;
private final ConcurrentHashMap<String, Long> lastWriteAtMillis = new ConcurrentHashMap<>();
@Value("${aiimage.task-progress-snapshot.min-update-interval-ms:500}")
private long minUpdateIntervalMillis = 500;
@Transactional
public void save(Long taskId,
@@ -54,6 +63,7 @@ public class TaskProgressSnapshotService {
entity.setUpdatedAt(now);
try {
taskProgressSnapshotMapper.insert(entity);
touchLastWrite(taskId, moduleType, status);
return;
} catch (DuplicateKeyException ignored) {
existing = find(taskId, moduleType);
@@ -66,6 +76,9 @@ public class TaskProgressSnapshotService {
currentScopeKey, message, snapshotJson)) {
return;
}
if (shouldThrottle(taskId, moduleType, status)) {
return;
}
taskProgressSnapshotMapper.update(null, new LambdaUpdateWrapper<TaskProgressSnapshotEntity>()
.eq(TaskProgressSnapshotEntity::getId, existing.getId())
.set(TaskProgressSnapshotEntity::getStatus, status)
@@ -77,6 +90,31 @@ public class TaskProgressSnapshotService {
.set(TaskProgressSnapshotEntity::getMessage, message)
.set(TaskProgressSnapshotEntity::getSnapshotJson, snapshotJson)
.set(TaskProgressSnapshotEntity::getUpdatedAt, now));
touchLastWrite(taskId, moduleType, status);
}
private boolean shouldThrottle(Long taskId, String moduleType, String status) {
if (TERMINAL_STATUSES.contains(status)) {
return false;
}
long interval = Math.max(0L, minUpdateIntervalMillis);
if (interval <= 0L) {
return false;
}
Long lastWrite = lastWriteAtMillis.get(cacheKey(taskId, moduleType));
return lastWrite != null && System.currentTimeMillis() - lastWrite < interval;
}
private void touchLastWrite(Long taskId, String moduleType, String status) {
if (TERMINAL_STATUSES.contains(status)) {
lastWriteAtMillis.remove(cacheKey(taskId, moduleType));
return;
}
lastWriteAtMillis.put(cacheKey(taskId, moduleType), System.currentTimeMillis());
}
private static String cacheKey(Long taskId, String moduleType) {
return taskId + ":" + moduleType;
}
private boolean isUnchanged(TaskProgressSnapshotEntity existing,
@@ -116,6 +154,7 @@ public class TaskProgressSnapshotService {
taskProgressSnapshotMapper.delete(new LambdaQueryWrapper<TaskProgressSnapshotEntity>()
.eq(TaskProgressSnapshotEntity::getTaskId, taskId)
.eq(TaskProgressSnapshotEntity::getModuleType, moduleType));
lastWriteAtMillis.remove(cacheKey(taskId, moduleType));
}
private String writeJson(Object value) {
backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskProgressSnapshotServiceThrottleTest.java
+200
View File
@@ -0,0 +1,200 @@
package com.nanri.aiimage.modules.task.service;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.modules.task.mapper.TaskProgressSnapshotMapper;
import com.nanri.aiimage.modules.task.model.entity.TaskProgressSnapshotEntity;
import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.test.util.ReflectionTestUtils;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* Task 63：为前端/后端进度快照增加写入去重和最小更新间隔。
* TaskProgressSnapshotService 对进度快照写入做节流：内容未变化时跳过写库
* （已有去重语义保留）；内容变化但距上次写入不足最小更新间隔（默认 500ms）
* 时跳过中间态写入，终态（SUCCESS/FAILED/CANCELLED）不受节流保证落库；
* 删除快照时同步清理节流跟踪，不残留内存状态。
*/
@ExtendWith(MockitoExtension.class)
class TaskProgressSnapshotServiceThrottleTest {
@BeforeAll
static void initializeMybatisMetadata() {
TableInfoHelper.initTableInfo(
new MapperBuilderAssistant(new MybatisConfiguration(), ""),
TaskProgressSnapshotEntity.class);
}
@Mock private TaskProgressSnapshotMapper taskProgressSnapshotMapper;
@Mock private ObjectMapper objectMapper;
private TaskProgressSnapshotService service;
@BeforeEach
void setUp() {
service = new TaskProgressSnapshotService(taskProgressSnapshotMapper, objectMapper);
ReflectionTestUtils.setField(service, "minUpdateIntervalMillis", 1000L);
}
private TaskProgressSnapshotEntity running() {
TaskProgressSnapshotEntity entity = new TaskProgressSnapshotEntity();
entity.setId(1L);
entity.setTaskId(20581L);
entity.setModuleType("SIMILAR_ASIN");
entity.setStatus("RUNNING");
entity.setTotalCount(10);
entity.setSuccessCount(4);
entity.setFailedCount(1);
entity.setPendingCount(5);
entity.setCurrentScopeKey("scope-1");
entity.setMessage("processing");
return entity;
}
/** 把某任务的节流跟踪时间拨到距今 millisAgo 毫秒前，模拟时间流逝。 */
@SuppressWarnings("unchecked")
private void rewindLastWrite(Long taskId, String moduleType, long millisAgo) {
Map<String, Long> tracking = (Map<String, Long>) ReflectionTestUtils
.getField(service, "lastWriteAtMillis");
tracking.put(taskId + ":" + moduleType, System.currentTimeMillis() - millisAgo);
}
@Test
void test_task_063_progress_frontend_normal_default_path() {
// 正常路径：距上次写入超过最小间隔的变更落库，写入一次。
when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(running());
rewindLastWrite(20581L, "SIMILAR_ASIN", 5_000L);
service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 5, 1,
"scope-1", "processing", null);
verify(taskProgressSnapshotMapper, times(1)).update(isNull(), any(LambdaUpdateWrapper.class));
}
@Test
void test_task_063_progress_frontend_normal_multiple_items() {
// 批量场景：多个任务各自跟踪节流时间，互不影响，全部落库。
for (long taskId = 1; taskId <= 3; taskId++) {
when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class)))
.thenReturn(null);
service.save(taskId, "MODULE_" + taskId, "RUNNING", 10, 0, 0,
"scope-" + taskId, "start", null);
}
verify(taskProgressSnapshotMapper, times(3)).insert(any(TaskProgressSnapshotEntity.class));
}
@Test
void test_task_063_progress_frontend_normal_repeated_operation_is_idempotent() {
// 幂等：相同内容重复保存不产生写库（内容去重），结果一致无副作用。
when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(running());
service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 4, 1,
"scope-1", "processing", null);
service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 4, 1,
"scope-1", "processing", null);
verify(taskProgressSnapshotMapper, never()).update(any(), any());
verify(taskProgressSnapshotMapper, never()).insert(any(TaskProgressSnapshotEntity.class));
}
@Test
void test_task_063_progress_frontend_boundary_empty_input() {
// 空输入：null/空模块、空状态安全跳过，不发起任何写库调用。
service.save(null, "SIMILAR_ASIN", "RUNNING", 10, 0, 0, null, null, null);
service.save(20581L, "", "RUNNING", 10, 0, 0, null, null, null);
service.save(20581L, "SIMILAR_ASIN", " ", 10, 0, 0, null, null, null);
verify(taskProgressSnapshotMapper, never()).insert(any(TaskProgressSnapshotEntity.class));
verify(taskProgressSnapshotMapper, never()).update(any(), any());
verify(taskProgressSnapshotMapper, never()).selectOne(any());
}
@Test
void test_task_063_progress_frontend_boundary_single_item() {
// 单元素：首次写入落库；间隔内变更被节流跳过；间隔过后变更落库。
when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(null);
service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 0, 0, "s1", "go", null);
verify(taskProgressSnapshotMapper, times(1)).insert(any(TaskProgressSnapshotEntity.class));
when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(running());
service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 5, 1, "s2", "half", null);
verify(taskProgressSnapshotMapper, never()).update(isNull(), any(LambdaUpdateWrapper.class));
rewindLastWrite(20581L, "SIMILAR_ASIN", 5_000L);
service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 6, 1, "s3", "more", null);
verify(taskProgressSnapshotMapper, times(1)).update(isNull(), any(LambdaUpdateWrapper.class));
}
@Test
void test_task_063_progress_frontend_boundary_limit_and_overflow() {
// 上限/超限：最小更新间隔设为 0（超限值）→ 节流禁用，
// 每次变更都落库，不发生无界内存增长（跟踪 map 仅随任务数增长）。
ReflectionTestUtils.setField(service, "minUpdateIntervalMillis", 0L);
when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(running());
service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 5, 1, "s1", "m1", null);
service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 6, 1, "s1", "m2", null);
service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 7, 1, "s1", "m3", null);
verify(taskProgressSnapshotMapper, times(3)).update(isNull(), any(LambdaUpdateWrapper.class));
}
@Test
void test_task_063_progress_frontend_invalid_input_rejected() {
// 非法参数：非正任务 id / 空白状态拒绝写入，不创建无效快照。
service.save(0L, "SIMILAR_ASIN", "RUNNING", 10, 0, 0, null, null, null);
service.save(-5L, "SIMILAR_ASIN", "RUNNING", 10, 0, 0, null, null, null);
verify(taskProgressSnapshotMapper, never()).insert(any(TaskProgressSnapshotEntity.class));
verify(taskProgressSnapshotMapper, never()).update(any(), any());
}
@Test
void test_task_063_progress_frontend_dependency_failure_releases_resources() {
// 依赖失败：唯一键冲突回落查询后正常更新（并发安全）；
// 写库异常传播不残留；删除快照后节流跟踪释放，间隔内再次写入不再被节流。
when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class)))
.thenReturn(null)
.thenReturn(running());
doThrow(new DuplicateKeyException("dup")).when(taskProgressSnapshotMapper)
.insert(any(TaskProgressSnapshotEntity.class));
service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 5, 1, "s1", "m1", null);
verify(taskProgressSnapshotMapper, times(1)).update(isNull(), any(LambdaUpdateWrapper.class));
rewindLastWrite(20581L, "SIMILAR_ASIN", 5_000L);
doThrow(new RuntimeException("db down")).when(taskProgressSnapshotMapper)
.update(isNull(), any(LambdaUpdateWrapper.class));
assertThrows(RuntimeException.class, () -> service.save(20581L, "SIMILAR_ASIN",
"RUNNING", 10, 6, 1, "s2", "m2", null), "写库异常传播");
service.delete(20581L, "SIMILAR_ASIN");
verify(taskProgressSnapshotMapper, times(1))
.delete(any(LambdaQueryWrapper.class));
when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(running());
when(taskProgressSnapshotMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1);
service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 7, 1, "s3", "m3", null);
verify(taskProgressSnapshotMapper, times(3)).update(isNull(), any(LambdaUpdateWrapper.class));
}
}