|
@@ -13,6 +13,9 @@ import com.jeeplus.assess.materialPrice.domain.MaterialPriceRecord;
|
|
|
import com.jeeplus.assess.materialPrice.domain.MaterialPriceSyncLog;
|
|
import com.jeeplus.assess.materialPrice.domain.MaterialPriceSyncLog;
|
|
|
import com.jeeplus.assess.materialPrice.enums.MaterialPriceSyncStatus;
|
|
import com.jeeplus.assess.materialPrice.enums.MaterialPriceSyncStatus;
|
|
|
import com.jeeplus.assess.materialPrice.mapper.MaterialPriceSyncLogMapper;
|
|
import com.jeeplus.assess.materialPrice.mapper.MaterialPriceSyncLogMapper;
|
|
|
|
|
+import com.jeeplus.assess.materialPrice.service.support.RangeChunk;
|
|
|
|
|
+import com.jeeplus.assess.materialPrice.service.support.RangeProgress;
|
|
|
|
|
+import com.jeeplus.assess.materialPrice.service.support.SiteSummary;
|
|
|
import com.jeeplus.common.SecurityUtils;
|
|
import com.jeeplus.common.SecurityUtils;
|
|
|
import com.jeeplus.common.constant.CommonConstants;
|
|
import com.jeeplus.common.constant.CommonConstants;
|
|
|
import com.jeeplus.common.redis.RedisUtils;
|
|
import com.jeeplus.common.redis.RedisUtils;
|
|
@@ -34,17 +37,14 @@ import java.util.ArrayList;
|
|
|
import java.util.Collections;
|
|
import java.util.Collections;
|
|
|
import java.util.Date;
|
|
import java.util.Date;
|
|
|
import java.util.LinkedHashMap;
|
|
import java.util.LinkedHashMap;
|
|
|
-import java.util.LinkedHashSet;
|
|
|
|
|
import java.util.List;
|
|
import java.util.List;
|
|
|
import java.util.Map;
|
|
import java.util.Map;
|
|
|
-import java.util.Set;
|
|
|
|
|
import java.util.concurrent.CompletableFuture;
|
|
import java.util.concurrent.CompletableFuture;
|
|
|
import java.util.concurrent.Executor;
|
|
import java.util.concurrent.Executor;
|
|
|
import java.util.concurrent.ScheduledExecutorService;
|
|
import java.util.concurrent.ScheduledExecutorService;
|
|
|
import java.util.concurrent.ScheduledFuture;
|
|
import java.util.concurrent.ScheduledFuture;
|
|
|
import java.util.concurrent.TimeUnit;
|
|
import java.util.concurrent.TimeUnit;
|
|
|
import java.util.concurrent.atomic.AtomicBoolean;
|
|
import java.util.concurrent.atomic.AtomicBoolean;
|
|
|
-import java.util.concurrent.atomic.AtomicInteger;
|
|
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
* 材料价格同步服务:Java 侧唯一调用 Python 抓取服务的入口。
|
|
* 材料价格同步服务:Java 侧唯一调用 Python 抓取服务的入口。
|
|
@@ -74,10 +74,7 @@ public class MaterialPriceSyncService {
|
|
|
/** 区间采集锁:与 XXL 锁分开,两种采集可以同时进行(写入幂等,不会冲突) */
|
|
/** 区间采集锁:与 XXL 锁分开,两种采集可以同时进行(写入幂等,不会冲突) */
|
|
|
private static final String RANGE_LOCK_KEY = "material-price:sync:range";
|
|
private static final String RANGE_LOCK_KEY = "material-price:sync:range";
|
|
|
|
|
|
|
|
- /** 单个站点在日志里保留的提示条数上限,避免异常时 site_result 字段过大 */
|
|
|
|
|
- private static final int NOTES_LIMIT = 100;
|
|
|
|
|
-
|
|
|
|
|
- /** 单个区间分片最多覆盖的天数:约 6 天 × 6000 个指数 ≈ 3.7 万条,单次响应可控 */
|
|
|
|
|
|
|
+ /** 单个区间分片最多覆盖的天数:约 6 天 × 数千个指数,单次响应可控 */
|
|
|
private static final int FFHSW_DAYS_PER_SLICE = 6;
|
|
private static final int FFHSW_DAYS_PER_SLICE = 6;
|
|
|
|
|
|
|
|
/** 区间分片入库时每批转换/写入的条数 */
|
|
/** 区间分片入库时每批转换/写入的条数 */
|
|
@@ -96,10 +93,8 @@ public class MaterialPriceSyncService {
|
|
|
/** 每次重试前的等待秒数,给站点限流一点恢复时间 */
|
|
/** 每次重试前的等待秒数,给站点限流一点恢复时间 */
|
|
|
private int rangeRetryBackoffSeconds=30;
|
|
private int rangeRetryBackoffSeconds=30;
|
|
|
|
|
|
|
|
- /**
|
|
|
|
|
- * 单个分片调 Python 的读超时(毫秒)。
|
|
|
|
|
- */
|
|
|
|
|
- private int rangeTimeoutMs=300000;
|
|
|
|
|
|
|
+ /**单个分片调 Python 的读超时(毫秒)。*/
|
|
|
|
|
+ private int rangeTimeoutMs=600000;
|
|
|
|
|
|
|
|
/** 整个区间任务的最长执行时间(秒),超时后剩余分片直接记失败,避免无限重试 */
|
|
/** 整个区间任务的最长执行时间(秒),超时后剩余分片直接记失败,避免无限重试 */
|
|
|
private int rangeMaxSeconds=7200;
|
|
private int rangeMaxSeconds=7200;
|
|
@@ -115,38 +110,29 @@ public class MaterialPriceSyncService {
|
|
|
/**
|
|
/**
|
|
|
* 废废网按指数切片的总片数上限,调大可以让单片更轻、片数更多。
|
|
* 废废网按指数切片的总片数上限,调大可以让单片更轻、片数更多。
|
|
|
*
|
|
*
|
|
|
- * 全站约 6000 个指数,180 天约 90 万条记录,一次返回会把 Python/Java 内存打爆,
|
|
|
|
|
|
|
+ * 目标省份合计数千个指数,180 天可达数十万条记录,一次返回会把 Python/Java 内存打爆,
|
|
|
* 因此按「指数 id 取模」切成多片逐片调用;实际片数 = ceil(天数 / 6),且不超过本值。
|
|
* 因此按「指数 id 取模」切成多片逐片调用;实际片数 = ceil(天数 / 6),且不超过本值。
|
|
|
* 例:180 天 30 片、151 天 26 片、3 天 1 片。
|
|
* 例:180 天 30 片、151 天 26 片、3 天 1 片。
|
|
|
*/
|
|
*/
|
|
|
private int ffhswSlices = 30;
|
|
private int ffhswSlices = 30;
|
|
|
|
|
|
|
|
- /**
|
|
|
|
|
- * 回收商网分片天数
|
|
|
|
|
- */
|
|
|
|
|
|
|
+ /** 回收商网分片天数*/
|
|
|
private int chunkDaysHss = 10;
|
|
private int chunkDaysHss = 10;
|
|
|
|
|
|
|
|
- /**
|
|
|
|
|
- * 废钢资讯网分片天数
|
|
|
|
|
- */
|
|
|
|
|
|
|
+ /** 废钢资讯网分片天数*/
|
|
|
private int chunkDaysFeigang = 50;
|
|
private int chunkDaysFeigang = 50;
|
|
|
|
|
|
|
|
- /**
|
|
|
|
|
- * 保太集团分片天数
|
|
|
|
|
- */
|
|
|
|
|
|
|
+ /** 保太集团分片天数*/
|
|
|
private int chunkDaysBaotai = 50;
|
|
private int chunkDaysBaotai = 50;
|
|
|
|
|
|
|
|
- /**
|
|
|
|
|
- * 中果废品网分片天数它只抓首页最近十几篇文章
|
|
|
|
|
- */
|
|
|
|
|
|
|
+ /** 中果废品网分片天数它只抓首页最近十几篇文章*/
|
|
|
private int chunkDaysZgfp = 300;
|
|
private int chunkDaysZgfp = 300;
|
|
|
|
|
|
|
|
/** 区间采集锁的过期时间(秒) */
|
|
/** 区间采集锁的过期时间(秒) */
|
|
|
private long rangeLockSeconds=300;
|
|
private long rangeLockSeconds=300;
|
|
|
|
|
|
|
|
/** 区间采集锁的续期间隔(秒),必须小于锁过期时间 */
|
|
/** 区间采集锁的续期间隔(秒),必须小于锁过期时间 */
|
|
|
- @Value("${material-price.range.lock-renew-seconds:60}")
|
|
|
|
|
- private long lockRenewSeconds;
|
|
|
|
|
|
|
+ private long lockRenewSeconds=60;
|
|
|
|
|
|
|
|
// ==================== 依赖注入 ====================
|
|
// ==================== 依赖注入 ====================
|
|
|
|
|
|
|
@@ -451,7 +437,7 @@ public class MaterialPriceSyncService {
|
|
|
logEntity.setCurrentStage("排队中");
|
|
logEntity.setCurrentStage("排队中");
|
|
|
syncLogMapper.insert(logEntity);
|
|
syncLogMapper.insert(logEntity);
|
|
|
|
|
|
|
|
- RangeProgress progress = new RangeProgress(logEntity.getId(), operator, totalChunks);
|
|
|
|
|
|
|
+ RangeProgress progress = new RangeProgress(syncLogMapper, logEntity.getId(), operator, totalChunks);
|
|
|
try {
|
|
try {
|
|
|
rangeExecutor.execute(() -> runRangeJob(
|
|
rangeExecutor.execute(() -> runRangeJob(
|
|
|
batchNo, logEntity.getId(), start, end, plan, totalChunks, operator, progress));
|
|
batchNo, logEntity.getId(), start, end, plan, totalChunks, operator, progress));
|
|
@@ -520,15 +506,15 @@ public class MaterialPriceSyncService {
|
|
|
SiteSummary summary = entry.getValue().get();
|
|
SiteSummary summary = entry.getValue().get();
|
|
|
summaries.add(summary);
|
|
summaries.add(summary);
|
|
|
log.info("区间采集站点结束 batchNo={} site={} chunks={} success={} noData={} failed={} saved={} dataDate={}",
|
|
log.info("区间采集站点结束 batchNo={} site={} chunks={} success={} noData={} failed={} saved={} dataDate={}",
|
|
|
- batchNo, siteKey, summary.chunks, summary.successChunks,
|
|
|
|
|
- summary.noDataChunks, summary.failedChunks, summary.saved, summary.dataDate);
|
|
|
|
|
|
|
+ batchNo, siteKey, summary.getChunks(), summary.getSuccessChunks(),
|
|
|
|
|
+ summary.getNoDataChunks(), summary.getFailedChunks(), summary.getSaved(), summary.getDataDate());
|
|
|
} catch (Exception e) {
|
|
} catch (Exception e) {
|
|
|
// CompletableFuture 里的异常到这里才拿得到,补一条失败摘要保证统计完整
|
|
// CompletableFuture 里的异常到这里才拿得到,补一条失败摘要保证统计完整
|
|
|
log.error("区间采集站点任务异常 batchNo={} site={}", batchNo, siteKey, e);
|
|
log.error("区间采集站点任务异常 batchNo={} site={}", batchNo, siteKey, e);
|
|
|
SiteSummary summary = new SiteSummary(siteKey, siteName(siteKey));
|
|
SiteSummary summary = new SiteSummary(siteKey, siteName(siteKey));
|
|
|
- summary.chunks = plan.get(siteKey).size();
|
|
|
|
|
- summary.failedChunks = summary.chunks;
|
|
|
|
|
- summary.error = "站点任务异常: " + e.getMessage();
|
|
|
|
|
|
|
+ summary.setChunks(plan.get(siteKey).size());
|
|
|
|
|
+ summary.addFailedChunks(summary.getChunks());
|
|
|
|
|
+ summary.setError("站点任务异常: " + e.getMessage());
|
|
|
summaries.add(summary);
|
|
summaries.add(summary);
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
@@ -542,14 +528,14 @@ public class MaterialPriceSyncService {
|
|
|
int failedChunks = 0;
|
|
int failedChunks = 0;
|
|
|
LocalDate dataDate = null;
|
|
LocalDate dataDate = null;
|
|
|
for (SiteSummary summary : summaries) {
|
|
for (SiteSummary summary : summaries) {
|
|
|
- totalCount += summary.total;
|
|
|
|
|
- saved += summary.saved;
|
|
|
|
|
- invalid += summary.invalid;
|
|
|
|
|
- successChunks += summary.successChunks;
|
|
|
|
|
- noDataChunks += summary.noDataChunks;
|
|
|
|
|
- failedChunks += summary.failedChunks;
|
|
|
|
|
- if (summary.dataDate != null && (dataDate == null || summary.dataDate.isAfter(dataDate))) {
|
|
|
|
|
- dataDate = summary.dataDate;
|
|
|
|
|
|
|
+ totalCount += summary.getTotal();
|
|
|
|
|
+ saved += summary.getSaved();
|
|
|
|
|
+ invalid += summary.getInvalid();
|
|
|
|
|
+ successChunks += summary.getSuccessChunks();
|
|
|
|
|
+ noDataChunks += summary.getNoDataChunks();
|
|
|
|
|
+ failedChunks += summary.getFailedChunks();
|
|
|
|
|
+ if (summary.getDataDate() != null && (dataDate == null || summary.getDataDate().isAfter(dataDate))) {
|
|
|
|
|
+ dataDate = summary.getDataDate();
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -591,13 +577,13 @@ public class MaterialPriceSyncService {
|
|
|
private SiteSummary runSite(String batchNo, String siteKey, List<RangeChunk> chunks,
|
|
private SiteSummary runSite(String batchNo, String siteKey, List<RangeChunk> chunks,
|
|
|
String operator, RangeProgress progress, long deadline) {
|
|
String operator, RangeProgress progress, long deadline) {
|
|
|
SiteSummary summary = new SiteSummary(siteKey, siteName(siteKey));
|
|
SiteSummary summary = new SiteSummary(siteKey, siteName(siteKey));
|
|
|
- summary.chunks = chunks.size();
|
|
|
|
|
|
|
+ summary.setChunks(chunks.size());
|
|
|
for (RangeChunk chunk : chunks) {
|
|
for (RangeChunk chunk : chunks) {
|
|
|
if (System.currentTimeMillis() > deadline) {
|
|
if (System.currentTimeMillis() > deadline) {
|
|
|
- int remain = chunks.size() - summary.successChunks - summary.noDataChunks - summary.failedChunks;
|
|
|
|
|
|
|
+ int remain = chunks.size() - summary.getSuccessChunks() - summary.getNoDataChunks() - summary.getFailedChunks();
|
|
|
log.warn("区间采集超过总时长上限 batchNo={} site={} 剩余 {} 个分片记为失败", batchNo, siteKey, remain);
|
|
log.warn("区间采集超过总时长上限 batchNo={} site={} 剩余 {} 个分片记为失败", batchNo, siteKey, remain);
|
|
|
- summary.failedChunks += remain;
|
|
|
|
|
- summary.notes.add("超过区间任务总时长上限(" + rangeMaxSeconds + "s),剩余 " + remain + " 个分片未执行");
|
|
|
|
|
|
|
+ summary.addFailedChunks(remain);
|
|
|
|
|
+ summary.addNote("超过区间任务总时长上限(" + rangeMaxSeconds + "s),剩余 " + remain + " 个分片未执行");
|
|
|
break;
|
|
break;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -632,29 +618,29 @@ public class MaterialPriceSyncService {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
if (response == null) {
|
|
if (response == null) {
|
|
|
- summary.failedChunks++;
|
|
|
|
|
- summary.notes.add(chunk.label() + " 抓取失败: " + lastError);
|
|
|
|
|
|
|
+ summary.addFailedChunk();
|
|
|
|
|
+ summary.addNote(chunk.label() + " 抓取失败: " + lastError);
|
|
|
progress.chunkDone();
|
|
progress.chunkDone();
|
|
|
continue;
|
|
continue;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- int savedBefore = summary.saved;
|
|
|
|
|
- int invalidBefore = summary.invalid;
|
|
|
|
|
|
|
+ int savedBefore = summary.getSaved();
|
|
|
|
|
+ int invalidBefore = summary.getInvalid();
|
|
|
try {
|
|
try {
|
|
|
applyChunkResponse(summary, batchNo, operator, chunk, response, chunkBegin);
|
|
applyChunkResponse(summary, batchNo, operator, chunk, response, chunkBegin);
|
|
|
} catch (Exception e) {
|
|
} catch (Exception e) {
|
|
|
// 入库或清洗异常只影响当前分片
|
|
// 入库或清洗异常只影响当前分片
|
|
|
log.error("区间采集分片处理异常 batchNo={} site={} chunk={}", batchNo, siteKey, chunk.label(), e);
|
|
log.error("区间采集分片处理异常 batchNo={} site={} chunk={}", batchNo, siteKey, chunk.label(), e);
|
|
|
- summary.failedChunks++;
|
|
|
|
|
- summary.notes.add(chunk.label() + " 处理失败: " + e.getMessage());
|
|
|
|
|
|
|
+ summary.addFailedChunk();
|
|
|
|
|
+ summary.addNote(chunk.label() + " 处理失败: " + e.getMessage());
|
|
|
}
|
|
}
|
|
|
// 只把本分片的增量写进进度,避免多站点并发覆盖彼此的数字
|
|
// 只把本分片的增量写进进度,避免多站点并发覆盖彼此的数字
|
|
|
- progress.addCounts(summary.saved - savedBefore, summary.invalid - invalidBefore);
|
|
|
|
|
|
|
+ progress.addCounts(summary.getSaved() - savedBefore, summary.getInvalid() - invalidBefore);
|
|
|
progress.chunkDone();
|
|
progress.chunkDone();
|
|
|
}
|
|
}
|
|
|
log.info("区间采集站点完成 batchNo={} site={} chunks={} success={} noData={} failed={} total={} saved={} invalid={}",
|
|
log.info("区间采集站点完成 batchNo={} site={} chunks={} success={} noData={} failed={} total={} saved={} invalid={}",
|
|
|
- batchNo, siteKey, summary.chunks, summary.successChunks, summary.noDataChunks,
|
|
|
|
|
- summary.failedChunks, summary.total, summary.saved, summary.invalid);
|
|
|
|
|
|
|
+ batchNo, siteKey, summary.getChunks(), summary.getSuccessChunks(), summary.getNoDataChunks(),
|
|
|
|
|
+ summary.getFailedChunks(), summary.getTotal(), summary.getSaved(), summary.getInvalid());
|
|
|
return summary;
|
|
return summary;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -680,9 +666,9 @@ public class MaterialPriceSyncService {
|
|
|
saved += recordService.upsertBatch(records);
|
|
saved += recordService.upsertBatch(records);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- summary.total += response.getTotal() == null ? items.size() : response.getTotal();
|
|
|
|
|
- summary.saved += saved;
|
|
|
|
|
- summary.invalid += invalid;
|
|
|
|
|
|
|
+ summary.addTotal(response.getTotal() == null ? items.size() : response.getTotal());
|
|
|
|
|
+ summary.addSaved(saved);
|
|
|
|
|
+ summary.addInvalid(invalid);
|
|
|
|
|
|
|
|
// 解析 Python 返回的抓取器状态,决定这个分片算成功 / 无数据 / 失败
|
|
// 解析 Python 返回的抓取器状态,决定这个分片算成功 / 无数据 / 失败
|
|
|
List<MaterialPriceScraperResultDTO> scrapers = safe(response.getScrapers());
|
|
List<MaterialPriceScraperResultDTO> scrapers = safe(response.getScrapers());
|
|
@@ -691,32 +677,29 @@ public class MaterialPriceSyncService {
|
|
|
for (MaterialPriceScraperResultDTO scraper : scrapers) {
|
|
for (MaterialPriceScraperResultDTO scraper : scrapers) {
|
|
|
if (MaterialPriceSyncStatus.SUCCESS.name().equals(scraper.getStatus())) {
|
|
if (MaterialPriceSyncStatus.SUCCESS.name().equals(scraper.getStatus())) {
|
|
|
hasSuccess = true;
|
|
hasSuccess = true;
|
|
|
- LocalDate dataDate = parseLocalDate(scraper.getDataDate());
|
|
|
|
|
- if (dataDate != null && (summary.dataDate == null || dataDate.isAfter(summary.dataDate))) {
|
|
|
|
|
- summary.dataDate = dataDate;
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ summary.updateDataDate(parseLocalDate(scraper.getDataDate()));
|
|
|
} else if (MaterialPriceSyncStatus.FAILED.name().equals(scraper.getStatus())) {
|
|
} else if (MaterialPriceSyncStatus.FAILED.name().equals(scraper.getStatus())) {
|
|
|
hasFailed = true;
|
|
hasFailed = true;
|
|
|
}
|
|
}
|
|
|
// notes 是站点内部非致命提示(如部分分类失败),error 是失败原因;
|
|
// notes 是站点内部非致命提示(如部分分类失败),error 是失败原因;
|
|
|
- // 单个站点最多保留 100 条,防止异常时 site_result 字段被撑爆
|
|
|
|
|
|
|
+ // 条数上限由 SiteSummary 内部控制,防止异常时 site_result 字段被撑爆
|
|
|
for (String note : safe(scraper.getNotes())) {
|
|
for (String note : safe(scraper.getNotes())) {
|
|
|
- if (StringUtils.isNotBlank(note) && summary.notes.size() < NOTES_LIMIT) {
|
|
|
|
|
- summary.notes.add(scraper.getScraperKey() + ": " + note);
|
|
|
|
|
|
|
+ if (StringUtils.isNotBlank(note)) {
|
|
|
|
|
+ summary.addNote(scraper.getScraperKey() + ": " + note);
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
- if (StringUtils.isNotBlank(scraper.getError()) && summary.notes.size() < NOTES_LIMIT) {
|
|
|
|
|
- summary.notes.add(scraper.getScraperKey() + " 失败: " + scraper.getError());
|
|
|
|
|
|
|
+ if (StringUtils.isNotBlank(scraper.getError())) {
|
|
|
|
|
+ summary.addNote(scraper.getScraperKey() + " 失败: " + scraper.getError());
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
if (hasFailed && !hasSuccess) {
|
|
if (hasFailed && !hasSuccess) {
|
|
|
- summary.failedChunks++;
|
|
|
|
|
|
|
+ summary.addFailedChunk();
|
|
|
} else if (hasSuccess) {
|
|
} else if (hasSuccess) {
|
|
|
- summary.successChunks++;
|
|
|
|
|
|
|
+ summary.addSuccessChunk();
|
|
|
} else {
|
|
} else {
|
|
|
// Python 对「没有该日期数据」「站点不支持历史」都返回 NO_DATA,属于正常跳过
|
|
// Python 对「没有该日期数据」「站点不支持历史」都返回 NO_DATA,属于正常跳过
|
|
|
- summary.noDataChunks++;
|
|
|
|
|
|
|
+ summary.addNoDataChunk();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
log.info("区间采集分片结束 batchNo={} chunk={} elapsedMs={} items={} saved={} invalid={} "
|
|
log.info("区间采集分片结束 batchNo={} chunk={} elapsedMs={} items={} saved={} invalid={} "
|
|
@@ -753,18 +736,18 @@ public class MaterialPriceSyncService {
|
|
|
List<Map<String, Object>> results = new ArrayList<>(summaries.size());
|
|
List<Map<String, Object>> results = new ArrayList<>(summaries.size());
|
|
|
for (SiteSummary summary : summaries) {
|
|
for (SiteSummary summary : summaries) {
|
|
|
Map<String, Object> item = new LinkedHashMap<>();
|
|
Map<String, Object> item = new LinkedHashMap<>();
|
|
|
- item.put("siteKey", summary.siteKey);
|
|
|
|
|
- item.put("siteName", summary.siteName);
|
|
|
|
|
- item.put("chunks", summary.chunks);
|
|
|
|
|
- item.put("successChunks", summary.successChunks);
|
|
|
|
|
- item.put("noDataChunks", summary.noDataChunks);
|
|
|
|
|
- item.put("failedChunks", summary.failedChunks);
|
|
|
|
|
- item.put("total", summary.total);
|
|
|
|
|
- item.put("saved", summary.saved);
|
|
|
|
|
- item.put("invalid", summary.invalid);
|
|
|
|
|
- item.put("dataDate", summary.dataDate == null ? null : summary.dataDate.toString());
|
|
|
|
|
- item.put("notes", new ArrayList<>(summary.notes));
|
|
|
|
|
- item.put("error", summary.error);
|
|
|
|
|
|
|
+ item.put("siteKey", summary.getSiteKey());
|
|
|
|
|
+ item.put("siteName", summary.getSiteName());
|
|
|
|
|
+ item.put("chunks", summary.getChunks());
|
|
|
|
|
+ item.put("successChunks", summary.getSuccessChunks());
|
|
|
|
|
+ item.put("noDataChunks", summary.getNoDataChunks());
|
|
|
|
|
+ item.put("failedChunks", summary.getFailedChunks());
|
|
|
|
|
+ item.put("total", summary.getTotal());
|
|
|
|
|
+ item.put("saved", summary.getSaved());
|
|
|
|
|
+ item.put("invalid", summary.getInvalid());
|
|
|
|
|
+ item.put("dataDate", summary.getDataDate() == null ? null : summary.getDataDate().toString());
|
|
|
|
|
+ item.put("notes", new ArrayList<>(summary.getNotes()));
|
|
|
|
|
+ item.put("error", summary.getError());
|
|
|
results.add(item);
|
|
results.add(item);
|
|
|
}
|
|
}
|
|
|
return results;
|
|
return results;
|
|
@@ -774,11 +757,11 @@ public class MaterialPriceSyncService {
|
|
|
private String failureMessage(List<SiteSummary> summaries) {
|
|
private String failureMessage(List<SiteSummary> summaries) {
|
|
|
StringBuilder builder = new StringBuilder("全部站点/分片失败");
|
|
StringBuilder builder = new StringBuilder("全部站点/分片失败");
|
|
|
for (SiteSummary summary : summaries) {
|
|
for (SiteSummary summary : summaries) {
|
|
|
- if (summary.notes.isEmpty()) {
|
|
|
|
|
|
|
+ if (summary.getNotes().isEmpty()) {
|
|
|
continue;
|
|
continue;
|
|
|
}
|
|
}
|
|
|
- builder.append(";").append(summary.siteKey).append(": ")
|
|
|
|
|
- .append(StringUtils.abbreviate(String.join(" | ", summary.notes), 200));
|
|
|
|
|
|
|
+ builder.append(";").append(summary.getSiteKey()).append(": ")
|
|
|
|
|
+ .append(StringUtils.abbreviate(String.join(" | ", summary.getNotes()), 200));
|
|
|
}
|
|
}
|
|
|
return StringUtils.abbreviate(builder.toString(), 500);
|
|
return StringUtils.abbreviate(builder.toString(), 500);
|
|
|
}
|
|
}
|
|
@@ -844,7 +827,7 @@ public class MaterialPriceSyncService {
|
|
|
*/
|
|
*/
|
|
|
private LinkedHashMap<String, List<RangeChunk>> buildRangePlan(LocalDate start, LocalDate end) {
|
|
private LinkedHashMap<String, List<RangeChunk>> buildRangePlan(LocalDate start, LocalDate end) {
|
|
|
LinkedHashMap<String, List<RangeChunk>> plan = new LinkedHashMap<>();
|
|
LinkedHashMap<String, List<RangeChunk>> plan = new LinkedHashMap<>();
|
|
|
- // 废废网指数多(约 6000 个),日期不切,整段按指数 id 取模切成多片,
|
|
|
|
|
|
|
+ // 废废网指数多(数千个),日期不切,整段按指数 id 取模切成多片,
|
|
|
// 每片只抓约 1/N 的指数,避免单次响应过大把 Java/Python 内存打爆;
|
|
// 每片只抓约 1/N 的指数,避免单次响应过大把 Java/Python 内存打爆;
|
|
|
// 其余站点只按天数切,天数常量见类顶部的 chunkDaysXxx
|
|
// 其余站点只按天数切,天数常量见类顶部的 chunkDaysXxx
|
|
|
addSiteChunks(plan, "ffhsw", start, end, chunkDaysFfhsw, Math.max(1, ffhswSlices));
|
|
addSiteChunks(plan, "ffhsw", start, end, chunkDaysFfhsw, Math.max(1, ffhswSlices));
|
|
@@ -930,6 +913,11 @@ public class MaterialPriceSyncService {
|
|
|
update.setUpdateById(operator);
|
|
update.setUpdateById(operator);
|
|
|
update.setUpdateTime(new Date());
|
|
update.setUpdateTime(new Date());
|
|
|
syncLogMapper.updateById(update);
|
|
syncLogMapper.updateById(update);
|
|
|
|
|
+ // 有新数据入库才需要刷新下拉选项/总行数缓存;失败/无数据时保持原缓存。
|
|
|
|
|
+ // 手动区间采集和 XXL 定时采集都走这里,保证采集一结束缓存就是最新的
|
|
|
|
|
+ if (saved != null && saved > 0) {
|
|
|
|
|
+ recordService.refreshQueryCache();
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/** 提交失败(线程池拒绝等)时的重载,日志对象已在内存中,可直接回写。 */
|
|
/** 提交失败(线程池拒绝等)时的重载,日志对象已在内存中,可直接回写。 */
|
|
@@ -1053,6 +1041,10 @@ public class MaterialPriceSyncService {
|
|
|
logEntity.setUpdateById(operator);
|
|
logEntity.setUpdateById(operator);
|
|
|
logEntity.setUpdateTime(new Date());
|
|
logEntity.setUpdateTime(new Date());
|
|
|
syncLogMapper.updateById(logEntity);
|
|
syncLogMapper.updateById(logEntity);
|
|
|
|
|
+ // 手动「立即采集」与 XXL 每日任务共用这个方法,采集结束后统一重建下拉选项缓存
|
|
|
|
|
+ if (saved != null && saved > 0) {
|
|
|
|
|
+ recordService.refreshQueryCache();
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
@@ -1199,125 +1191,5 @@ public class MaterialPriceSyncService {
|
|
|
}
|
|
}
|
|
|
return result;
|
|
return result;
|
|
|
}
|
|
}
|
|
|
-
|
|
|
|
|
- /** 区间任务的一个分片(站点 + 起止日期 + 片内下标)。 */
|
|
|
|
|
- private static class RangeChunk {
|
|
|
|
|
- private final String siteKey;
|
|
|
|
|
- private final LocalDate start;
|
|
|
|
|
- private final LocalDate end;
|
|
|
|
|
- // 分片下标/总数:废废网按指数取模切片,其他站点固定 0/1
|
|
|
|
|
- private final int sliceIndex;
|
|
|
|
|
- private final int sliceCount;
|
|
|
|
|
-
|
|
|
|
|
- RangeChunk(String siteKey, LocalDate start, LocalDate end,
|
|
|
|
|
- int sliceIndex, int sliceCount) {
|
|
|
|
|
- this.siteKey = siteKey;
|
|
|
|
|
- this.start = start;
|
|
|
|
|
- this.end = end;
|
|
|
|
|
- this.sliceIndex = sliceIndex;
|
|
|
|
|
- this.sliceCount = sliceCount;
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- LocalDate getStart() {
|
|
|
|
|
- return start;
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- LocalDate getEnd() {
|
|
|
|
|
- return end;
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- int getSliceIndex() {
|
|
|
|
|
- return sliceIndex;
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- int getSliceCount() {
|
|
|
|
|
- return sliceCount;
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- String label() {
|
|
|
|
|
- String base = siteKey + " " + start + "~" + end;
|
|
|
|
|
- if (sliceCount <= 1) {
|
|
|
|
|
- return base;
|
|
|
|
|
- }
|
|
|
|
|
- return base + " 第" + (sliceIndex + 1) + "/" + sliceCount + "片";
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- /**
|
|
|
|
|
- * 区间采集进度:多个站点线程同时完成分片,用 synchronized 保证
|
|
|
|
|
- * 「计数 + 写库」不会互相覆盖。
|
|
|
|
|
- */
|
|
|
|
|
- private class RangeProgress {
|
|
|
|
|
- private final String logId;
|
|
|
|
|
- private final String operator;
|
|
|
|
|
- private final int total;
|
|
|
|
|
- private final AtomicInteger done = new AtomicInteger(0);
|
|
|
|
|
- // 运行中累计的入库/无效条数,前端据此展示「已入库 X 条,跳过 Y 条」
|
|
|
|
|
- private final AtomicInteger saved = new AtomicInteger(0);
|
|
|
|
|
- private final AtomicInteger invalid = new AtomicInteger(0);
|
|
|
|
|
- private volatile String stage = "";
|
|
|
|
|
-
|
|
|
|
|
- RangeProgress(String logId, String operator, int total) {
|
|
|
|
|
- this.logId = logId;
|
|
|
|
|
- this.operator = operator;
|
|
|
|
|
- this.total = total;
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- /** 分片开始时记录当前阶段。 */
|
|
|
|
|
- void stage(String value) {
|
|
|
|
|
- this.stage = value;
|
|
|
|
|
- flush();
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- /** 分片结束后累加入库/无效条数(传增量,避免多站点并发时互相覆盖)。 */
|
|
|
|
|
- synchronized void addCounts(int savedDelta, int invalidDelta) {
|
|
|
|
|
- saved.addAndGet(savedDelta);
|
|
|
|
|
- invalid.addAndGet(invalidDelta);
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- /** 分片结束(无论成功失败)时进度 +1。 */
|
|
|
|
|
- synchronized void chunkDone() {
|
|
|
|
|
- done.incrementAndGet();
|
|
|
|
|
- flush();
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- private void flush() {
|
|
|
|
|
- try {
|
|
|
|
|
- MaterialPriceSyncLog update = new MaterialPriceSyncLog();
|
|
|
|
|
- update.setId(logId);
|
|
|
|
|
- update.setProgressDone(done.get());
|
|
|
|
|
- update.setProgressTotal(total);
|
|
|
|
|
- update.setSavedCount(saved.get());
|
|
|
|
|
- update.setFailedCount(invalid.get());
|
|
|
|
|
- update.setCurrentStage(stage);
|
|
|
|
|
- update.setUpdateById(operator);
|
|
|
|
|
- update.setUpdateTime(new Date());
|
|
|
|
|
- syncLogMapper.updateById(update);
|
|
|
|
|
- } catch (Exception e) {
|
|
|
|
|
- // 进度写库失败不影响抓取,前端下一次轮询会拿到旧值
|
|
|
|
|
- log.warn("更新区间采集进度失败 logId={} done={}/{}: {}", logId, done.get(), total, e.getMessage());
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- /** 单个站点的区间抓取结果,用于汇总状态和写 site_result。 */
|
|
|
|
|
- private static class SiteSummary {
|
|
|
|
|
- private final String siteKey;
|
|
|
|
|
- private final String siteName;
|
|
|
|
|
- private int chunks;
|
|
|
|
|
- private int successChunks;
|
|
|
|
|
- private int noDataChunks;
|
|
|
|
|
- private int failedChunks;
|
|
|
|
|
- private int total;
|
|
|
|
|
- private int saved;
|
|
|
|
|
- private int invalid;
|
|
|
|
|
- private LocalDate dataDate;
|
|
|
|
|
- private final Set<String> notes = new LinkedHashSet<>();
|
|
|
|
|
- private String error;
|
|
|
|
|
-
|
|
|
|
|
- SiteSummary(String siteKey, String siteName) {
|
|
|
|
|
- this.siteKey = siteKey;
|
|
|
|
|
- this.siteName = siteName;
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
}
|
|
}
|
|
|
|
|
+
|