From e7828dc926630b289a8309f6ec1c758e6b0b070b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sun, 30 Aug 2026 16:55:37 +0800 Subject: [PATCH] =?UTF-8?q?task-55:=20=E9=87=87=E9=9B=86=E8=BF=9B=E5=BA=A6?= =?UTF-8?q?=E5=BF=83=E8=B7=B3=E6=94=B9=E4=B8=BA=E8=8A=82=E6=B5=81/?= =?UTF-8?q?=E5=90=88=E5=B9=B6=E5=86=99=EF=BC=8C=E7=AA=97=E5=8F=A3=E5=86=85?= =?UTF-8?q?=E5=BF=83=E8=B7=B3=E5=8F=AA=E6=9B=B4=E6=96=B0=E5=86=85=E5=AD=98?= =?UTF-8?q?=E6=80=81=E4=B8=8D=E8=90=BD=E5=BA=93=EF=BC=8C=E7=AA=97=E5=8F=A3?= =?UTF-8?q?=E5=88=B0=E6=9C=9F=E6=88=96=E4=BB=BB=E5=8A=A1=E7=BB=93=E6=9D=9F?= =?UTF-8?q?=E8=B7=AF=E5=BE=84=E5=BC=BA=E5=88=B6=E6=8C=81=E4=B9=85=E5=8C=96?= =?UTF-8?q?=EF=BC=8C=E9=81=BF=E5=85=8D=E9=AB=98=E9=A2=91=20task=20UPDATE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/CollectDataService.java | 34 ++ .../CollectDataProgressThrottleTest.java | 301 ++++++++++++++++++ 2 files changed, 335 insertions(+) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataProgressThrottleTest.java 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 0cd1d586..8340b81e 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 @@ -165,6 +165,17 @@ public class CollectDataService { @Value("${aiimage.collect-data.max-chunk-rows:0}") private Integer maxChunkRows; + /** 进度统计节流窗口（毫秒）：窗口内 changed 心跳合并写，避免高频 task UPDATE。0 关闭节流。 */ + @Value("${aiimage.collect-data.progress-throttle-millis:15000}") + private long progressThrottleMillis; + + /** 进度统计脏数据兜底窗口（毫秒）：窗口到期后即使心跳未变化也强制刷新一次。0 关闭。 */ + @Value("${aiimage.collect-data.progress-dirty-window-millis:30000}") + private long progressDirtyWindowMillis; + + /** 上次进度统计实际落库时间戳（任务级，单机内存态；无锁竞争风险由任务分布式锁兜底）。 */ + private long lastProgressFlushMillis; + /** 采集源文件大小上限。0/负值回退默认 50MB，防止解析无界增长。 */ private long resolveMaxSourceFileBytes() { Long configured = maxSourceFileBytes; @@ -425,14 +436,37 @@ public class CollectDataService { changed = true; } if (!changed) { + // 内容未变化：仅当脏数据兜底窗口到期时才强制刷新一次， + // 否则零 UPDATE（重复心跳幂等）。 + if (!shouldForceProgressFlush()) { + return; + } + } else if (shouldThrottleProgressFlush()) { + // 节流窗口内：合并写（只更新内存态、不落库），窗口到期后统一持久化。 return; } persistStats(task, stats); task.setUpdatedAt(LocalDateTime.now()); fileTaskMapper.updateById(task); + lastProgressFlushMillis = System.currentTimeMillis(); } } + /** + * 节流判定：progressThrottleMillis>0 且距上次实际落库未超过窗口 → 合并写（跳过 UPDATE）。 + * progressThrottleMillis<=0 视为关闭节流，恒返回 false（每次心跳都落库，兼容旧行为）。 + */ + private boolean shouldThrottleProgressFlush() { + return progressThrottleMillis > 0 + && System.currentTimeMillis() - lastProgressFlushMillis < progressThrottleMillis; + } + + /** 脏数据兜底判定：progressDirtyWindowMillis>0 且距上次实际落库超过窗口 → 强制刷新一次。 */ + private boolean shouldForceProgressFlush() { + return progressDirtyWindowMillis > 0 + && System.currentTimeMillis() - lastProgressFlushMillis >= progressDirtyWindowMillis; + } + @Scheduled(cron = "${aiimage.collect-data.stale-check-cron:*/30 * * * * *}") public void finalizeStaleTasks() { long timeoutMinutes = Math.max(1L, staleTimeoutMinutes); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataProgressThrottleTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataProgressThrottleTest.java new file mode 100644 index 00000000..7014ddd1 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataProgressThrottleTest.java @@ -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); + } +}