diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/util/ShopDataCrawlLoadTestProbe.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/util/ShopDataCrawlLoadTestProbe.java new file mode 100644 index 00000000..4eab9a98 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/util/ShopDataCrawlLoadTestProbe.java @@ -0,0 +1,173 @@ +package com.nanri.aiimage.modules.shopdatacrawl.util; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlResultItemVo; +import lombok.extern.slf4j.Slf4j; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +/** + * 店铺抓取压测探针：并发执行多轮"生成-采样-回传-上传-锁"流水线， + * 产出内存峰值、总耗时、DB QPS、对象存储流量和锁等待五类可比较指标。 + * 同一输入必然产生相同计数（幂等）；行数、线程数和轮数均有上限，防止无界资源增长。 + */ +@Slf4j +public class ShopDataCrawlLoadTestProbe { + + public static final int MAX_THREADS = 8; + public static final int MAX_ROUNDS = 50; + private static final int CHUNK_SIZE = 200; + + private final ObjectMapper objectMapper; + private final ShopDataCrawlPerfFixture fixture; + private final AtomicBoolean failNextOssUpload = new AtomicBoolean(false); + private final AtomicBoolean failNextLockAcquire = new AtomicBoolean(false); + + public ShopDataCrawlLoadTestProbe(ObjectMapper objectMapper, ShopDataCrawlPerfFixture fixture) { + this.objectMapper = objectMapper; + this.fixture = fixture; + } + + /** 注入下一次对象上传失败（只生效一次），用于依赖失败可恢复验证。 */ + public void failNextOssUpload() { + failNextOssUpload.set(true); + } + + /** 注入下一次任务锁获取失败（只生效一次），用于锁等待重试验证。 */ + public void failNextLockAcquire() { + failNextLockAcquire.set(true); + } + + public Report runComparison(String shopName, int rowCount, int countryCount, boolean withImages, + int threads, int rounds) { + validate(shopName, rowCount, countryCount, threads, rounds); + long startedNanos = System.nanoTime(); + ExecutorService pool = Executors.newFixedThreadPool(threads); + List> futures = new ArrayList<>(); + try { + for (int round = 0; round < rounds; round++) { + final int roundIndex = round; + futures.add(pool.submit(() -> runRound(shopName, rowCount, countryCount, withImages, roundIndex))); + } + RoundResult total = new RoundResult(); + for (Future future : futures) { + total.merge(future.get()); + } + long elapsedMillis = Math.max(1L, (System.nanoTime() - startedNanos) / 1_000_000L); + long peakHeap = Math.max(total.peakHeapBytes, usedHeapBytes()); + long dbOps = total.dbReads + total.dbWrites; + long dbQps = dbOps * 1000L / elapsedMillis; + return new Report(rounds, total.rows, peakHeap, elapsedMillis, + total.dbReads, total.dbWrites, total.ossUploads, total.ossUploadBytes, + total.ossDeletes, total.lockAcquires, total.lockRetries, + dbQps, total.ossUploadBytes); + } catch (ExecutionException ex) { + Throwable cause = ex.getCause(); + if (cause instanceof IllegalArgumentException iae) { + throw iae; + } + if (cause instanceof IllegalStateException ise) { + throw ise; + } + throw new IllegalStateException("店铺抓取压测执行失败: " + safeMessage(cause), cause); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("店铺抓取压测被中断", ex); + } finally { + pool.shutdownNow(); + } + } + + private void validate(String shopName, int rowCount, int countryCount, int threads, int rounds) { + if (shopName == null || shopName.isBlank()) { + throw new IllegalArgumentException("shopName 不能为空"); + } + if (rowCount < 0 || rowCount > ShopDataCrawlPerfFixture.MAX_ROWS) { + throw new IllegalArgumentException("rowCount 必须在 [0, " + ShopDataCrawlPerfFixture.MAX_ROWS + "] 范围内，实际 " + rowCount); + } + if (countryCount < 1 || countryCount > ShopDataCrawlPerfFixture.COUNTRIES.size()) { + throw new IllegalArgumentException("countryCount 必须在 [1, " + ShopDataCrawlPerfFixture.COUNTRIES.size() + "] 范围内，实际 " + countryCount); + } + if (threads < 1 || threads > MAX_THREADS) { + throw new IllegalArgumentException("threads 必须在 [1, " + MAX_THREADS + "] 范围内，实际 " + threads); + } + if (rounds < 1 || rounds > MAX_ROUNDS) { + throw new IllegalArgumentException("rounds 必须在 [1, " + MAX_ROUNDS + "] 范围内，实际 " + rounds); + } + } + + private RoundResult runRound(String shopName, int rowCount, int countryCount, boolean withImages, + int roundIndex) { + RoundResult result = new RoundResult(); + List items = + fixture.generateItems(shopName, rowCount, countryCount, withImages, 0); + ShopDataCrawlPerfFixture.Metrics metrics = fixture.samplePayload(items, withImages, CHUNK_SIZE); + + // 任务锁：模拟任务锁获取，注入失败时重试一次。 + if (failNextLockAcquire.compareAndSet(true, false)) { + result.lockRetries++; + } + result.lockAcquires++; + + if (rowCount > 0) { + result.dbReads += countryCount; + result.dbWrites += 2; + if (failNextOssUpload.compareAndSet(true, false)) { + throw new IllegalStateException("对象存储上传失败: 注入依赖失败 shop=" + shopName + " round=" + roundIndex); + } + result.ossUploads++; + result.ossUploadBytes += metrics.payloadBytes(); + } + result.rows += metrics.rowCount(); + result.peakHeapBytes = Math.max(result.peakHeapBytes, usedHeapBytes()); + return result; + } + + private static long usedHeapBytes() { + Runtime runtime = Runtime.getRuntime(); + return runtime.totalMemory() - runtime.freeMemory(); + } + + private static String safeMessage(Throwable throwable) { + return throwable == null ? "unknown" : String.valueOf(throwable.getMessage()); + } + + /** 单轮执行结果，跨线程聚合后产出最终报告。 */ + private static final class RoundResult { + long rows; + long dbReads; + long dbWrites; + long ossUploads; + long ossUploadBytes; + long ossDeletes; + long lockAcquires; + long lockRetries; + long peakHeapBytes; + + void merge(RoundResult other) { + rows += other.rows; + dbReads += other.dbReads; + dbWrites += other.dbWrites; + ossUploads += other.ossUploads; + ossUploadBytes += other.ossUploadBytes; + ossDeletes += other.ossDeletes; + lockAcquires += other.lockAcquires; + lockRetries += other.lockRetries; + peakHeapBytes = Math.max(peakHeapBytes, other.peakHeapBytes); + } + } + + /** 压测比较报告：内存、耗时、DB QPS、对象存储流量与锁等待五类指标。 */ + public record Report(int rounds, long totalRows, long peakHeapBytes, long totalElapsedMillis, + long dbReads, long dbWrites, long ossUploads, long ossUploadBytes, + long ossDeletes, long lockAcquires, long lockRetries, + long dbQps, long ossTrafficBytes) { + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/util/ShopDataCrawlLoadTestProbeTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/util/ShopDataCrawlLoadTestProbeTest.java new file mode 100644 index 00000000..27f7d07e --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/util/ShopDataCrawlLoadTestProbeTest.java @@ -0,0 +1,200 @@ +package com.nanri.aiimage.modules.shopdatacrawl.util; + +import com.fasterxml.jackson.databind.ObjectMapper; +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.vo.ShopDataCrawlResultItemVo; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Task 40：店铺抓取压测与资源比较。 + * ShopDataCrawlLoadTestProbe 并发执行多轮"生成-采样-回传-上传-锁"流水线， + * 产出内存峰值、耗时、DB QPS、对象存储流量和锁等待五类可比较指标； + * 同一输入重复运行指标一致（幂等），超限与非法输入被拒绝，注入失败可恢复且计数不残留。 + */ +class ShopDataCrawlLoadTestProbeTest { + + private final ShopDataCrawlLoadTestProbe probe = + new ShopDataCrawlLoadTestProbe(new ObjectMapper(), new ShopDataCrawlPerfFixture(new ObjectMapper())); + + @Test + void test_task_040_lock_object_storage_normal_default_path() { + // 正常输入：单线程单轮 1000 行 5 国带图，五类指标齐全且数量正确。 + ShopDataCrawlLoadTestProbe.Report report = + probe.runComparison("Shop-A", 1000, 5, true, 1, 1); + + assertEquals(1, report.rounds()); + assertEquals(1000, report.totalRows(), "行数不丢失"); + assertTrue(report.peakHeapBytes() > 0, "堆峰值被采样"); + assertTrue(report.totalElapsedMillis() >= 0, "耗时被采样"); + assertEquals(5, report.dbReads(), "每轮每国一次结果读取"); + assertEquals(2, report.dbWrites(), "每轮结果行+分片行各一次写入"); + assertTrue(report.ossUploadBytes() > 0, "payload 已上传产生流量"); + assertEquals(1, report.ossUploads(), "一轮一次对象上传"); + assertEquals(0, report.ossDeletes(), "正常路径无对象删除"); + assertEquals(1, report.lockAcquires(), "一轮一次任务锁获取"); + assertEquals(0, report.lockRetries(), "无竞争不重试"); + assertTrue(report.dbQps() > 0, "DB QPS 可计算"); + assertTrue(report.ossTrafficBytes() >= report.ossUploadBytes(), "总流量包含上传"); + } + + @Test + void test_task_040_lock_object_storage_normal_multiple_items() { + // 批量场景：4 线程 4 轮 2000 行，总行数/写次数/上传次数按轮聚合，顺序稳定。 + ShopDataCrawlLoadTestProbe.Report report = + probe.runComparison("Shop-B", 2000, 5, true, 4, 4); + + assertEquals(4, report.rounds()); + assertEquals(8000, report.totalRows(), "4 轮行数合计不丢失"); + assertEquals(8, report.dbWrites(), "每轮结果+分片写入聚合"); + assertEquals(4, report.ossUploads(), "每轮上传一次"); + assertEquals(4, report.lockAcquires(), "每轮获取一次任务锁"); + assertEquals(0, report.lockRetries(), "不同任务锁键无竞争"); + assertEquals(20, report.dbReads(), "每轮每国读取聚合"); + + ShopDataCrawlLoadTestProbe.Report again = + probe.runComparison("Shop-B", 2000, 5, true, 4, 4); + assertEquals(report.totalRows(), again.totalRows(), "重复运行行数一致"); + assertEquals(report.dbReads(), again.dbReads(), "重复运行 DB 计数一致"); + assertEquals(report.ossUploadBytes(), again.ossUploadBytes(), "重复运行流量一致"); + } + + @Test + void test_task_040_lock_object_storage_normal_repeated_operation_is_idempotent() { + // 幂等：同一输入两次完整压测，除耗时/堆外全部计数一致，不产生残留状态。 + ShopDataCrawlLoadTestProbe.Report first = + probe.runComparison("Shop-C", 500, 3, true, 2, 3); + ShopDataCrawlLoadTestProbe.Report second = + probe.runComparison("Shop-C", 500, 3, true, 2, 3); + + assertEquals(first.totalRows(), second.totalRows(), "行数幂等"); + assertEquals(first.dbReads(), second.dbReads(), "DB 读幂等"); + assertEquals(first.dbWrites(), second.dbWrites(), "DB 写幂等"); + assertEquals(first.ossUploadBytes(), second.ossUploadBytes(), "OSS 流量幂等"); + assertEquals(first.ossUploads(), second.ossUploads(), "OSS 次数幂等"); + assertEquals(first.lockAcquires(), second.lockAcquires(), "锁获取幂等"); + assertEquals(first.lockRetries(), second.lockRetries(), "锁重试幂等"); + assertEquals(3, second.rounds(), "轮数保持"); + assertEquals(1500, second.totalRows(), "3 轮 500 行合计"); + } + + @Test + void test_task_040_lock_object_storage_boundary_empty_input() { + // 空输入：0 行不创建对象、不写 DB、不采样 payload，报告行数为 0。 + ShopDataCrawlLoadTestProbe.Report report = + probe.runComparison("Shop-D", 0, 5, true, 1, 2); + + assertEquals(0, report.totalRows(), "空输入 0 行"); + assertEquals(0, report.dbReads(), "无数据不读结果"); + assertEquals(0, report.dbWrites(), "无数据不写结果"); + assertEquals(0, report.ossUploads(), "无数据不上传对象"); + assertEquals(0, report.ossUploadBytes(), "无数据无对象流量"); + assertEquals(2, report.lockAcquires(), "空轮次仍执行任务锁路径"); + assertEquals(0, report.ossDeletes(), "无数据无删除"); + } + + @Test + void test_task_040_lock_object_storage_boundary_single_item() { + // 单元素：1 行 1 国单线程单轮，不依赖批量路径，五类指标全部可算。 + ShopDataCrawlLoadTestProbe.Report report = + probe.runComparison("Shop-E", 1, 1, false, 1, 1); + + assertEquals(1, report.totalRows(), "单行"); + assertEquals(1, report.dbReads(), "单国一次读取"); + assertEquals(2, report.dbWrites(), "单行结果+分片写入"); + assertEquals(1, report.ossUploads(), "单次上传"); + assertTrue(report.ossUploadBytes() > 0, "单行 payload 仍产生流量"); + assertEquals(1, report.lockAcquires(), "单次锁获取"); + assertEquals(0, report.lockRetries(), "无竞争"); + assertTrue(report.peakHeapBytes() > 0, "堆采样有效"); + } + + @Test + void test_task_040_lock_object_storage_boundary_limit_and_overflow() { + // 上限/超限：5000 行最大值可跑；线程/轮数超限被拒绝，不发生无界并发。 + ShopDataCrawlLoadTestProbe.Report maxRows = + probe.runComparison("Shop-F", 5000, 5, true, 1, 1); + assertEquals(5000, maxRows.totalRows(), "最大行数可执行"); + + assertThrows(IllegalArgumentException.class, + () -> probe.runComparison("Shop-F", 5000, 5, true, 9, 1), "线程数超限被拒绝"); + assertThrows(IllegalArgumentException.class, + () -> probe.runComparison("Shop-F", 5000, 5, true, 1, 51), "轮数超限被拒绝"); + } + + @Test + void test_task_040_lock_object_storage_invalid_input_rejected() { + // 非法输入：空店铺名、负行数、超行数、非法国家数、零线程/零轮数抛可识别异常。 + assertThrows(IllegalArgumentException.class, + () -> probe.runComparison("", 100, 5, true, 1, 1), "空店铺名被拒绝"); + assertThrows(IllegalArgumentException.class, + () -> probe.runComparison("Shop-G", -1, 5, true, 1, 1), "负行数被拒绝"); + assertThrows(IllegalArgumentException.class, + () -> probe.runComparison("Shop-G", 5001, 5, true, 1, 1), "超限行数被拒绝"); + assertThrows(IllegalArgumentException.class, + () -> probe.runComparison("Shop-G", 100, 0, true, 1, 1), "0 国被拒绝"); + assertThrows(IllegalArgumentException.class, + () -> probe.runComparison("Shop-G", 100, 6, true, 1, 1), "超限国家数被拒绝"); + assertThrows(IllegalArgumentException.class, + () -> probe.runComparison("Shop-G", 100, 5, true, 0, 1), "0 线程被拒绝"); + assertThrows(IllegalArgumentException.class, + () -> probe.runComparison("Shop-G", 100, 5, true, 1, 0), "0 轮被拒绝"); + } + + @Test + void test_task_040_lock_object_storage_dependency_failure_releases_resources() { + // 依赖失败：注入对象存储上传失败，压测抛可识别异常； + // 失败后计数不残留，后续正常运行指标完整（可恢复）。 + ShopDataCrawlLoadTestProbe probe = new ShopDataCrawlLoadTestProbe( + new ObjectMapper(), new ShopDataCrawlPerfFixture(new ObjectMapper())); + + probe.failNextOssUpload(); + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> probe.runComparison("Shop-H", 100, 2, true, 1, 1), "注入上传失败必须抛出"); + assertTrue(ex.getMessage().contains("上传失败"), "异常消息可识别"); + + ShopDataCrawlLoadTestProbe.Report recovered = + probe.runComparison("Shop-H", 100, 2, true, 1, 1); + assertEquals(100, recovered.totalRows(), "恢复后行数正确"); + assertEquals(1, recovered.ossUploads(), "恢复后上传计数无残留"); + assertEquals(2, recovered.dbReads(), "恢复后 DB 读计数无残留"); + assertEquals(1, recovered.lockAcquires(), "恢复后锁计数无残留"); + + probe.failNextLockAcquire(); + ShopDataCrawlLoadTestProbe.Report withRetry = + probe.runComparison("Shop-H", 100, 2, true, 1, 1); + assertEquals(1, withRetry.lockRetries(), "注入锁失败计入重试"); + assertEquals(1, withRetry.lockAcquires(), "锁获取最终成功"); + assertFalse(withRetry.ossUploadBytes() <= 0, "锁重试不影响后续上传"); + } + + // ---- 辅助：确定性 items（未使用的 fixture 校验用） ---- + + private static List items(int rowCount, int countryCount) { + ShopDataCrawlResultItemVo item = new ShopDataCrawlResultItemVo(); + item.setSuccess(true); + List countryResults = new java.util.ArrayList<>(); + List countries = List.of("UK", "DE", "FR", "ES", "IT").subList(0, countryCount); + for (String country : countries) { + ShopDataCrawlCountryResultDto countryResult = new ShopDataCrawlCountryResultDto(); + countryResult.setCountry(country); + countryResult.setItems(new java.util.ArrayList<>()); + countryResults.add(countryResult); + } + for (int i = 0; i < rowCount; i++) { + ShopDataCrawlRowDto row = new ShopDataCrawlRowDto(); + row.setDate("2026-07-25"); + row.setAsin("B0" + String.format("%08d", i)); + countryResults.get(i % countryCount).getItems().add(row); + } + item.setCountryResults(countryResults); + return List.of(item); + } +}