task-30: 为国家结果行建立稳定去重键，替换线性重复扫描
This commit is contained in:
+36
-4
@@ -86,6 +86,8 @@ public class ShopDataCrawlTaskService {
|
||||
private static final String INTERRUPTED_MESSAGE = "Python 在该店铺结果提交完成前中断";
|
||||
private static final String PARTIAL_RESULT_MESSAGE = "Python 中断，已保留已回传的部分数据";
|
||||
private static final String RESULT_CHUNK_SCOPE_PREFIX = "result-chunks:";
|
||||
/** 国家结果行去重键字段分隔符（控制字符，字段值 trim 后不可能包含）。 */
|
||||
private static final String ROW_KEY_SEPARATOR = "";
|
||||
|
||||
private final FileTaskMapper fileTaskMapper;
|
||||
private final FileResultMapper fileResultMapper;
|
||||
@@ -1604,9 +1606,22 @@ public class ShopDataCrawlTaskService {
|
||||
map.put(item.getCountry(), item);
|
||||
continue;
|
||||
}
|
||||
List<ShopDataCrawlRowDto> merged = new ArrayList<>(existing.getItems() == null ? List.of() : existing.getItems());
|
||||
List<ShopDataCrawlRowDto> merged = existing.getItems() == null ? new ArrayList<>() : new ArrayList<>(existing.getItems());
|
||||
Set<String> seen = new HashSet<>();
|
||||
for (ShopDataCrawlRowDto old : merged) {
|
||||
String key = rowDedupKey(old);
|
||||
if (key != null) {
|
||||
seen.add(key);
|
||||
}
|
||||
}
|
||||
for (ShopDataCrawlRowDto row : item.getItems() == null ? List.<ShopDataCrawlRowDto>of() : item.getItems()) {
|
||||
if (row != null && merged.stream().noneMatch(old -> sameRow(old, row))) merged.add(copyRow(row));
|
||||
if (row == null) {
|
||||
continue;
|
||||
}
|
||||
String key = rowDedupKey(row);
|
||||
if (key != null && seen.add(key)) {
|
||||
merged.add(copyRow(row));
|
||||
}
|
||||
}
|
||||
existing.setItems(merged);
|
||||
}
|
||||
@@ -2366,8 +2381,13 @@ public class ShopDataCrawlTaskService {
|
||||
item.setCountry(normalizeCountry(source.getCountry()));
|
||||
List<ShopDataCrawlRowDto> rows = new ArrayList<>();
|
||||
if (source.getItems() != null) {
|
||||
Set<String> seen = new HashSet<>();
|
||||
for (ShopDataCrawlRowDto sourceRow : source.getItems()) {
|
||||
if (sourceRow != null && !rowEmpty(sourceRow) && rows.stream().noneMatch(old -> sameRow(old, sourceRow))) {
|
||||
if (sourceRow == null || rowEmpty(sourceRow)) {
|
||||
continue;
|
||||
}
|
||||
String key = rowDedupKey(sourceRow);
|
||||
if (key != null && seen.add(key)) {
|
||||
rows.add(copyRow(sourceRow));
|
||||
}
|
||||
}
|
||||
@@ -2455,13 +2475,25 @@ public class ShopDataCrawlTaskService {
|
||||
&& Objects.equals(trim(left.getRecommendedOffer()), trim(right.getRecommendedOffer()));
|
||||
}
|
||||
|
||||
/** 国家结果行稳定去重键：与 sameRow 的 10 字段 trim 比较语义等价，用于 O(1) 去重。 */
|
||||
static String rowDedupKey(ShopDataCrawlRowDto row) {
|
||||
if (row == null) {
|
||||
return null;
|
||||
}
|
||||
return trim(row.getDate()) + ROW_KEY_SEPARATOR + trim(row.getAsin()) + ROW_KEY_SEPARATOR + trim(row.getBrand())
|
||||
+ ROW_KEY_SEPARATOR + trim(row.getCommodityImage()) + ROW_KEY_SEPARATOR + trim(row.getInventorySales())
|
||||
+ ROW_KEY_SEPARATOR + trim(row.getSalesRank()) + ROW_KEY_SEPARATOR + trim(row.getPageViews())
|
||||
+ ROW_KEY_SEPARATOR + trim(row.getUnitsSold()) + ROW_KEY_SEPARATOR + trim(row.getPrice())
|
||||
+ ROW_KEY_SEPARATOR + trim(row.getRecommendedOffer());
|
||||
}
|
||||
|
||||
private boolean rowEmpty(ShopDataCrawlRowDto row) {
|
||||
return row == null || (blank(row.getDate()) && blank(row.getAsin()) && blank(row.getBrand()) && blank(row.getCommodityImage()) && blank(row.getInventorySales())
|
||||
&& blank(row.getSalesRank()) && blank(row.getPageViews()) && blank(row.getUnitsSold())
|
||||
&& blank(row.getPrice()) && blank(row.getRecommendedOffer()));
|
||||
}
|
||||
|
||||
private String trim(String value) {
|
||||
private static String trim(String value) {
|
||||
return value == null ? "" : value.trim();
|
||||
}
|
||||
|
||||
|
||||
+654
@@ -0,0 +1,654 @@
|
||||
package com.nanri.aiimage.modules.shopdatacrawl.service;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.nanri.aiimage.common.exception.BusinessException;
|
||||
import com.nanri.aiimage.config.InstanceMetadata;
|
||||
import com.nanri.aiimage.config.TaskPressureProperties;
|
||||
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlCountryResultDto;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlRowDto;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlShopPayloadDto;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlSubmitResultRequest;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlResultItemVo;
|
||||
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.mapper.TaskScopeStateMapper;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity;
|
||||
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
|
||||
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
|
||||
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
|
||||
import com.nanri.aiimage.modules.task.service.TaskResultItemService;
|
||||
import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
|
||||
import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService;
|
||||
import org.apache.ibatis.builder.MapperBuilderAssistant;
|
||||
import org.apache.ibatis.session.Configuration;
|
||||
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.Spy;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
import org.springframework.dao.DuplicateKeyException;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Comparator;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyLong;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.ArgumentMatchers.nullable;
|
||||
import static org.mockito.Mockito.lenient;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* Task 30：为国家结果行建立稳定去重键，替换线性重复扫描。
|
||||
* mergeCountryResults / copyCountryResults 原用 sameRow 的 O(n²) noneMatch 扫描去重；
|
||||
* 实现改为按 10 个 trim 后字段构造稳定去重键（date|asin|brand|commodityImage|inventorySales|
|
||||
* salesRank|pageViews|unitsSold|price|recommendedOffer），LinkedHashSet 一次遍历去重，
|
||||
* 保留首现顺序与语义等价性（去重结果与 sameRow 逐行比较完全一致）。
|
||||
*/
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class ShopDataCrawlRowDedupKeyTest {
|
||||
|
||||
private static final String MODULE_TYPE = "SHOP_DATA_CRAWL";
|
||||
private static final String SHOP_NAME = "Demo Shop";
|
||||
|
||||
@BeforeAll
|
||||
static void initializeMybatisMetadata() {
|
||||
MapperBuilderAssistant assistant = new MapperBuilderAssistant(new Configuration(), "");
|
||||
TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, FileResultEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class);
|
||||
}
|
||||
|
||||
@Mock private FileTaskMapper fileTaskMapper;
|
||||
@Mock private FileResultMapper fileResultMapper;
|
||||
@Mock private ShopDataCrawlResolveService shopDataCrawlResolveService;
|
||||
@Mock private ShopDataCrawlExcelAssemblyService excelAssemblyService;
|
||||
@Mock private ShopDataCrawlTaskCacheService taskCacheService;
|
||||
@Mock private OssStorageService ossStorageService;
|
||||
@Mock private ZiniaoShopSwitchService ziniaoShopSwitchService;
|
||||
@Mock private TaskPressureProperties taskPressureProperties;
|
||||
@Mock private TaskFileJobService taskFileJobService;
|
||||
@Mock private TaskResultItemService taskResultItemService;
|
||||
@Mock private TaskProgressSnapshotService taskProgressSnapshotService;
|
||||
@Mock private TaskDistributedLockService taskDistributedLockService;
|
||||
@Mock private TaskChunkMapper taskChunkMapper;
|
||||
@Mock private TaskScopeStateMapper taskScopeStateMapper;
|
||||
@Mock private TransientPayloadStorageService transientPayloadStorageService;
|
||||
@Mock private InstanceMetadata instanceMetadata;
|
||||
@Mock private ShopDataCrawlDailyFileService dailyFileService;
|
||||
@Spy private ObjectMapper objectMapper = new ObjectMapper().findAndRegisterModules();
|
||||
|
||||
private ShopDataCrawlTaskService service;
|
||||
|
||||
private final List<TaskChunkEntity> storedChunks = new ArrayList<>();
|
||||
private final List<TaskScopeStateEntity> storedScopes = new ArrayList<>();
|
||||
private final Map<String, String> rustfsPayloads = new LinkedHashMap<>();
|
||||
private FileTaskEntity task;
|
||||
private FileResultEntity result;
|
||||
private int nextPayloadId;
|
||||
private boolean insertFails;
|
||||
private boolean scopeUpdateFails;
|
||||
|
||||
@BeforeEach
|
||||
void configureStorage() {
|
||||
service = new ShopDataCrawlTaskService(
|
||||
fileTaskMapper,
|
||||
fileResultMapper,
|
||||
shopDataCrawlResolveService,
|
||||
excelAssemblyService,
|
||||
taskCacheService,
|
||||
ossStorageService,
|
||||
ziniaoShopSwitchService,
|
||||
objectMapper,
|
||||
taskPressureProperties,
|
||||
taskFileJobService,
|
||||
taskResultItemService,
|
||||
taskProgressSnapshotService,
|
||||
taskDistributedLockService,
|
||||
taskChunkMapper,
|
||||
taskScopeStateMapper,
|
||||
transientPayloadStorageService,
|
||||
instanceMetadata,
|
||||
dailyFileService,
|
||||
null);
|
||||
|
||||
storedChunks.clear();
|
||||
storedScopes.clear();
|
||||
rustfsPayloads.clear();
|
||||
nextPayloadId = 0;
|
||||
insertFails = false;
|
||||
scopeUpdateFails = false;
|
||||
|
||||
lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
|
||||
lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong()))
|
||||
.thenReturn(mock(TaskDistributedLockService.LockHandle.class));
|
||||
lenient().when(taskCacheService.getTaskCacheBatch(any())).thenReturn(Map.of());
|
||||
lenient().when(taskCacheService.getAllShopMergedPayload(anyLong())).thenReturn(Map.of());
|
||||
lenient().when(taskResultItemService.listResultSnapshots(anyLong(), eq(MODULE_TYPE), any())).thenReturn(List.of());
|
||||
lenient().when(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())).thenReturn(null);
|
||||
lenient().when(excelAssemblyService.countRows(any())).thenReturn(1);
|
||||
|
||||
lenient().when(fileTaskMapper.selectById(anyLong())).thenAnswer(invocation -> {
|
||||
Long taskId = invocation.getArgument(0);
|
||||
return task != null && Objects.equals(taskId, task.getId()) ? task : null;
|
||||
});
|
||||
lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1);
|
||||
lenient().when(fileResultMapper.selectById(anyLong())).thenAnswer(invocation -> {
|
||||
Long resultId = invocation.getArgument(0);
|
||||
return result != null && Objects.equals(resultId, result.getId()) ? result : null;
|
||||
});
|
||||
lenient().when(fileResultMapper.selectList(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<FileResultEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
return result != null && Objects.equals(taskId, result.getTaskId()) ? List.of(result) : List.of();
|
||||
});
|
||||
lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenReturn(1);
|
||||
|
||||
lenient().when(ziniaoShopSwitchService.normalizeShopName(nullable(String.class))).thenAnswer(invocation -> {
|
||||
String value = invocation.getArgument(0);
|
||||
return value == null ? "" : value.trim();
|
||||
});
|
||||
|
||||
configureTransientPayloadStorage();
|
||||
configureChunkMapper();
|
||||
configureScopeMapper();
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_030_task_normal_default_path() {
|
||||
// 正常路径：同一国家内重复行按稳定去重键合并，保留首现顺序；
|
||||
// 全部 10 个字段被保留（copyRow 语义），无重复行。
|
||||
givenRunningTask(1301L, 2301L);
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 1, "DE",
|
||||
row("2026-07-25", "B001"),
|
||||
row("2026-07-26", "B002"),
|
||||
sameRow("2026-07-25", "B001"))));
|
||||
|
||||
List<ShopDataCrawlRowDto> rows = finalRows();
|
||||
assertEquals(2, rows.size(), "重复行只保留一份");
|
||||
assertEquals("B001", rows.get(0).getAsin(), "首现顺序保留");
|
||||
assertEquals("B002", rows.get(1).getAsin());
|
||||
assertEquals("https://m.media-amazon.com/images/I/B001.jpg", rows.get(0).getCommodityImage());
|
||||
assertEquals("10", rows.get(0).getInventorySales());
|
||||
assertEquals("20", rows.get(0).getSalesRank());
|
||||
assertEquals("30", rows.get(0).getPageViews());
|
||||
assertEquals("40", rows.get(0).getUnitsSold());
|
||||
assertEquals("50", rows.get(0).getPrice());
|
||||
assertEquals("60", rows.get(0).getRecommendedOffer());
|
||||
|
||||
// 稳定去重键：语义相同（含空白差异）的行键相等，null 行安全。
|
||||
assertEquals(ShopDataCrawlTaskService.rowDedupKey(row("2026-07-25", "B001")),
|
||||
ShopDataCrawlTaskService.rowDedupKey(sameRow("2026-07-25", "B001")));
|
||||
assertEquals(ShopDataCrawlTaskService.rowDedupKey(row("2026-07-25", "B001")),
|
||||
ShopDataCrawlTaskService.rowDedupKey(paddedRow("2026-07-25", "B001")));
|
||||
assertEquals(null, ShopDataCrawlTaskService.rowDedupKey(null));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_030_task_normal_multiple_items() {
|
||||
// 批量场景：同一国家跨多个分片合并去重（跨分片重复行只保留首现），
|
||||
// 国家顺序与行顺序稳定（国家首次出现顺序、行首次出现顺序）。
|
||||
givenRunningTask(1302L, 2302L);
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 3, "DE",
|
||||
row("2026-07-25", "B001"),
|
||||
row("2026-07-26", "B002"))));
|
||||
service.submitResult(task.getId(), request(chunk(2, 3, "DE",
|
||||
row("2026-07-26", "B002"),
|
||||
row("2026-07-27", "B003"))));
|
||||
service.submitResult(task.getId(), request(chunk(3, 3, "UK",
|
||||
row("2026-07-27", "B003"),
|
||||
row("2026-07-28", "B004"))));
|
||||
|
||||
List<ShopDataCrawlCountryResultDto> countries = finalCountries();
|
||||
assertEquals(2, countries.size(), "国家按首次出现顺序");
|
||||
assertEquals("DE", countries.get(0).getCountry());
|
||||
assertEquals("UK", countries.get(1).getCountry());
|
||||
assertEquals(List.of("B001", "B002", "B003"), asins(countries.get(0).getItems()), "跨分片重复合并");
|
||||
assertEquals(List.of("B003", "B004"), asins(countries.get(1).getItems()));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_030_task_normal_repeated_operation_is_idempotent() {
|
||||
// 幂等：同一分片重放不会放大结果（命中唯一键，payload 清理，行不重复）。
|
||||
givenRunningTask(1303L, 2303L);
|
||||
ShopDataCrawlSubmitResultRequest request = request(chunk(1, 2, "DE",
|
||||
row("2026-07-25", "B001"),
|
||||
row("2026-07-26", "B002")));
|
||||
|
||||
service.submitResult(task.getId(), request);
|
||||
service.submitResult(task.getId(), request);
|
||||
service.submitResult(task.getId(), request(chunk(2, 2, "DE",
|
||||
row("2026-07-26", "B002"),
|
||||
row("2026-07-27", "B003"))));
|
||||
|
||||
assertEquals(List.of("B001", "B002", "B003"), asins(finalRows()), "重放不放大结果");
|
||||
assertEquals(1, storedScopes.size(), "scope 状态只有一行");
|
||||
assertEquals(2, storedChunks.size(), "分片行只有两份");
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_030_task_boundary_empty_input() {
|
||||
// 空输入：无可处理数据的分片在去重之前被拒，无结果行、无资源创建。
|
||||
givenRunningTask(1304L, 2304L);
|
||||
|
||||
ShopDataCrawlShopPayloadDto emptyItems = new ShopDataCrawlShopPayloadDto();
|
||||
emptyItems.setShopName(SHOP_NAME);
|
||||
emptyItems.setChunkIndex(1);
|
||||
emptyItems.setChunkTotal(1);
|
||||
emptyItems.setCountryResults(List.of(countryWithItems("DE", List.of())));
|
||||
BusinessException error = assertThrows(BusinessException.class, () ->
|
||||
service.submitResult(task.getId(), request(emptyItems)));
|
||||
assertTrue(error.getMessage().contains("内容为空"));
|
||||
|
||||
assertEquals(0, storedChunks.size());
|
||||
assertEquals(0, rustfsPayloads.size());
|
||||
assertEquals(0, storedScopes.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_030_task_boundary_single_item() {
|
||||
// 单元素：单国家单行，去重键单元素路径结果正确、顺序稳定。
|
||||
givenRunningTask(1305L, 2305L);
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 1, "DE", row("2026-07-25", "B001"))));
|
||||
|
||||
List<ShopDataCrawlRowDto> rows = finalRows();
|
||||
assertEquals(1, rows.size());
|
||||
assertEquals("B001", rows.get(0).getAsin());
|
||||
assertEquals("2026-07-25", rows.get(0).getDate());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_030_task_boundary_limit_and_overflow() {
|
||||
// 上限/超限：大量行合并只保留唯一行（每行唯一键不同），
|
||||
// 去重为一次遍历，行数正确、顺序稳定，无重复。
|
||||
givenRunningTask(1306L, 2306L);
|
||||
List<ShopDataCrawlRowDto> rows = new ArrayList<>();
|
||||
for (int i = 0; i < 200; i++) {
|
||||
rows.add(row(String.format("2026-07-%02d", i % 28 + 1), "B" + String.format("%04d", i)));
|
||||
}
|
||||
rows.add(rows.get(0));
|
||||
rows.add(rows.get(42));
|
||||
rows.add(rows.get(199));
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 1, "UK", rows)));
|
||||
List<ShopDataCrawlRowDto> finalRows = finalRows();
|
||||
assertEquals(200, finalRows.size(), "200 唯一行 + 3 个重复行只保留 200 行");
|
||||
for (int i = 0; i < finalRows.size(); i++) {
|
||||
assertEquals("B" + String.format("%04d", i), finalRows.get(i).getAsin(), "首现顺序稳定");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_030_task_invalid_input_rejected() {
|
||||
// 非法参数：分片元数据非法在去重之前拒绝；含空国家/空行的分片不产生任何结果行。
|
||||
givenRunningTask(1307L, 2307L);
|
||||
|
||||
BusinessException zeroIndex = assertThrows(BusinessException.class, () ->
|
||||
service.submitResult(task.getId(), request(chunk(0, 1, "DE", row("2026-07-25", "B001")))));
|
||||
assertTrue(zeroIndex.getMessage().contains("chunk_index"));
|
||||
|
||||
BusinessException zeroTotal = assertThrows(BusinessException.class, () ->
|
||||
service.submitResult(task.getId(), request(chunk(1, 0, "DE", row("2026-07-25", "B001")))));
|
||||
assertTrue(zeroTotal.getMessage().contains("chunk_total"));
|
||||
|
||||
BusinessException empty = assertThrows(BusinessException.class, () ->
|
||||
service.submitResult(task.getId(), request(blankRowsChunk(1, 1))));
|
||||
assertTrue(empty.getMessage().contains("内容为空"));
|
||||
|
||||
assertEquals(0, storedChunks.size(), "非法输入不落库");
|
||||
assertEquals(0, storedScopes.size());
|
||||
assertEquals(0, rustfsPayloads.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_030_task_dependency_failure_releases_resources() {
|
||||
// 依赖失败：chunk 插入异常 → 无结果行、payload 清理，恢复后重试成功；
|
||||
// scope 更新失败 → 回滚分片行与 payload，恢复后重试成功，去重结果不变。
|
||||
givenRunningTask(1308L, 2308L);
|
||||
|
||||
insertFails = true;
|
||||
assertThrows(RuntimeException.class, () ->
|
||||
service.submitResult(task.getId(), request(chunk(1, 2, "DE",
|
||||
row("2026-07-25", "B001")))));
|
||||
assertEquals(0, storedChunks.size());
|
||||
assertEquals(0, rustfsPayloads.size());
|
||||
assertEquals(0, storedScopes.size());
|
||||
insertFails = false;
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 2, "DE",
|
||||
row("2026-07-25", "B001"),
|
||||
row("2026-07-26", "B002"))));
|
||||
|
||||
scopeUpdateFails = true;
|
||||
assertThrows(RuntimeException.class, () ->
|
||||
service.submitResult(task.getId(), request(chunk(2, 2, "DE",
|
||||
row("2026-07-26", "B002"),
|
||||
row("2026-07-27", "B003")))));
|
||||
assertEquals(1, storedChunks.size(), "状态写入失败回滚分片行");
|
||||
assertEquals(1, rustfsPayloads.size(), "状态写入失败清理 payload");
|
||||
assertEquals(1, storedScopes.get(0).getReceivedChunkCount(), "状态写入失败计数器未推进");
|
||||
scopeUpdateFails = false;
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(2, 2, "DE",
|
||||
row("2026-07-26", "B002"),
|
||||
row("2026-07-27", "B003"))));
|
||||
|
||||
assertEquals(List.of("B001", "B002", "B003"), asins(finalRows()), "恢复后去重结果不变");
|
||||
assertEquals(1, storedScopes.size(), "scope 行始终只有一行");
|
||||
assertEquals(2, storedScopes.get(0).getReceivedChunkCount());
|
||||
assertEquals(1, storedScopes.get(0).getCompleted());
|
||||
}
|
||||
|
||||
private List<ShopDataCrawlRowDto> finalRows() {
|
||||
return finalCountries().get(0).getItems();
|
||||
}
|
||||
|
||||
private List<ShopDataCrawlCountryResultDto> finalCountries() {
|
||||
return parseResultJson().get(0).getCountryResults();
|
||||
}
|
||||
|
||||
private List<ShopDataCrawlResultItemVo> parseResultJson() {
|
||||
try {
|
||||
return objectMapper.readValue(task.getResultJson(),
|
||||
objectMapper.getTypeFactory().constructCollectionType(List.class, ShopDataCrawlResultItemVo.class));
|
||||
} catch (Exception ex) {
|
||||
throw new IllegalStateException("解析最终结果 JSON 失败", ex);
|
||||
}
|
||||
}
|
||||
|
||||
private List<String> asins(List<ShopDataCrawlRowDto> rows) {
|
||||
List<String> result = new ArrayList<>();
|
||||
if (rows == null) {
|
||||
return result;
|
||||
}
|
||||
for (ShopDataCrawlRowDto row : rows) {
|
||||
result.add(row.getAsin());
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
private void configureTransientPayloadStorage() {
|
||||
lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true);
|
||||
lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned(
|
||||
anyString(), anyLong(), anyString(), any(), anyString())).thenAnswer(invocation -> {
|
||||
String pointer = "rustfs:payload-" + (++nextPayloadId);
|
||||
rustfsPayloads.put(pointer, invocation.getArgument(4));
|
||||
return pointer;
|
||||
});
|
||||
lenient().when(transientPayloadStorageService.extractPointer(any())).thenAnswer(invocation -> {
|
||||
String value = invocation.getArgument(0);
|
||||
if (value == null) {
|
||||
return null;
|
||||
}
|
||||
return value.startsWith("rustfs:") || value.startsWith("local:") || value.startsWith("oss:")
|
||||
? value : null;
|
||||
});
|
||||
lenient().when(transientPayloadStorageService.resolvePayload(any(), any())).thenAnswer(invocation -> {
|
||||
String pointer = invocation.getArgument(0);
|
||||
String payload = rustfsPayloads.get(pointer);
|
||||
if (payload == null) {
|
||||
throw new IllegalStateException("missing test RustFS payload: " + pointer);
|
||||
}
|
||||
return payload;
|
||||
});
|
||||
lenient().doAnswer(invocation -> {
|
||||
rustfsPayloads.remove(invocation.getArgument(0));
|
||||
return null;
|
||||
}).when(transientPayloadStorageService).deletePayloadIfPresent(any());
|
||||
}
|
||||
|
||||
private void configureChunkMapper() {
|
||||
lenient().when(taskChunkMapper.insert(any(TaskChunkEntity.class))).thenAnswer(invocation -> {
|
||||
if (insertFails) {
|
||||
throw new RuntimeException("db down");
|
||||
}
|
||||
TaskChunkEntity chunk = invocation.getArgument(0);
|
||||
boolean duplicate = storedChunks.stream().anyMatch(existing ->
|
||||
Objects.equals(existing.getTaskId(), chunk.getTaskId())
|
||||
&& Objects.equals(existing.getScopeHash(), chunk.getScopeHash())
|
||||
&& Objects.equals(existing.getChunkIndex(), chunk.getChunkIndex()));
|
||||
if (duplicate) {
|
||||
throw new DuplicateKeyException("duplicate chunk key: " + chunk.getScopeHash() + "/" + chunk.getChunkIndex());
|
||||
}
|
||||
chunk.setId((long) storedChunks.size() + 1L);
|
||||
storedChunks.add(chunk);
|
||||
return 1;
|
||||
});
|
||||
lenient().when(taskChunkMapper.selectOne(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<TaskChunkEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
String scopeHash = queryScopeHash(query);
|
||||
Integer chunkIndex = queryInteger(query);
|
||||
return storedChunks.stream()
|
||||
.filter(chunk -> matchesChunk(chunk, taskId, scopeHash, chunkIndex))
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
});
|
||||
lenient().when(taskChunkMapper.selectList(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<TaskChunkEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
String scopeHash = queryScopeHash(query);
|
||||
return storedChunks.stream()
|
||||
.filter(chunk -> matchesChunk(chunk, taskId, scopeHash, null))
|
||||
.sorted(Comparator.comparing(TaskChunkEntity::getChunkIndex))
|
||||
.toList();
|
||||
});
|
||||
lenient().when(taskChunkMapper.delete(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<TaskChunkEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
String scopeHash = queryScopeHash(query);
|
||||
Integer chunkIndex = queryInteger(query);
|
||||
int before = storedChunks.size();
|
||||
storedChunks.removeIf(chunk -> matchesChunk(chunk, taskId, scopeHash, chunkIndex));
|
||||
return before - storedChunks.size();
|
||||
});
|
||||
}
|
||||
|
||||
private void configureScopeMapper() {
|
||||
lenient().when(taskScopeStateMapper.insert(any(TaskScopeStateEntity.class))).thenAnswer(invocation -> {
|
||||
TaskScopeStateEntity scope = invocation.getArgument(0);
|
||||
scope.setId((long) storedScopes.size() + 1L);
|
||||
storedScopes.add(scope);
|
||||
return 1;
|
||||
});
|
||||
lenient().when(taskScopeStateMapper.selectOne(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<TaskScopeStateEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
String scopeHash = queryScopeHash(query);
|
||||
return storedScopes.stream()
|
||||
.filter(scope -> Objects.equals(taskId, scope.getTaskId()))
|
||||
.filter(scope -> scopeHash == null || Objects.equals(scopeHash, scope.getScopeHash()))
|
||||
.findFirst()
|
||||
.map(ShopDataCrawlRowDedupKeyTest::copyScope)
|
||||
.orElse(null);
|
||||
});
|
||||
lenient().when(taskScopeStateMapper.updateById(any(TaskScopeStateEntity.class))).thenAnswer(invocation -> {
|
||||
if (scopeUpdateFails) {
|
||||
throw new RuntimeException("scope update down");
|
||||
}
|
||||
TaskScopeStateEntity updated = invocation.getArgument(0);
|
||||
for (int i = 0; i < storedScopes.size(); i++) {
|
||||
if (Objects.equals(storedScopes.get(i).getId(), updated.getId())) {
|
||||
storedScopes.set(i, copyScope(updated));
|
||||
}
|
||||
}
|
||||
return 1;
|
||||
});
|
||||
}
|
||||
|
||||
private static TaskScopeStateEntity copyScope(TaskScopeStateEntity source) {
|
||||
TaskScopeStateEntity copy = new TaskScopeStateEntity();
|
||||
copy.setId(source.getId());
|
||||
copy.setTaskId(source.getTaskId());
|
||||
copy.setModuleType(source.getModuleType());
|
||||
copy.setScopeKey(source.getScopeKey());
|
||||
copy.setScopeHash(source.getScopeHash());
|
||||
copy.setChunkTotal(source.getChunkTotal());
|
||||
copy.setReceivedChunkCount(source.getReceivedChunkCount());
|
||||
copy.setCompleted(source.getCompleted());
|
||||
copy.setLastChunkAt(source.getLastChunkAt());
|
||||
copy.setLastError(source.getLastError());
|
||||
copy.setCreatedAt(source.getCreatedAt());
|
||||
copy.setUpdatedAt(source.getUpdatedAt());
|
||||
return copy;
|
||||
}
|
||||
|
||||
private boolean matchesChunk(TaskChunkEntity chunk, Long taskId, String scopeHash, Integer chunkIndex) {
|
||||
return (taskId == null || Objects.equals(taskId, chunk.getTaskId()))
|
||||
&& (scopeHash == null || Objects.equals(scopeHash, chunk.getScopeHash()))
|
||||
&& (chunkIndex == null || Objects.equals(chunkIndex, chunk.getChunkIndex()));
|
||||
}
|
||||
|
||||
private Long queryLong(LambdaQueryWrapper<?> query) {
|
||||
query.getSqlSegment();
|
||||
return query.getParamNameValuePairs().values().stream()
|
||||
.filter(Long.class::isInstance)
|
||||
.map(Long.class::cast)
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
private Integer queryInteger(LambdaQueryWrapper<?> query) {
|
||||
query.getSqlSegment();
|
||||
return query.getParamNameValuePairs().values().stream()
|
||||
.filter(Integer.class::isInstance)
|
||||
.map(Integer.class::cast)
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
private String queryScopeHash(LambdaQueryWrapper<?> query) {
|
||||
query.getSqlSegment();
|
||||
return query.getParamNameValuePairs().values().stream()
|
||||
.filter(String.class::isInstance)
|
||||
.map(String.class::cast)
|
||||
.filter(value -> value.length() == 64)
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
private void givenRunningTask(long taskId, long resultId) {
|
||||
task = new FileTaskEntity();
|
||||
task.setId(taskId);
|
||||
task.setUserId(7L);
|
||||
task.setModuleType(MODULE_TYPE);
|
||||
task.setStatus("RUNNING");
|
||||
task.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}");
|
||||
task.setResultJson("[]");
|
||||
task.setCreatedAt(LocalDateTime.now());
|
||||
task.setUpdatedAt(LocalDateTime.now());
|
||||
|
||||
result = new FileResultEntity();
|
||||
result.setId(resultId);
|
||||
result.setTaskId(taskId);
|
||||
result.setUserId(7L);
|
||||
result.setModuleType(MODULE_TYPE);
|
||||
result.setSourceFilename(SHOP_NAME);
|
||||
result.setSourceFileUrl("shop-1");
|
||||
result.setSuccess(-1);
|
||||
result.setCreatedAt(LocalDateTime.now());
|
||||
}
|
||||
|
||||
private ShopDataCrawlSubmitResultRequest request(ShopDataCrawlShopPayloadDto payload) {
|
||||
ShopDataCrawlSubmitResultRequest request = new ShopDataCrawlSubmitResultRequest();
|
||||
request.setShops(List.of(payload));
|
||||
return request;
|
||||
}
|
||||
|
||||
private ShopDataCrawlShopPayloadDto chunk(int chunkIndex,
|
||||
int chunkTotal,
|
||||
String country,
|
||||
List<ShopDataCrawlRowDto> rows) {
|
||||
ShopDataCrawlShopPayloadDto payload = new ShopDataCrawlShopPayloadDto();
|
||||
payload.setShopName(SHOP_NAME);
|
||||
payload.setChunkIndex(chunkIndex);
|
||||
payload.setChunkTotal(chunkTotal);
|
||||
payload.setCountryResults(List.of(countryWithItems(country, rows)));
|
||||
return payload;
|
||||
}
|
||||
|
||||
private ShopDataCrawlShopPayloadDto chunk(int chunkIndex,
|
||||
int chunkTotal,
|
||||
String country,
|
||||
ShopDataCrawlRowDto... rows) {
|
||||
return chunk(chunkIndex, chunkTotal, country, List.of(rows));
|
||||
}
|
||||
|
||||
private ShopDataCrawlShopPayloadDto blankRowsChunk(int chunkIndex, int chunkTotal) {
|
||||
ShopDataCrawlShopPayloadDto payload = new ShopDataCrawlShopPayloadDto();
|
||||
payload.setShopName(SHOP_NAME);
|
||||
payload.setChunkIndex(chunkIndex);
|
||||
payload.setChunkTotal(chunkTotal);
|
||||
ShopDataCrawlCountryResultDto country = new ShopDataCrawlCountryResultDto();
|
||||
country.setCountry("DE");
|
||||
ShopDataCrawlRowDto blank = new ShopDataCrawlRowDto();
|
||||
blank.setDate("");
|
||||
blank.setAsin(" ");
|
||||
country.setItems(List.of(blank));
|
||||
payload.setCountryResults(List.of(country));
|
||||
return payload;
|
||||
}
|
||||
|
||||
private ShopDataCrawlCountryResultDto countryWithItems(String country, List<ShopDataCrawlRowDto> items) {
|
||||
ShopDataCrawlCountryResultDto result = new ShopDataCrawlCountryResultDto();
|
||||
result.setCountry(country);
|
||||
result.setItems(items);
|
||||
return result;
|
||||
}
|
||||
|
||||
private ShopDataCrawlRowDto row(String date, String asin) {
|
||||
ShopDataCrawlRowDto row = new ShopDataCrawlRowDto();
|
||||
row.setDate(date);
|
||||
row.setAsin(asin);
|
||||
row.setCommodityImage("https://m.media-amazon.com/images/I/" + asin + ".jpg");
|
||||
row.setInventorySales("10");
|
||||
row.setSalesRank("20");
|
||||
row.setPageViews("30");
|
||||
row.setUnitsSold("40");
|
||||
row.setPrice("50");
|
||||
row.setRecommendedOffer("60");
|
||||
return row;
|
||||
}
|
||||
|
||||
private ShopDataCrawlRowDto sameRow(String date, String asin) {
|
||||
return row(date, asin);
|
||||
}
|
||||
|
||||
private ShopDataCrawlRowDto paddedRow(String date, String asin) {
|
||||
ShopDataCrawlRowDto row = new ShopDataCrawlRowDto();
|
||||
row.setDate(" " + date + " ");
|
||||
row.setAsin(" " + asin + " ");
|
||||
row.setCommodityImage(" https://m.media-amazon.com/images/I/" + asin + ".jpg ");
|
||||
row.setInventorySales(" 10 ");
|
||||
row.setSalesRank(" 20 ");
|
||||
row.setPageViews(" 30 ");
|
||||
row.setUnitsSold(" 40 ");
|
||||
row.setPrice(" 50 ");
|
||||
row.setRecommendedOffer(" 60 ");
|
||||
return row;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user