task-40: 完成店铺抓取压测探针与资源比较
新增 ShopDataCrawlLoadTestProbe：并发多轮执行"生成-采样-回传-上传-锁" 流水线，产出内存峰值、耗时、DB QPS、对象存储流量与锁等待五类可比较 指标；行数/线程/轮数均有上限，注入 OSS 上传与锁获取失败验证依赖降级 可恢复且计数不残留。8 个用例覆盖正常/批量/幂等/空/单元素/超限/非法 输入/依赖失败路径，mvn 全量 672 测试通过。
This commit is contained in:
+173
@@ -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<Future<RoundResult>> 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<RoundResult> 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<ShopDataCrawlResultItemVo> 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) {
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user