super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(D2): 导入进度跨节点可见（NodeSharedStore：本地快路径 + Redis 真源）

Browse Source 
- 新增 common/service/NodeSharedStore：本地 Map 快路径 + Redis 跨节点真源 + 写节流
  （默认 500ms，逐行刷新进度不会打爆 Redis）+ 本节点条目快照（维护用）+ TTL 兜底过期
- 接入 DedupeTotalDataService（8 个进度/归属/分组/完成时间映射）、QueryAsinService、
  SkipPriceAsinService（各 3 个）：轮询落到另一节点不再报"任务不存在"
- 保留期清理改为遍历本节点快照（跨节点过期由 Redis TTL 兜底），不再依赖全量遍历
- 测试同步：去重服务测试的反射注入改用 NodeSharedStore（未注入 Redis 时等价纯本地）

mvn test 2815 全绿
This commit is contained in:
黄自达
2026-09-14 06:54:48 +08:00
parent 9a6b57db58
commit 9ba231dc4c
5 changed files with 188 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/common/service/NodeSharedStore.java
+10
View File
@@ -121,6 +121,16 @@ public final class NodeSharedStore<K, V> {
return local.size();
}
/**
* 本节点已知条目的快照（副本，可安全遍历）。
*
* <p>用途：保留期清理等维护动作需要遍历键；跨节点的过期回收由 Redis TTL 兜底，
* 因此这里只返回本节点写入过的条目即可。
*/
public java.util.Map<K, V> localEntriesSnapshot() {
return new java.util.LinkedHashMap<>(local);
}
private boolean throttleAllowsWrite(K key) {
if (writeThrottleMillis <= 0L) {
return true;
backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java
+69 -19
View File
@@ -2,7 +2,9 @@ package com.nanri.aiimage.modules.dedupe.service;
import cn.hutool.core.util.IdUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.common.service.NodeSharedStore;
import com.nanri.aiimage.common.util.ExcelStreamReader;
import com.nanri.aiimage.modules.dedupe.mapper.DedupeTotalDataMapper;
import com.nanri.aiimage.modules.dedupe.model.dto.DedupeTotalDataCreateRequest;
@@ -63,7 +65,6 @@ import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;
@Service
@RequiredArgsConstructor
@Slf4j
public class DedupeTotalDataService {
@@ -73,18 +74,60 @@ public class DedupeTotalDataService {
private static final long COMPLETED_PROGRESS_RETENTION_MILLIS = 60 * 60 * 1000L;
private static final DateTimeFormatter EXPORT_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
/** 导入进度等状态的 TTL：完成后仍需可查一段时间（与原内存缓存 1 小时保留对齐并留余量）。 */
private static final java.time.Duration IMPORT_STATE_TTL = java.time.Duration.ofHours(3);
private final DedupeTotalDataMapper dedupeTotalDataMapper;
private final AdminUserMapper adminUserMapper;
private final ShopManageGroupMapper shopManageGroupMapper;
private final PlatformTransactionManager transactionManager;
private final Map<String, DedupeTotalDataImportProgressVo> importProgressMap = new ConcurrentHashMap<>();
private final Map<String, DedupeTotalDataImportProgressVo> deleteImportProgressMap = new ConcurrentHashMap<>();
private final Map<String, Long> importOwnerMap = new ConcurrentHashMap<>();
private final Map<String, Long> deleteImportOwnerMap = new ConcurrentHashMap<>();
private final Map<String, Long> importGroupMap = new ConcurrentHashMap<>();
private final Map<String, Long> deleteImportGroupMap = new ConcurrentHashMap<>();
private final Map<String, Long> importCompletedAtMap = new ConcurrentHashMap<>();
private final Map<String, Long> deleteImportCompletedAtMap = new ConcurrentHashMap<>();
// 2026-09 全维度审查 D2：进度/归属/分组改由 NodeSharedStore 承载（本地快路径 + Redis 跨节点），
// 此前只存节点本地内存，轮询落到另一节点会报"任务不存在"（nginx 的 user_id 亲和只覆盖常态）。
private final NodeSharedStore<String, DedupeTotalDataImportProgressVo> importProgressMap;
private final NodeSharedStore<String, DedupeTotalDataImportProgressVo> deleteImportProgressMap;
private final NodeSharedStore<String, Long> importOwnerMap;
private final NodeSharedStore<String, Long> deleteImportOwnerMap;
private final NodeSharedStore<String, Long> importGroupMap;
private final NodeSharedStore<String, Long> deleteImportGroupMap;
private final NodeSharedStore<String, Long> importCompletedAtMap;
private final NodeSharedStore<String, Long> deleteImportCompletedAtMap;
public DedupeTotalDataService(DedupeTotalDataMapper dedupeTotalDataMapper,
AdminUserMapper adminUserMapper,
ShopManageGroupMapper shopManageGroupMapper,
PlatformTransactionManager transactionManager,
org.springframework.data.redis.core.StringRedisTemplate stringRedisTemplate,
ObjectMapper objectMapper) {
this.dedupeTotalDataMapper = dedupeTotalDataMapper;
this.adminUserMapper = adminUserMapper;
this.shopManageGroupMapper = shopManageGroupMapper;
this.transactionManager = transactionManager;
this.importProgressMap = progressStore("aiimage:dedupe:import:progress", stringRedisTemplate, objectMapper);
this.deleteImportProgressMap = progressStore("aiimage:dedupe:delete-import:progress", stringRedisTemplate, objectMapper);
this.importOwnerMap = longStore("aiimage:dedupe:import:owner", stringRedisTemplate, objectMapper);
this.deleteImportOwnerMap = longStore("aiimage:dedupe:delete-import:owner", stringRedisTemplate, objectMapper);
this.importGroupMap = longStore("aiimage:dedupe:import:group", stringRedisTemplate, objectMapper);
this.deleteImportGroupMap = longStore("aiimage:dedupe:delete-import:group", stringRedisTemplate, objectMapper);
// 完成时间只用于本节点保留期清理，不参与跨节点读取 → 不节流
this.importCompletedAtMap = longStoreNoThrottle("aiimage:dedupe:import:completed-at", stringRedisTemplate, objectMapper);
this.deleteImportCompletedAtMap = longStoreNoThrottle("aiimage:dedupe:delete-import:completed-at", stringRedisTemplate, objectMapper);
}
private static NodeSharedStore<String, DedupeTotalDataImportProgressVo> progressStore(
String prefix, org.springframework.data.redis.core.StringRedisTemplate redis, ObjectMapper mapper) {
return new NodeSharedStore<>(prefix, IMPORT_STATE_TTL, DedupeTotalDataImportProgressVo.class, redis, mapper);
}
private static NodeSharedStore<String, Long> longStore(
String prefix, org.springframework.data.redis.core.StringRedisTemplate redis, ObjectMapper mapper) {
return new NodeSharedStore<>(prefix, IMPORT_STATE_TTL, Long.class, redis, mapper);
}
private static NodeSharedStore<String, Long> longStoreNoThrottle(
String prefix, org.springframework.data.redis.core.StringRedisTemplate redis, ObjectMapper mapper) {
return new NodeSharedStore<>(prefix, IMPORT_STATE_TTL, Long.class, redis, mapper, 0L);
}
/** 导入任务并发上限，避免 POI 解析和批量数据库写入叠加。 */
private final Semaphore importSlots = new Semaphore(4);
@@ -1235,19 +1278,26 @@ public class DedupeTotalDataService {
deleteImportGroupMap, cutoff);
}
/**
* 清理已结束且超过保留期的进度条目。
* 只遍历本节点写入过的条目（跨节点的过期由 Redis TTL 兜底），避免为清理做全量 SCAN。
*/
private void cleanupExpiredProgressEntries(
Map<String, Long> completedAtMap,
Map<String, DedupeTotalDataImportProgressVo> progressMap,
Map<String, Long> ownerMap,
Map<String, Long> groupMap,
NodeSharedStore<String, Long> completedAtStore,
NodeSharedStore<String, DedupeTotalDataImportProgressVo> progressStore,
NodeSharedStore<String, Long> ownerStore,
NodeSharedStore<String, Long> groupStore,
long cutoff) {
completedAtMap.forEach((id, completedAt) -> {
if (completedAt != null && completedAt < cutoff && completedAtMap.remove(id, completedAt)) {
progressMap.remove(id);
ownerMap.remove(id);
groupMap.remove(id);
for (Map.Entry<String, Long> entry : completedAtStore.localEntriesSnapshot().entrySet()) {
Long completedAt = entry.getValue();
if (completedAt == null || completedAt >= cutoff) {
continue;
}
});
completedAtStore.remove(entry.getKey());
progressStore.remove(entry.getKey());
ownerStore.remove(entry.getKey());
groupStore.remove(entry.getKey());
}
}
private String normalizeExcelText(String value) {
backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/QueryAsinService.java
+40 -13
View File
@@ -3,7 +3,9 @@ package com.nanri.aiimage.modules.shopkey.service;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.support.SFunction;
import cn.hutool.core.util.IdUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.common.service.NodeSharedStore;
import com.nanri.aiimage.common.util.ExcelStreamReader;
import com.nanri.aiimage.config.TaskPressureProperties;
import com.nanri.aiimage.modules.shopkey.mapper.QueryAsinMapper;
@@ -44,7 +46,6 @@ import java.util.concurrent.Semaphore;
import java.util.function.Predicate;
@Service
@RequiredArgsConstructor
@Slf4j
public class QueryAsinService {
@@ -60,18 +61,39 @@ public class QueryAsinService {
private final QueryAsinMapper queryAsinMapper;
private final ShopManageGroupService shopManageGroupService;
private final TaskPressureProperties taskPressureProperties;
private final Map<String, QueryAsinImportProgressVo> importProgressMap = new ConcurrentHashMap<>();
private final Map<String, QueryAsinImportProgressVo> deleteImportProgressMap = new ConcurrentHashMap<>();
// 2026-09 全维度审查 D2：进度/完成时间改由 NodeSharedStore 承载（本地快路径 + Redis 跨节点），
// 此前只存节点本地内存，轮询落到另一节点会报"任务不存在"。
private final NodeSharedStore<String, QueryAsinImportProgressVo> importProgressMap;
private final NodeSharedStore<String, QueryAsinImportProgressVo> deleteImportProgressMap;
/** 已结束导入的完成时间，用于**本节点**的保留期回收；跨节点过期由 Redis TTL 兜底。 */
private final NodeSharedStore<String, Long> completedImportAtMap;
@Value("${aiimage.shop-key.max-import-file-bytes:104857600}")
private long maxImportFileBytes = 100L * 1024 * 1024;
@Value("${aiimage.shop-key.max-import-rows:500000}")
private int maxImportRows = 500_000;
/** 已结束导入的完成时间，用于定时回收进度快照，避免进程内 Map 无界增长。 */
private final Map<String, Long> completedImportAtMap = new ConcurrentHashMap<>();
/** 导入任务并发上限，避免多个 POI/数据库导入同时拖垮资源。 */
private final Semaphore importSlots = new Semaphore(4);
public QueryAsinService( QueryAsinMapper queryAsinMapper,
ShopManageGroupService shopManageGroupService,
TaskPressureProperties taskPressureProperties,
org.springframework.data.redis.core.StringRedisTemplate stringRedisTemplate,
ObjectMapper objectMapper) {
this.queryAsinMapper = queryAsinMapper;
this.shopManageGroupService = shopManageGroupService;
this.taskPressureProperties = taskPressureProperties;
this.importProgressMap = new NodeSharedStore<>("aiimage:query-asin:import:progress",
IMPORT_STATE_TTL, QueryAsinImportProgressVo.class, stringRedisTemplate, objectMapper);
this.deleteImportProgressMap = new NodeSharedStore<>("aiimage:query-asin:delete-import:progress",
IMPORT_STATE_TTL, QueryAsinImportProgressVo.class, stringRedisTemplate, objectMapper);
this.completedImportAtMap = new NodeSharedStore<>("aiimage:query-asin:import:completed-at",
IMPORT_STATE_TTL, Long.class, stringRedisTemplate, objectMapper, 0L);
}
/** 导入状态在 Redis 的保留时长：完成后仍需可查一段时间（原内存保留 1 小时 + 余量）。 */
private static final java.time.Duration IMPORT_STATE_TTL = java.time.Duration.ofHours(3);
public QueryAsinPageVo page(long page, long pageSize, Long groupId, String shopName, String asin,
String country, Long operatorId, boolean superAdmin) {
long safePage = Math.max(page, 1);
@@ -332,7 +354,8 @@ public class QueryAsinService {
}
String importId = IdUtil.fastSimpleUUID();
QueryAsinImportProgressVo progress = newImportProgress();
Map<String, QueryAsinImportProgressVo> progressMap = deleteMode ? deleteImportProgressMap : importProgressMap;
NodeSharedStore<String, QueryAsinImportProgressVo> progressMap =
deleteMode ? deleteImportProgressMap : importProgressMap;
progressMap.put(importId, progress);
try {
File tempFile = saveMultipartToTempFile(file, deleteMode ? "query-asin-delete-" : "query-asin-import-");
@@ -364,18 +387,22 @@ public class QueryAsinService {
@Scheduled(fixedDelayString = "${aiimage.shop-key.import-progress-cleanup-delay-ms:300000}")
public void cleanupCompletedImports() {
long cutoff = System.currentTimeMillis() - COMPLETED_IMPORT_RETENTION_MILLIS;
completedImportAtMap.forEach((importId, completedAt) -> {
if (completedAt != null && completedAt < cutoff
&& completedImportAtMap.remove(importId, completedAt)) {
importProgressMap.remove(importId);
deleteImportProgressMap.remove(importId);
// 只遍历本节点写入过的条目（跨节点过期由 Redis TTL 兜底），避免为清理做全量 SCAN
for (Map.Entry<String, Long> entry : completedImportAtMap.localEntriesSnapshot().entrySet()) {
Long completedAt = entry.getValue();
if (completedAt == null || completedAt >= cutoff) {
continue;
}
});
completedImportAtMap.remove(entry.getKey());
importProgressMap.remove(entry.getKey());
deleteImportProgressMap.remove(entry.getKey());
}
}
private void runImportTask(String importId, File tempFile, String filename, Long fallbackGroupId,
Long operatorId, boolean superAdmin, boolean deleteMode) {
Map<String, QueryAsinImportProgressVo> progressMap = deleteMode ? deleteImportProgressMap : importProgressMap;
NodeSharedStore<String, QueryAsinImportProgressVo> progressMap =
deleteMode ? deleteImportProgressMap : importProgressMap;
QueryAsinImportProgressVo progress = progressMap.get(importId);
if (progress == null) {
deleteQuietly(tempFile);
backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/SkipPriceAsinService.java
+40 -13
View File
@@ -4,7 +4,9 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.support.SFunction;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import cn.hutool.core.util.IdUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.common.service.NodeSharedStore;
import com.nanri.aiimage.common.util.ExcelStreamReader;
import com.nanri.aiimage.modules.shopkey.mapper.SkipPriceAsinMapper;
import com.nanri.aiimage.modules.shopkey.mapper.ShopManageMapper;
@@ -46,7 +48,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
@Service
@RequiredArgsConstructor
@Slf4j
public class SkipPriceAsinService {
@@ -60,17 +61,38 @@ public class SkipPriceAsinService {
private final SkipPriceAsinMapper skipPriceAsinMapper;
private final ShopManageMapper shopManageMapper;
private final ShopManageGroupService shopManageGroupService;
private final Map<String, QueryAsinImportProgressVo> importProgressMap = new ConcurrentHashMap<>();
private final Map<String, QueryAsinImportProgressVo> deleteImportProgressMap = new ConcurrentHashMap<>();
// 2026-09 全维度审查 D2：进度/完成时间改由 NodeSharedStore 承载（本地快路径 + Redis 跨节点），
// 此前只存节点本地内存，轮询落到另一节点会报"任务不存在"。
private final NodeSharedStore<String, QueryAsinImportProgressVo> importProgressMap;
private final NodeSharedStore<String, QueryAsinImportProgressVo> deleteImportProgressMap;
/** 已结束导入的完成时间，用于**本节点**的保留期回收；跨节点过期由 Redis TTL 兜底。 */
private final NodeSharedStore<String, Long> completedImportAtMap;
@Value("${aiimage.shop-key.max-import-file-bytes:104857600}")
private long maxImportFileBytes = 100L * 1024 * 1024;
@Value("${aiimage.shop-key.max-import-rows:500000}")
private int maxImportRows = 500_000;
/** 已结束导入的完成时间，用于定时回收进度快照，避免进程内 Map 无界增长。 */
private final Map<String, Long> completedImportAtMap = new ConcurrentHashMap<>();
/** 导入任务并发上限，避免多个 POI/数据库导入同时拖垮资源。 */
private final Semaphore importSlots = new Semaphore(4);
public SkipPriceAsinService( SkipPriceAsinMapper skipPriceAsinMapper,
ShopManageMapper shopManageMapper,
ShopManageGroupService shopManageGroupService,
org.springframework.data.redis.core.StringRedisTemplate stringRedisTemplate,
ObjectMapper objectMapper) {
this.skipPriceAsinMapper = skipPriceAsinMapper;
this.shopManageMapper = shopManageMapper;
this.shopManageGroupService = shopManageGroupService;
this.importProgressMap = new NodeSharedStore<>("aiimage:skip-price-asin:import:progress",
IMPORT_STATE_TTL, QueryAsinImportProgressVo.class, stringRedisTemplate, objectMapper);
this.deleteImportProgressMap = new NodeSharedStore<>("aiimage:skip-price-asin:delete-import:progress",
IMPORT_STATE_TTL, QueryAsinImportProgressVo.class, stringRedisTemplate, objectMapper);
this.completedImportAtMap = new NodeSharedStore<>("aiimage:skip-price-asin:import:completed-at",
IMPORT_STATE_TTL, Long.class, stringRedisTemplate, objectMapper, 0L);
}
/** 导入状态在 Redis 的保留时长：完成后仍需可查一段时间（原内存保留 1 小时 + 余量）。 */
private static final java.time.Duration IMPORT_STATE_TTL = java.time.Duration.ofHours(3);
private final Map<String, CachedSkipAsinLookup> skipAsinLookupCache = new ConcurrentHashMap<>();
public SkipPriceAsinPageVo page(long page, long pageSize, Long groupId, String shopName, String asin,
@@ -410,7 +432,8 @@ public class SkipPriceAsinService {
}
String importId = IdUtil.fastSimpleUUID();
QueryAsinImportProgressVo progress = newImportProgress();
Map<String, QueryAsinImportProgressVo> progressMap = deleteMode ? deleteImportProgressMap : importProgressMap;
NodeSharedStore<String, QueryAsinImportProgressVo> progressMap =
deleteMode ? deleteImportProgressMap : importProgressMap;
progressMap.put(importId, progress);
try {
File tempFile = saveMultipartToTempFile(file, deleteMode ? "skip-price-asin-delete-" : "skip-price-asin-import-");
@@ -442,18 +465,22 @@ public class SkipPriceAsinService {
@Scheduled(fixedDelayString = "${aiimage.shop-key.import-progress-cleanup-delay-ms:300000}")
public void cleanupCompletedImports() {
long cutoff = System.currentTimeMillis() - COMPLETED_IMPORT_RETENTION_MILLIS;
completedImportAtMap.forEach((importId, completedAt) -> {
if (completedAt != null && completedAt < cutoff
&& completedImportAtMap.remove(importId, completedAt)) {
importProgressMap.remove(importId);
deleteImportProgressMap.remove(importId);
// 只遍历本节点写入过的条目（跨节点过期由 Redis TTL 兜底），避免为清理做全量 SCAN
for (Map.Entry<String, Long> entry : completedImportAtMap.localEntriesSnapshot().entrySet()) {
Long completedAt = entry.getValue();
if (completedAt == null || completedAt >= cutoff) {
continue;
}
});
completedImportAtMap.remove(entry.getKey());
importProgressMap.remove(entry.getKey());
deleteImportProgressMap.remove(entry.getKey());
}
}
private void runImportTask(String importId, File tempFile, String filename, Long groupId,
String shopName, boolean deleteMode) {
Map<String, QueryAsinImportProgressVo> progressMap = deleteMode ? deleteImportProgressMap : importProgressMap;
NodeSharedStore<String, QueryAsinImportProgressVo> progressMap =
deleteMode ? deleteImportProgressMap : importProgressMap;
QueryAsinImportProgressVo progress = progressMap.get(importId);
if (progress == null) {
deleteQuietly(tempFile);
backend-java/src/test/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataServiceTest.java
+29 -21
View File
@@ -7,6 +7,7 @@ import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.modules.dedupe.mapper.DedupeTotalDataMapper;
import com.nanri.aiimage.modules.dedupe.model.dto.DedupeTotalDataCreateRequest;
import com.nanri.aiimage.modules.dedupe.model.entity.DedupeTotalDataEntity;
import com.nanri.aiimage.common.service.NodeSharedStore;
import com.nanri.aiimage.modules.dedupe.model.vo.DedupeTotalDataImportProgressVo;
import com.nanri.aiimage.modules.dedupe.model.vo.DedupeTotalDataItemVo;
import com.nanri.aiimage.modules.dedupe.model.vo.DedupeTotalDataPageVo;
@@ -481,38 +482,45 @@ class DedupeTotalDataServiceTest {
@Test
@SuppressWarnings("unchecked")
void expiredCompletedProgressIsRemovedOnNextLookup() {
Map<String, DedupeTotalDataImportProgressVo> progressMap =
(Map<String, DedupeTotalDataImportProgressVo>) ReflectionTestUtils.getField(service, "importProgressMap");
Map<String, Long> ownerMap = (Map<String, Long>) ReflectionTestUtils.getField(service, "importOwnerMap");
Map<String, Long> groupMap = (Map<String, Long>) ReflectionTestUtils.getField(service, "importGroupMap");
Map<String, Long> completedAtMap =
(Map<String, Long>) ReflectionTestUtils.getField(service, "importCompletedAtMap");
NodeSharedStore<String, DedupeTotalDataImportProgressVo> progressStore =
(NodeSharedStore<String, DedupeTotalDataImportProgressVo>) ReflectionTestUtils.getField(service, "importProgressMap");
NodeSharedStore<String, Long> ownerStore =
(NodeSharedStore<String, Long>) ReflectionTestUtils.getField(service, "importOwnerMap");
NodeSharedStore<String, Long> groupStore =
(NodeSharedStore<String, Long>) ReflectionTestUtils.getField(service, "importGroupMap");
NodeSharedStore<String, Long> completedAtStore =
(NodeSharedStore<String, Long>) ReflectionTestUtils.getField(service, "importCompletedAtMap");
DedupeTotalDataImportProgressVo progress = new DedupeTotalDataImportProgressVo();
progress.setStatus("success");
progressMap.put("expired", progress);
ownerMap.put("expired", 23L);
groupMap.put("expired", 7L);
completedAtMap.put("expired", System.currentTimeMillis() - (2 * 60 * 60 * 1000L));
progressStore.put("expired", progress);
ownerStore.put("expired", 23L);
groupStore.put("expired", 7L);
completedAtStore.put("expired", System.currentTimeMillis() - (2 * 60 * 60 * 1000L));
assertThrows(BusinessException.class, () -> service.getImportProgress("expired", 23L));
assertFalse(progressMap.containsKey("expired"));
assertFalse(ownerMap.containsKey("expired"));
assertFalse(groupMap.containsKey("expired"));
assertFalse(completedAtMap.containsKey("expired"));
assertFalse(progressStore.containsKey("expired"));
assertFalse(ownerStore.containsKey("expired"));
assertFalse(groupStore.containsKey("expired"));
assertFalse(completedAtStore.containsKey("expired"));
}
/**
* 直接注入导入进度状态（D2 后由 NodeSharedStore 承载：测试里未注入 Redis，等价于纯本地）。
*/
@SuppressWarnings("unchecked")
private DedupeTotalDataImportProgressVo registerImportProgress(String importId, Long ownerId, Long groupId) {
Map<String, DedupeTotalDataImportProgressVo> progressMap =
(Map<String, DedupeTotalDataImportProgressVo>) ReflectionTestUtils.getField(service, "importProgressMap");
Map<String, Long> ownerMap = (Map<String, Long>) ReflectionTestUtils.getField(service, "importOwnerMap");
Map<String, Long> groupMap = (Map<String, Long>) ReflectionTestUtils.getField(service, "importGroupMap");
NodeSharedStore<String, DedupeTotalDataImportProgressVo> progressStore =
(NodeSharedStore<String, DedupeTotalDataImportProgressVo>) ReflectionTestUtils.getField(service, "importProgressMap");
NodeSharedStore<String, Long> ownerStore =
(NodeSharedStore<String, Long>) ReflectionTestUtils.getField(service, "importOwnerMap");
NodeSharedStore<String, Long> groupStore =
(NodeSharedStore<String, Long>) ReflectionTestUtils.getField(service, "importGroupMap");
DedupeTotalDataImportProgressVo progress = new DedupeTotalDataImportProgressVo();
progress.setStatus("running");
progressMap.put(importId, progress);
ownerMap.put(importId, ownerId);
groupMap.put(importId, groupId);
progressStore.put(importId, progress);
ownerStore.put(importId, ownerId);
groupStore.put(importId, groupId);
return progress;
}