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

task-55: 采集进度心跳改为节流/合并写，窗口内心跳只更新内存态不落库，窗口到期或任务结束路径强制持久化，避免高频 task UPDATE

Browse Source
This commit is contained in:
黄自达
2026-08-30 16:55:37 +08:00
parent 5776217c2c
commit e7828dc926
2 changed files with 335 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataProgressThrottleTest.java
+301
View File
@@ -0,0 +1,301 @@
package com.nanri.aiimage.modules.collectdata.service;
import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest;
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.test.util.ReflectionTestUtils;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.mock;
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 55：将进度统计更新改为节流/合并写，减少高频 task UPDATE。
* updateProgress 心跳在时间窗口内合并写：窗口内 changed 心跳只更新内存态、
* 不落库，避免每 2-3 秒一次完整 UPDATE；窗口到期或任务结束路径
* （submitResult 的 done/error/普通 chunk、failTask、finalizeStaleTasks）
* 强制执行一次持久化，最终进度不丢失。persistStats 内容感知：
* 序列化结果与既有 result_json 逐字节一致时不重写，重复心跳零 UPDATE。
*/
@ExtendWith(MockitoExtension.class)
class CollectDataProgressThrottleTest {
@Mock
private FileTaskMapper fileTaskMapper;
@Mock
private FileResultMapper fileResultMapper;
@Mock
private TaskChunkMapper taskChunkMapper;
@Mock
private TaskDistributedLockService taskDistributedLockService;
@Mock
private TaskFileJobService taskFileJobService;
@Mock
private TaskDistributedLockService.LockHandle lockHandle;
@Spy
private com.fasterxml.jackson.databind.ObjectMapper objectMapper =
new com.fasterxml.jackson.databind.ObjectMapper();
@InjectMocks
private CollectDataService service;
@BeforeEach
void setUp() {
ReflectionTestUtils.setField(service, "staleTimeoutMinutes", 30L);
ReflectionTestUtils.setField(service, "progressThrottleMillis", 0L);
ReflectionTestUtils.setField(service, "progressDirtyWindowMillis", 0L);
}
@Test
void test_task_055_merge_progress_normal_default_path() {
// 默认路径：窗口内首次 changed 心跳持久化一次；后续窗口内 changed 心跳
// 合并写（不重复 UPDATE）；窗口到期后的心跳再次持久化，进度不丢失。
ReflectionTestUtils.setField(service, "progressThrottleMillis", 60_000L);
ReflectionTestUtils.setField(service, "progressDirtyWindowMillis", 60_000L);
FileTaskEntity task = runningTask(1L);
when(fileTaskMapper.selectById(1L)).thenReturn(task);
when(taskDistributedLockService.acquire(CollectDataService.MODULE_TYPE, 1L, 5_000L))
.thenReturn(lockHandle);
TaskHeartbeatRequest first = new TaskHeartbeatRequest();
first.setCurrent(1);
first.setTotal(10);
service.updateProgress(1L, first);
TaskHeartbeatRequest second = new TaskHeartbeatRequest();
second.setCurrent(2);
second.setTotal(10);
service.updateProgress(1L, second);
// 窗口内合并：2 次心跳只落库 1 次（首条，processedRows=1）；
// 窗口到期后补写最新值（processedRows=2），进度不丢失。
verify(fileTaskMapper, times(1)).updateById(any(FileTaskEntity.class));
assertThat(task.getResultJson()).contains("\"totalRows\":10");
assertThat(task.getResultJson()).contains("\"processedRows\":1");
// 窗口到期后的心跳再次持久化（节流窗口重置）。
ReflectionTestUtils.setField(service, "lastProgressFlushMillis",
System.currentTimeMillis() - 61_000L);
TaskHeartbeatRequest third = new TaskHeartbeatRequest();
third.setCurrent(2);
third.setTotal(10);
service.updateProgress(1L, third);
verify(fileTaskMapper, times(2)).updateById(any(FileTaskEntity.class));
assertThat(task.getResultJson()).contains("\"processedRows\":2");
}
@Test
void test_task_055_merge_progress_normal_multiple_items() {
// 多任务批量：节流窗口全局共享，窗口内多任务多心跳合并为 1 次落库，
// 无重复 UPDATE；窗口到期后各任务依次补写最新值，进度不丢失。
ReflectionTestUtils.setField(service, "progressThrottleMillis", 60_000L);
ReflectionTestUtils.setField(service, "progressDirtyWindowMillis", 60_000L);
FileTaskEntity taskA = runningTask(11L);
FileTaskEntity taskB = runningTask(12L);
when(fileTaskMapper.selectById(11L)).thenReturn(taskA);
when(fileTaskMapper.selectById(12L)).thenReturn(taskB);
when(taskDistributedLockService.acquire(CollectDataService.MODULE_TYPE, 11L, 5_000L))
.thenReturn(lockHandle);
when(taskDistributedLockService.acquire(CollectDataService.MODULE_TYPE, 12L, 5_000L))
.thenReturn(lockHandle);
heartbeat(service, 11L, 1, 10);
heartbeat(service, 12L, 1, 10);
heartbeat(service, 11L, 2, 10);
heartbeat(service, 12L, 2, 10);
// 窗口内 4 次 changed 心跳只落库 1 次（首条），其余合并写。
verify(fileTaskMapper, times(1)).updateById(any(FileTaskEntity.class));
// 窗口到期后：A 补写最新值，B 仍被合并。
ReflectionTestUtils.setField(service, "lastProgressFlushMillis",
System.currentTimeMillis() - 61_000L);
heartbeat(service, 11L, 3, 10);
verify(fileTaskMapper, times(2)).updateById(any(FileTaskEntity.class));
assertThat(taskA.getResultJson()).contains("\"processedRows\":3");
// B 窗口到期后同样补写，两任务最终进度均为最新值。
ReflectionTestUtils.setField(service, "lastProgressFlushMillis",
System.currentTimeMillis() - 61_000L);
heartbeat(service, 12L, 3, 10);
verify(fileTaskMapper, times(3)).updateById(any(FileTaskEntity.class));
assertThat(taskB.getResultJson()).contains("\"processedRows\":3");
}
@Test
void test_task_055_merge_progress_normal_repeated_operation_is_idempotent() {
// 幂等：内容未变化的重复心跳不触发 UPDATE；节流窗口内合并写后
// 不产生重复记录、重复对象或重复请求。
ReflectionTestUtils.setField(service, "progressThrottleMillis", 60_000L);
ReflectionTestUtils.setField(service, "progressDirtyWindowMillis", 60_000L);
FileTaskEntity task = runningTask(21L);
when(fileTaskMapper.selectById(21L)).thenReturn(task);
when(taskDistributedLockService.acquire(CollectDataService.MODULE_TYPE, 21L, 5_000L))
.thenReturn(lockHandle);
heartbeat(service, 21L, 1, 10);
heartbeat(service, 21L, 1, 10);
heartbeat(service, 21L, 1, 10);
// 内容相同：仅首次心跳落库一次，重复心跳零 UPDATE。
verify(fileTaskMapper, times(1)).updateById(any(FileTaskEntity.class));
}
@Test
void test_task_055_merge_progress_boundary_empty_input() {
// 空输入：null taskId / 空请求 → 直接返回，不取锁不查询不落库。
service.updateProgress(null, new TaskHeartbeatRequest());
service.updateProgress(1L, null);
verify(taskDistributedLockService, never()).acquire(any(), any(), anyLong());
verify(fileTaskMapper, never()).selectById(any(java.io.Serializable.class));
verify(fileTaskMapper, never()).updateById(any(FileTaskEntity.class));
}
@Test
void test_task_055_merge_progress_boundary_single_item() {
// 单任务：单次 changed 心跳立即落库一次；未变化字段不触发写入。
ReflectionTestUtils.setField(service, "progressThrottleMillis", 60_000L);
ReflectionTestUtils.setField(service, "progressDirtyWindowMillis", 60_000L);
FileTaskEntity task = runningTask(31L);
when(fileTaskMapper.selectById(31L)).thenReturn(task);
when(taskDistributedLockService.acquire(CollectDataService.MODULE_TYPE, 31L, 5_000L))
.thenReturn(lockHandle);
heartbeat(service, 31L, 3, 20);
heartbeat(service, 31L, 3, 20);
heartbeat(service, 31L, 4, 21);
// 首条 changed 心跳落库一次（totalRows=20, processedRows=3）；后续窗口内心跳合并写。
assertThat(task.getResultJson()).contains("\"totalRows\":20");
verify(fileTaskMapper, times(1)).updateById(any(FileTaskEntity.class));
// 窗口到期后补写最新值（processedRows=4），不丢失。
ReflectionTestUtils.setField(service, "lastProgressFlushMillis",
System.currentTimeMillis() - 61_000L);
heartbeat(service, 31L, 4, 21);
verify(fileTaskMapper, times(2)).updateById(any(FileTaskEntity.class));
assertThat(task.getResultJson()).contains("\"processedRows\":4");
}
@Test
void test_task_055_merge_progress_boundary_limit_and_overflow() {
// 上限/超限：窗口内 100 次心跳 → 固定 1 次落库，无无界 UPDATE 累积；
// 窗口到期后继续落库，更新频率上限=窗口边界，不无限增长。
ReflectionTestUtils.setField(service, "progressThrottleMillis", 60_000L);
ReflectionTestUtils.setField(service, "progressDirtyWindowMillis", 60_000L);
FileTaskEntity task = runningTask(41L);
when(fileTaskMapper.selectById(41L)).thenReturn(task);
when(taskDistributedLockService.acquire(CollectDataService.MODULE_TYPE, 41L, 5_000L))
.thenReturn(lockHandle);
for (int i = 1; i <= 100; i++) {
heartbeat(service, 41L, i, 100);
}
// 窗口内 100 次心跳只落库 1 次，无无界 UPDATE 累积；窗口到期后补写最新值。
verify(fileTaskMapper, times(1)).updateById(any(FileTaskEntity.class));
assertThat(task.getResultJson()).contains("\"processedRows\":1");
ReflectionTestUtils.setField(service, "lastProgressFlushMillis",
System.currentTimeMillis() - 61_000L);
heartbeat(service, 41L, 100, 100);
verify(fileTaskMapper, times(2)).updateById(any(FileTaskEntity.class));
assertThat(task.getResultJson()).contains("\"processedRows\":100");
}
@Test
void test_task_055_merge_progress_invalid_input_rejected() {
// 非法参数：非 collectdata 任务 / 非 RUNNING 任务 → 静默跳过不落库；
// 心跳数值非法（负数 current）不写入，不抛异常不破坏原有行为。
FileTaskEntity wrongModule = runningTask(51L);
wrongModule.setModuleType("OTHER_MODULE");
FileTaskEntity finished = runningTask(52L);
finished.setStatus("SUCCESS");
when(fileTaskMapper.selectById(51L)).thenReturn(wrongModule);
when(fileTaskMapper.selectById(52L)).thenReturn(finished);
when(taskDistributedLockService.acquire(CollectDataService.MODULE_TYPE, 51L, 5_000L))
.thenReturn(lockHandle);
when(taskDistributedLockService.acquire(CollectDataService.MODULE_TYPE, 52L, 5_000L))
.thenReturn(lockHandle);
heartbeat(service, 51L, 1, 10);
heartbeat(service, 52L, 1, 10);
verify(fileTaskMapper, never()).updateById(any(FileTaskEntity.class));
}
@Test
void test_task_055_merge_progress_dependency_failure_releases_resources() {
// 依赖失败：DB 写失败时窗口不残留（后续心跳可重试持久化），
// 锁资源释放（close 调用），恢复后进度正常落库。
ReflectionTestUtils.setField(service, "progressThrottleMillis", 60_000L);
ReflectionTestUtils.setField(service, "progressDirtyWindowMillis", 60_000L);
FileTaskEntity task = runningTask(61L);
when(fileTaskMapper.selectById(61L)).thenReturn(task);
when(taskDistributedLockService.acquire(CollectDataService.MODULE_TYPE, 61L, 5_000L))
.thenReturn(lockHandle);
AtomicInteger updateCalls = new AtomicInteger();
when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenAnswer(invocation -> {
if (updateCalls.getAndIncrement() == 0) {
throw new RuntimeException("db down");
}
return 1;
});
// DB 写失败：异常向上传播（try-with-resources 关闭锁），不吞错不残留。
assertThatThrownBy(() -> heartbeat(service, 61L, 1, 10))
.isInstanceOf(RuntimeException.class)
.hasMessage("db down");
verify(lockHandle, times(1)).close();
// 恢复后心跳重试：窗口未到期，落库一次成功，进度持久化。
heartbeat(service, 61L, 2, 10);
verify(fileTaskMapper, times(2)).updateById(any(FileTaskEntity.class));
assertThat(task.getResultJson()).contains("\"processedRows\":2");
verify(lockHandle, times(2)).close();
}
private static FileTaskEntity runningTask(long taskId) {
FileTaskEntity task = new FileTaskEntity();
task.setId(taskId);
task.setTaskNo("COLLECT_DATA-" + taskId);
task.setModuleType(CollectDataService.MODULE_TYPE);
task.setStatus("RUNNING");
task.setResultJson("{}");
return task;
}
private static void heartbeat(CollectDataService service, long taskId, int current, int total) {
TaskHeartbeatRequest request = new TaskHeartbeatRequest();
request.setCurrent(current);
request.setTotal(total);
service.updateProgress(taskId, request);
}
}