feat(撞款): 采集明细落库后自动触发重扫，重复检查无需等次日 00:00
- shopdatacrawl 新增 DuplicateCheckRefreshPort 端口，TaskService 在明细落库成功后请求重扫 - shopduplicatecheck 新增 DuplicateCheckRefreshScheduler：10s 合并窗口去抖、单飞、 扫描锁忙自动重试（最多 6 次），失败不影响采集归档 - 复用现有 SCAN_LOCK，与定时扫描、手动「重新分析」互斥；新增 5 个调度器单测
This commit is contained in:
+9
@@ -27,6 +27,7 @@ import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlTaskBatchVo
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlCreateTaskVo;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlHistoryVo;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlResultItemVo;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.spi.DuplicateCheckRefreshPort;
|
||||
import com.nanri.aiimage.common.model.vo.ProductRiskDashboardVo;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||
@@ -121,6 +122,7 @@ public class ShopDataCrawlTaskService {
|
||||
private final InstanceMetadata instanceMetadata;
|
||||
private final ShopDataCrawlDailyFileService dailyFileService;
|
||||
private final ShopDataCrawlItemStoreService shopDataCrawlItemStoreService;
|
||||
private final DuplicateCheckRefreshPort duplicateCheckRefreshPort;
|
||||
private final PlatformTransactionManager transactionManager;
|
||||
private final TaskProgressLightAssembler taskProgressLightAssembler;
|
||||
|
||||
@@ -2116,6 +2118,13 @@ public class ShopDataCrawlTaskService {
|
||||
shopDataCrawlItemStoreService.saveShopBatchFromSnapshot(
|
||||
snapshot.getShopName(), itemBatchDate, accumulatedItems,
|
||||
snapshot.getResultId(), task.getId(), baseDailyFile == null ? null : baseDailyFile.getId());
|
||||
// 明细已落库：请求撞款重扫（异步合并执行，不阻塞归档；端口契约保证不抛错）
|
||||
try {
|
||||
duplicateCheckRefreshPort.requestRefresh("shop-data-crawl:" + snapshot.getShopName());
|
||||
} catch (RuntimeException ex) {
|
||||
log.warn("[shop-data-crawl] 请求撞款重扫失败（忽略，不影响归档） shop={} msg={}",
|
||||
snapshot.getShopName(), ex.getMessage());
|
||||
}
|
||||
int rowCount = excelAssemblyService.writeWorkbook(outputXlsx, accumulatedItems);
|
||||
String objectKey = ossStorageService.uploadResultFile(outputXlsx, MODULE_TYPE);
|
||||
if (blank(objectKey)) {
|
||||
|
||||
+17
@@ -0,0 +1,17 @@
|
||||
package com.nanri.aiimage.modules.shopdatacrawl.spi;
|
||||
|
||||
/**
|
||||
* 采集明细就绪后的撞款重扫触发端口（2026-09：店铺数据采集落库后即时刷新重复检查）。
|
||||
*
|
||||
* <p>实现方在 shopduplicatecheck 模块（{@code ShopDataDuplicateCheckScanService}）。
|
||||
* 契约：实现必须异步执行、去抖合并，不得阻塞调用方、不得向外抛出异常。
|
||||
*/
|
||||
public interface DuplicateCheckRefreshPort {
|
||||
|
||||
/**
|
||||
* 请求一次撞款重扫（异步；合并窗口内的多次触发聚合为一次扫描）。
|
||||
*
|
||||
* @param reason 触发来源，仅用于日志排查
|
||||
*/
|
||||
void requestRefresh(String reason);
|
||||
}
|
||||
+32
-2
@@ -9,6 +9,7 @@ import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlCountryRes
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlRowDto;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlResultItemVo;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.service.ShopDataCrawlItemStoreService;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.spi.DuplicateCheckRefreshPort;
|
||||
import com.nanri.aiimage.modules.shopduplicatecheck.mapper.ShopDataDuplicateScanMapper;
|
||||
import com.nanri.aiimage.modules.shopduplicatecheck.mapper.ShopDuplicateCheckItemMapper;
|
||||
import com.nanri.aiimage.modules.shopduplicatecheck.mapper.ShopDuplicateCheckSourceMapper;
|
||||
@@ -21,6 +22,7 @@ import com.nanri.aiimage.modules.shopduplicatecheck.model.entity.ShopDataDuplica
|
||||
import com.nanri.aiimage.modules.shopduplicatecheck.model.payload.DuplicateScanPayload;
|
||||
import com.nanri.aiimage.modules.shopduplicatecheck.model.payload.DuplicateScanSummary;
|
||||
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.DuplicateCheckAggregator;
|
||||
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.DuplicateCheckRefreshScheduler;
|
||||
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.DuplicateCheckWorkbookParser;
|
||||
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.RawRow;
|
||||
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.ShopParsed;
|
||||
@@ -43,14 +45,14 @@ import java.util.Set;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
/**
|
||||
* 撞款扫描：每日 00:00 定时 + force 同步全量重扫。
|
||||
* 撞款扫描：每日 00:00 定时 + force 同步全量重扫 + 采集落库触发的异步合并重扫。
|
||||
* 数据源 = 采集明细表（biz_shop_data_crawl_item，采集先落库再更新文件），
|
||||
* 直接查库聚合 shops/items/summary 落库 shop_data_duplicate_scan（输出契约不变）。
|
||||
* 双实例通过 Redis 分布式锁防重；扫描失败落 FAILED 行并上抛，force 场景由端点转 409/500。
|
||||
*/
|
||||
@Service
|
||||
@Slf4j
|
||||
public class ShopDataDuplicateCheckScanService {
|
||||
public class ShopDataDuplicateCheckScanService implements DuplicateCheckRefreshPort {
|
||||
|
||||
public static final String SCAN_LOCK = "shop-data-duplicate-check:scan";
|
||||
static final DateTimeFormatter SCANNED_AT_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
||||
@@ -71,6 +73,9 @@ public class ShopDataDuplicateCheckScanService {
|
||||
private final AtomicLong cachedRowId = new AtomicLong(-1L);
|
||||
private volatile CachedScan cachedScan;
|
||||
|
||||
/** 采集落库触发的异步合并重扫调度器（单飞 + 去抖 + 锁忙重试）。 */
|
||||
private final DuplicateCheckRefreshScheduler refreshScheduler;
|
||||
|
||||
@Autowired
|
||||
public ShopDataDuplicateCheckScanService(ShopDataDuplicateScanMapper scanMapper,
|
||||
ShopDuplicateCheckSourceMapper sourceMapper,
|
||||
@@ -86,6 +91,7 @@ public class ShopDataDuplicateCheckScanService {
|
||||
this.objectMapper = objectMapper;
|
||||
this.itemMapper = itemMapper;
|
||||
this.itemStoreService = itemStoreService;
|
||||
this.refreshScheduler = new DuplicateCheckRefreshScheduler(this::runRefreshOnce);
|
||||
}
|
||||
|
||||
/** 读侧视图：scanned_at 为最新 SUCCESS 行 created_at（yyyy-MM-dd HH:mm:ss）。 */
|
||||
@@ -122,6 +128,30 @@ public class ShopDataDuplicateCheckScanService {
|
||||
}
|
||||
}
|
||||
|
||||
/** 采集明细落库后的重扫请求（端口实现）：异步合并执行，不阻塞、不抛错。 */
|
||||
@Override
|
||||
public void requestRefresh(String reason) {
|
||||
refreshScheduler.request(reason);
|
||||
}
|
||||
|
||||
/** 调度器单次扫描动作：锁被占返回 LOCK_BUSY 供其重试；失败只记日志（FAILED 行已落库）。 */
|
||||
private DuplicateCheckRefreshScheduler.Outcome runRefreshOnce() {
|
||||
try {
|
||||
scanNow();
|
||||
return DuplicateCheckRefreshScheduler.Outcome.DONE;
|
||||
} catch (BusinessException ex) {
|
||||
if (ex.getCode() != null && ex.getCode() == 409) {
|
||||
log.info("[shop-duplicate-check] 自动重扫未执行：其它扫描进行中 msg={}", ex.getMessage());
|
||||
return DuplicateCheckRefreshScheduler.Outcome.LOCK_BUSY;
|
||||
}
|
||||
log.warn("[shop-duplicate-check] 自动重扫失败 code={} msg={}", ex.getCode(), ex.getMessage());
|
||||
return DuplicateCheckRefreshScheduler.Outcome.FAILED;
|
||||
} catch (Exception ex) {
|
||||
log.error("[shop-duplicate-check] 自动重扫异常", ex);
|
||||
return DuplicateCheckRefreshScheduler.Outcome.FAILED;
|
||||
}
|
||||
}
|
||||
|
||||
/** 最新 SUCCESS 扫描视图；无扫描结果返回 null。 */
|
||||
public DuplicateScanView loadLatest() {
|
||||
ScanLightRowDto light = scanMapper.selectLatestLightRow();
|
||||
|
||||
+138
@@ -0,0 +1,138 @@
|
||||
package com.nanri.aiimage.modules.shopduplicatecheck.service.support;
|
||||
|
||||
import com.nanri.aiimage.common.util.ThreadPools;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
/**
|
||||
* 采集落库触发的撞款重扫调度器：合并窗口去抖 + 单飞 + 锁忙重试。
|
||||
*
|
||||
* <p>语义：{@link #request} 只置位并异步执行，永不阻塞调用方、永不向外抛错；
|
||||
* 合并窗口内的多次触发聚合为一次扫描；扫描动作执行期间到达的触发在下一轮执行；
|
||||
* 扫描因分布式锁被占用未执行（{@link Outcome#LOCK_BUSY}）时按固定间隔重试有限次。
|
||||
*/
|
||||
@Slf4j
|
||||
public class DuplicateCheckRefreshScheduler {
|
||||
|
||||
/** 单次扫描动作的终态：完成 / 锁被占（可重试）/ 失败（不重试，等下次触发或定时扫描）。 */
|
||||
public enum Outcome {
|
||||
DONE, LOCK_BUSY, FAILED
|
||||
}
|
||||
|
||||
private static final long DEFAULT_DEBOUNCE_MILLIS = 10_000L;
|
||||
private static final long DEFAULT_LOCK_RETRY_MILLIS = 20_000L;
|
||||
private static final int DEFAULT_MAX_LOCK_RETRIES = 6;
|
||||
|
||||
private final Supplier<Outcome> scanAction;
|
||||
private final long debounceMillis;
|
||||
private final long lockRetryMillis;
|
||||
private final int maxLockRetries;
|
||||
private final ExecutorService executor;
|
||||
private final AtomicBoolean pending = new AtomicBoolean(false);
|
||||
private final AtomicBoolean running = new AtomicBoolean(false);
|
||||
|
||||
public DuplicateCheckRefreshScheduler(Supplier<Outcome> scanAction) {
|
||||
this(scanAction, DEFAULT_DEBOUNCE_MILLIS, DEFAULT_LOCK_RETRY_MILLIS, DEFAULT_MAX_LOCK_RETRIES);
|
||||
}
|
||||
|
||||
/** 测试用：注入更短的窗口与重试参数。 */
|
||||
DuplicateCheckRefreshScheduler(Supplier<Outcome> scanAction, long debounceMillis,
|
||||
long lockRetryMillis, int maxLockRetries) {
|
||||
this.scanAction = scanAction;
|
||||
this.debounceMillis = Math.max(0L, debounceMillis);
|
||||
this.lockRetryMillis = Math.max(0L, lockRetryMillis);
|
||||
this.maxLockRetries = Math.max(0, maxLockRetries);
|
||||
this.executor = ThreadPools.boundedFixed("shop-dup-refresh", 1, 8);
|
||||
}
|
||||
|
||||
/** 请求一次重扫（异步、去抖合并）。调用方不被阻塞，也不会收到异常。 */
|
||||
public void request(String reason) {
|
||||
pending.set(true);
|
||||
if (running.compareAndSet(false, true)) {
|
||||
submit(reason);
|
||||
}
|
||||
}
|
||||
|
||||
private void submit(String reason) {
|
||||
try {
|
||||
log.info("[shop-duplicate-check] 触发撞款重扫（异步合并执行，窗口={}ms） reason={}", debounceMillis, reason);
|
||||
executor.execute(this::drain);
|
||||
} catch (Exception ex) {
|
||||
// 提交失败（如线程池拒绝）时复位单飞标记，避免后续触发被永久吞掉
|
||||
running.set(false);
|
||||
log.warn("[shop-duplicate-check] 撞款重扫任务提交失败 reason={} msg={}", reason, ex.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private void drain() {
|
||||
try {
|
||||
while (true) {
|
||||
// 合并窗口：窗口内到达的多次触发聚合为同一轮扫描
|
||||
if (!sleepQuietly(debounceMillis)) {
|
||||
return;
|
||||
}
|
||||
if (!pending.compareAndSet(true, false)) {
|
||||
return;
|
||||
}
|
||||
int lockRetries = 0;
|
||||
while (true) {
|
||||
Outcome outcome = runOnceSafely();
|
||||
if (outcome != Outcome.LOCK_BUSY) {
|
||||
break;
|
||||
}
|
||||
if (lockRetries >= maxLockRetries) {
|
||||
log.warn("[shop-duplicate-check] 撞款重扫连续 {} 次未取得扫描锁，放弃本轮（等待下次触发或定时扫描）",
|
||||
lockRetries + 1);
|
||||
break;
|
||||
}
|
||||
lockRetries++;
|
||||
log.info("[shop-duplicate-check] 撞款重扫未取得扫描锁，{}ms 后重试（第 {}/{} 次）",
|
||||
lockRetryMillis, lockRetries, maxLockRetries);
|
||||
if (!sleepQuietly(lockRetryMillis)) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
running.set(false);
|
||||
// 竞态兜底：running 复位前到达的触发可能没能提交，补一次
|
||||
if (pending.get() && running.compareAndSet(false, true)) {
|
||||
submit("race-guard");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** 执行一次扫描动作；动作自身异常也被吸收（调度器对外零抛出）。 */
|
||||
private Outcome runOnceSafely() {
|
||||
long startedAt = System.currentTimeMillis();
|
||||
try {
|
||||
Outcome outcome = scanAction.get();
|
||||
long elapsed = System.currentTimeMillis() - startedAt;
|
||||
if (outcome == Outcome.DONE) {
|
||||
log.info("[shop-duplicate-check] 采集后自动重扫完成 耗时={}ms", elapsed);
|
||||
} else if (outcome == Outcome.FAILED) {
|
||||
log.warn("[shop-duplicate-check] 采集后自动重扫失败 耗时={}ms", elapsed);
|
||||
}
|
||||
return outcome == null ? Outcome.FAILED : outcome;
|
||||
} catch (Exception ex) {
|
||||
log.error("[shop-duplicate-check] 采集后自动重扫异常", ex);
|
||||
return Outcome.FAILED;
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean sleepQuietly(long millis) {
|
||||
if (millis <= 0) {
|
||||
return true;
|
||||
}
|
||||
try {
|
||||
Thread.sleep(millis);
|
||||
return true;
|
||||
} catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user