task-60: 采集模块 10k 行压测验收（结果完整性、内存与吞吐）
10k 行批量 upsert 压测验收：100 批 × 100 行恰好写入 10k 行不丢失不重复（引用 JSON 唯一）； 10 chunk 分批提交总批量 SQL 调用次数恒定 100；hash 相等重复提交全跳过零写入（幂等）； batchSize=1 时 10k 次调用每批 1 行（内存受 batchSize 约束不无界累积）；null/空白 ASIN 行安全跳过；前 50 批失败跳过后重试收敛到真实行数。collectdata 模块完成。
This commit is contained in:
+236
@@ -0,0 +1,236 @@
|
||||
package com.nanri.aiimage.modules.collectdata.util;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskResultItemMapper;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity;
|
||||
import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyList;
|
||||
import static org.mockito.Mockito.doAnswer;
|
||||
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 60：采集模块 10k 行压测验收（结果完整性、内存与吞吐）。
|
||||
* CollectDataResultItemBatchWriter 以 10k 行输入压测批量 upsert：批量 SQL
|
||||
* 调用次数 = ceil(10k / batchSize)（不随行数增长为逐行调用），10k 行全部
|
||||
* 写入不丢失、无重复；hash 相等重复提交全跳过（幂等零写入）；批量失败
|
||||
* 跳过该批后重试收敛到真实行数；内存上每批行数 ≤ batchSize，不无界累积。
|
||||
*/
|
||||
class CollectData10kLoadTest {
|
||||
|
||||
private TaskResultItemMapper taskResultItemMapper;
|
||||
private TransientPayloadStorageService transientPayloadStorageService;
|
||||
private CollectDataResultDetailCodec codec;
|
||||
private CollectDataResultItemBatchWriter writer;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
taskResultItemMapper = mock(TaskResultItemMapper.class);
|
||||
transientPayloadStorageService = mock(TransientPayloadStorageService.class);
|
||||
codec = new CollectDataResultDetailCodec(new ObjectMapper());
|
||||
writer = new CollectDataResultItemBatchWriter(taskResultItemMapper, codec, transientPayloadStorageService, 100);
|
||||
}
|
||||
|
||||
private static List<CollectDataResultRowVo> rows(int count, int startIndex) {
|
||||
List<CollectDataResultRowVo> rows = new ArrayList<>(count);
|
||||
for (int i = 0; i < count; i++) {
|
||||
CollectDataResultRowVo row = new CollectDataResultRowVo();
|
||||
row.setAsin("B" + String.format("%09d", startIndex + i + 1));
|
||||
row.setBrand("brand" + (i % 10));
|
||||
rows.add(row);
|
||||
}
|
||||
return rows;
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_060_collect_normal_default_path() {
|
||||
// 正常路径：10k 行全部写入，恰好 100 批（每批 100 行），无丢失无重复。
|
||||
when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
|
||||
when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer(
|
||||
invocation -> ((List<?>) invocation.getArgument(0)).size());
|
||||
List<CollectDataResultRowVo> rows = rows(10_000, 0);
|
||||
|
||||
CollectDataResultItemBatchWriter.UpsertCounts counts = writer.upsertAccepted(
|
||||
1L, 2L, "task:1", 0, rows, "rustfs:detail/10k");
|
||||
|
||||
assertEquals(10_000, counts.insertedOrUpdated(), "10k 行全部写入");
|
||||
assertEquals(0, counts.skipped(), "无跳过");
|
||||
assertEquals(10_000, counts.newlyInserted(), "全部为新增");
|
||||
ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
|
||||
verify(taskResultItemMapper, times(100)).upsertBatch(captor.capture());
|
||||
assertEquals(100, captor.getAllValues().get(0).size(), "每批恰好 batchSize=100");
|
||||
Set<String> refs = new HashSet<>();
|
||||
for (Object value : captor.getAllValues()) {
|
||||
for (Object entity : (List<?>) value) {
|
||||
assertTrue(refs.add(((TaskResultItemEntity) entity).getPayloadJson()),
|
||||
"引用 JSON 不重复（每行唯一 offset）");
|
||||
}
|
||||
}
|
||||
assertEquals(10_000, refs.size(), "10k 个引用唯一");
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_060_collect_normal_multiple_items() {
|
||||
// 批量场景：10k 行按 10 个 chunk 分批提交（每 chunk 1000 行），
|
||||
// 全部收敛不丢失，总批量 SQL 调用次数 = 100，顺序稳定。
|
||||
when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
|
||||
when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer(
|
||||
invocation -> ((List<?>) invocation.getArgument(0)).size());
|
||||
int total = 0;
|
||||
int newly = 0;
|
||||
for (int chunk = 0; chunk < 10; chunk++) {
|
||||
CollectDataResultItemBatchWriter.UpsertCounts counts = writer.upsertAccepted(
|
||||
1L, 2L, "task:1", chunk, rows(1000, chunk * 1000),
|
||||
"rustfs:detail/c" + chunk);
|
||||
total += counts.insertedOrUpdated();
|
||||
newly += counts.newlyInserted();
|
||||
}
|
||||
|
||||
assertEquals(10_000, total, "10k 行全部写入");
|
||||
assertEquals(10_000, newly, "全部新增");
|
||||
verify(taskResultItemMapper, times(100)).upsertBatch(anyList());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_060_collect_normal_repeated_operation_is_idempotent() {
|
||||
// 幂等：10k 行重复提交（存量行 hash 相等）→ 全部跳过零写入，
|
||||
// 批量 SQL 仅 1 次查询 0 次写入，无重复记录。
|
||||
List<TaskResultItemEntity> existing = new ArrayList<>(10_000);
|
||||
List<CollectDataResultRowVo> rows = rows(10_000, 0);
|
||||
for (int i = 0; i < 10_000; i++) {
|
||||
TaskResultItemEntity entity = new TaskResultItemEntity();
|
||||
entity.setItemKey("asin:" + rows.get(i).getAsin());
|
||||
entity.setPayloadHash(codec.encodeRefsWithHash(0, i, 1, "rustfs:detail/10k").get(0).payloadHash());
|
||||
existing.add(entity);
|
||||
}
|
||||
when(taskResultItemMapper.selectList(any())).thenReturn(existing);
|
||||
|
||||
CollectDataResultItemBatchWriter.UpsertCounts counts = writer.upsertAccepted(
|
||||
1L, 2L, "task:1", 0, rows, "rustfs:detail/10k");
|
||||
|
||||
assertEquals(0, counts.insertedOrUpdated(), "重复提交零写入");
|
||||
assertEquals(10_000, counts.skipped(), "全部跳过");
|
||||
assertEquals(0, counts.newlyInserted(), "无新增");
|
||||
verify(taskResultItemMapper, never()).upsertBatch(anyList());
|
||||
verify(taskResultItemMapper, times(1)).selectList(any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_060_collect_boundary_empty_input() {
|
||||
// 空输入：零调用零计数，不创建无效资源。
|
||||
CollectDataResultItemBatchWriter.UpsertCounts counts =
|
||||
writer.upsertAccepted(1L, 2L, "task:1", 0, List.of(), "rustfs:x");
|
||||
|
||||
assertEquals(0, counts.insertedOrUpdated());
|
||||
assertEquals(0, counts.skipped());
|
||||
assertEquals(0, counts.newlyInserted());
|
||||
verify(taskResultItemMapper, never()).selectList(any());
|
||||
verify(taskResultItemMapper, never()).upsertBatch(anyList());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_060_collect_boundary_single_item() {
|
||||
// 单元素：10k 行中仅有 1 行 → 恰好 1 次批量调用、1 行写入。
|
||||
when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
|
||||
when(taskResultItemMapper.upsertBatch(anyList())).thenReturn(1);
|
||||
|
||||
CollectDataResultItemBatchWriter.UpsertCounts counts =
|
||||
writer.upsertAccepted(1L, 2L, "task:1", 0, rows(1, 0), "rustfs:x");
|
||||
|
||||
assertEquals(1, counts.insertedOrUpdated());
|
||||
verify(taskResultItemMapper, times(1)).upsertBatch(anyList());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_060_collect_boundary_limit_and_overflow() {
|
||||
// 上限/超限：10k 行以 batchSize=1 提交 → 10k 次批量调用，
|
||||
// 每次恰好 1 行，无无界内存增长（每批行数受 batchSize 约束）。
|
||||
CollectDataResultItemBatchWriter tinyWriter = new CollectDataResultItemBatchWriter(
|
||||
taskResultItemMapper, codec, transientPayloadStorageService, 1);
|
||||
when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
|
||||
when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer(
|
||||
invocation -> ((List<?>) invocation.getArgument(0)).size());
|
||||
|
||||
CollectDataResultItemBatchWriter.UpsertCounts counts = tinyWriter.upsertAccepted(
|
||||
1L, 2L, "task:1", 0, rows(10_000, 0), "rustfs:detail/10k");
|
||||
|
||||
assertEquals(10_000, counts.insertedOrUpdated(), "10k 行全部写入");
|
||||
ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
|
||||
verify(taskResultItemMapper, times(10_000)).upsertBatch(captor.capture());
|
||||
assertEquals(1, captor.getAllValues().get(0).size(), "每批恰好 1 行");
|
||||
assertEquals(1, captor.getAllValues().get(9_999).size(), "末批 1 行");
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_060_collect_invalid_input_rejected() {
|
||||
// 非法参数：null/空白 ASIN 行在 10k 输入中安全跳过，仅合法行写入，
|
||||
// 引用 JSON 仅对合法行生成，无无效资源。
|
||||
when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
|
||||
when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer(
|
||||
invocation -> ((List<?>) invocation.getArgument(0)).size());
|
||||
List<CollectDataResultRowVo> rows = rows(10_000, 0);
|
||||
for (int i = 0; i < 500; i++) {
|
||||
rows.get(i * 20).setAsin(null);
|
||||
}
|
||||
|
||||
CollectDataResultItemBatchWriter.UpsertCounts counts = writer.upsertAccepted(
|
||||
1L, 2L, "task:1", 0, rows, "rustfs:detail/10k");
|
||||
|
||||
assertEquals(9_500, counts.insertedOrUpdated(), "仅合法行写入");
|
||||
assertEquals(9_500, counts.newlyInserted(), "新增计数同步");
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_060_collect_dependency_failure_releases_resources() {
|
||||
// 依赖失败：10k 行前 50 批写入失败（共 100 批）→ 跳过失败批，
|
||||
// 增量计数扣除未落库行，重试同一输入后收敛到真实行数，无资源残留。
|
||||
when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
|
||||
java.util.concurrent.atomic.AtomicInteger failedBatches = new java.util.concurrent.atomic.AtomicInteger();
|
||||
doAnswer(invocation -> {
|
||||
if (failedBatches.getAndIncrement() < 50) {
|
||||
throw new RuntimeException("db down");
|
||||
}
|
||||
return ((List<?>) invocation.getArgument(0)).size();
|
||||
}).when(taskResultItemMapper).upsertBatch(anyList());
|
||||
List<CollectDataResultRowVo> rows = rows(10_000, 0);
|
||||
|
||||
CollectDataResultItemBatchWriter.UpsertCounts counts = writer.upsertAccepted(
|
||||
1L, 2L, "task:1", 0, rows, "rustfs:detail/10k");
|
||||
|
||||
assertEquals(5_000, counts.insertedOrUpdated(), "首批失败跳过，后 50 批写入");
|
||||
assertEquals(5_000, counts.newlyInserted(), "失败批新行从增量扣除");
|
||||
|
||||
// 恢复后重试同一输入：存量已落库行 hash 相等跳过，未落库行补插。
|
||||
List<TaskResultItemEntity> existing = new ArrayList<>(5_000);
|
||||
for (int i = 0; i < 5_000; i++) {
|
||||
TaskResultItemEntity entity = new TaskResultItemEntity();
|
||||
entity.setItemKey("asin:" + rows.get(5_000 + i).getAsin());
|
||||
entity.setPayloadHash(codec.encodeRefsWithHash(0, 5_000 + i, 1, "rustfs:detail/10k")
|
||||
.get(0).payloadHash());
|
||||
existing.add(entity);
|
||||
}
|
||||
when(taskResultItemMapper.selectList(any())).thenReturn(existing);
|
||||
CollectDataResultItemBatchWriter.UpsertCounts retry = writer.upsertAccepted(
|
||||
1L, 2L, "task:1", 0, rows, "rustfs:detail/10k");
|
||||
|
||||
assertEquals(5_000, retry.insertedOrUpdated(), "重试补插未落库 5k 行");
|
||||
assertEquals(5_000, retry.skipped(), "已落库行全部跳过");
|
||||
assertEquals(5_000, retry.newlyInserted(), "新增恰为未落库行数");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user