fix: 全票池同步切片续扫+有变化才写——单查询窗口2.5万硬上限触顶自动二分时间片递归（根治窗口尾部永久漏同步），镜像upsert轻量比对（流水号/价税合计/税额/可用金额/使用状态）无变化跳过重写消整窗无效写
This commit is contained in:
刘泉佳
+131
-51
@@ -50,7 +50,7 @@ import org.springblade.thirdparty.bfi.pojo.vo.BfiInvoicePoolVO;
|
||||
import org.springblade.thirdparty.bfi.pojo.vo.BfiPageVO;
|
||||
import org.springblade.thirdparty.bfi.pojo.vo.BfiReceivedInvoiceVO;
|
||||
import org.springblade.thirdparty.bfi.service.IBfiService;
|
||||
import org.springblade.transport.constantq.KingdeeAttachmentConstant;
|
||||
import org.springblade.transport.constant.KingdeeAttachmentConstant;
|
||||
import org.springblade.transport.constant.KingdeeBfiConstant;
|
||||
import org.springblade.transport.constant.TransportBusiConstant;
|
||||
import org.springblade.transport.mapper.CustomerArchiveMapper;
|
||||
@@ -70,7 +70,6 @@ import org.springblade.transport.pojo.entity.CustomerArchive;
|
||||
import org.springblade.transport.pojo.entity.CustomerContact;
|
||||
import org.springblade.transport.pojo.entity.FormalSettlement;
|
||||
import org.springblade.transport.pojo.entity.FormalSettlementDetail;
|
||||
import org.springblade.transport.pojo.entity.FormalSettlementInvoice;
|
||||
import org.springblade.transport.pojo.entity.InvoiceReceipt;
|
||||
import org.springblade.transport.pojo.entity.InvoiceReceiptRecord;
|
||||
import org.springblade.transport.pojo.entity.InvoiceReceiptSettlement;
|
||||
@@ -1181,11 +1180,12 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
||||
}
|
||||
|
||||
/**
|
||||
* 同步主体（39ffac1 绑定回填口径瘦身版）：滚动增量窗口分页拉取 A801-1（按票池更新时间）
|
||||
* + 镜像批量 upsert（全列覆盖）+ 未绑定存量登记顺带补绑（同页发票号码批量定位登记，页级事务批量落库）。
|
||||
* 同步窗口内所有可用态票全量重写，无指纹比对（ponytail: 万行级窗口全列重写成本可接受，
|
||||
* 人工字段清单式指纹有"新增列漏维护即静默不同步"坑；量级上来后升级为整行载荷规范化哈希）。
|
||||
* 拉取只认正常票（invoiceStatus=0），不推单据使用状态——在用/已用票随 updateTime 进窗口刷新
|
||||
* 同步主体（39ffac1 绑定回填口径瘦身版）：滚动增量窗口按 updateTime **切片续扫** + 镜像批量 upsert（有变化才写）
|
||||
* + 未绑定存量登记顺带补绑（同页发票号码批量定位登记，页级事务批量落库）。
|
||||
* <p>切片续扫（ponytail: 单查询窗口有 MAX_POOL_PAGES×500=2.5 万硬上限，窗口数据超限尾部永久漏同步；
|
||||
* 切片让每片各自独立受翻页保护，总覆盖不受限）：把大窗口切成小片独立翻页，某片触顶说明该时间片内
|
||||
* 数据 >2.5 万，二分为 [start, mid) [mid, end) 两片递归处理；updateTime 单调，切片边界不重不漏。
|
||||
* <p>拉取只认正常票（invoiceStatus=0），不推单据使用状态——在用/已用票随 updateTime 进窗口刷新
|
||||
* 金额使用状态，选票可用过滤在本地 SQL 执行；被删除/红冲票不再出现在拉取结果，由保存/认领重查闸门兜底。
|
||||
*
|
||||
* @return 执行摘要
|
||||
@@ -1194,55 +1194,99 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
||||
long startMillis = System.currentTimeMillis();
|
||||
LocalDateTime windowStart = LocalDateTime.now().minusDays(poolIncrementDays);
|
||||
LocalDateTime windowEnd = LocalDateTime.now();
|
||||
// 以下为本轮统计口径，全部累计整轮并进入结束摘要
|
||||
// A801-1 原始拉取张数
|
||||
int pulledCount = 0;
|
||||
// 镜像新增条数（票池有、本地无）
|
||||
int created = 0;
|
||||
// 镜像重写条数
|
||||
int updated = 0;
|
||||
// 登记首次绑定条数（罕见：存量手工数据未绑定时顺带补绑）
|
||||
int bound = 0;
|
||||
// 登记无变化跳过条数
|
||||
int unchanged = 0;
|
||||
// 页级批量落库失败条数（整页跳过，下轮幂等重试）
|
||||
int failed = 0;
|
||||
// 实际处理的翻页数，触达最大页数保护时配合 warn 定位截断
|
||||
int pageCount = 0;
|
||||
log.info("金蝶全票池同步开始，增量窗口 {} ~ {}，单页 {} 条最多 {} 页",
|
||||
SyncStats stats = new SyncStats();
|
||||
log.info("金蝶全票池同步开始，增量窗口 {} ~ {}，单页 {} 条最多 {} 页（触顶自动二分切片）",
|
||||
windowStart.format(POOL_TIME_FORMATTER), windowEnd.format(POOL_TIME_FORMATTER),
|
||||
POOL_PAGE_SIZE, MAX_POOL_PAGES);
|
||||
this.syncWindowSlice(windowStart, windowEnd, stats);
|
||||
String summary = "金蝶全票池同步：翻页" + stats.pageCount + "页切片" + stats.sliceCount + "片拉取"
|
||||
+ stats.pulledCount + "张发票，镜像新增" + stats.created + "更新" + stats.updated + "无变化"
|
||||
+ stats.mirrorUnchanged + "，补绑" + stats.bound + "，登记无变化跳过" + stats.unchanged
|
||||
+ "，失败" + stats.failed + "，耗时" + (System.currentTimeMillis() - startMillis) + "ms";
|
||||
log.info("金蝶全票池同步结束，{}", summary);
|
||||
return summary;
|
||||
}
|
||||
|
||||
/** 整轮同步累计统计（各切片共享累加，结束进摘要） */
|
||||
private static final class SyncStats {
|
||||
/** A801-1 原始拉取张数 */
|
||||
int pulledCount;
|
||||
/** 镜像新增条数（票池有、本地无） */
|
||||
int created;
|
||||
/** 镜像有变化并重写条数 */
|
||||
int updated;
|
||||
/** 镜像无变化跳过条数 */
|
||||
int mirrorUnchanged;
|
||||
/** 登记补绑条数（罕见：存量手工数据未绑定时顺带补绑） */
|
||||
int bound;
|
||||
/** 登记无变化跳过条数 */
|
||||
int unchanged;
|
||||
/** 页级批量落库失败条数（整页跳过，下轮幂等重试） */
|
||||
int failed;
|
||||
/** 实际处理的翻页总页数 */
|
||||
int pageCount;
|
||||
/** 实际处理的切片数 */
|
||||
int sliceCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* 同步单个时间片 [sliceStart, sliceEnd)：独立翻页拉取（最多 MAX_POOL_PAGES 页），
|
||||
* 触顶说明该时间片数据 >2.5 万，从中点二分为两片递归续扫（递归深度受时间粒度自然收敛，
|
||||
* 单片 500 条以内 1 页即短页终止）
|
||||
*
|
||||
* @param sliceStart 切片起点（含）
|
||||
* @param sliceEnd 切片终点（不含，避免与下一片边界重复拉取）
|
||||
* @param stats 整轮统计累加器
|
||||
*/
|
||||
private void syncWindowSlice(LocalDateTime sliceStart, LocalDateTime sliceEnd, SyncStats stats) {
|
||||
// 二分收敛保护：切片粒度细到 1 秒仍触顶说明远端异常（同秒 >2.5 万条不可能），不再二分直接记漏
|
||||
if (!sliceStart.isBefore(sliceEnd)) {
|
||||
log.error("金蝶全票池同步切片区间为空（{} ~ {}），远端 updateTime 粒度异常，放弃该切片", sliceStart, sliceEnd);
|
||||
return;
|
||||
}
|
||||
boolean truncated = false;
|
||||
for (int pageNo = 1; pageNo <= MAX_POOL_PAGES; pageNo++) {
|
||||
List<BfiInvoicePoolVO> rows = this.pullPoolWindowPage(windowStart, windowEnd, pageNo);
|
||||
List<BfiInvoicePoolVO> rows = this.pullPoolWindowPage(sliceStart, sliceEnd, pageNo);
|
||||
if (rows.isEmpty()) {
|
||||
break;
|
||||
}
|
||||
pageCount++;
|
||||
pulledCount += rows.size();
|
||||
stats.pageCount++;
|
||||
stats.pulledCount += rows.size();
|
||||
BatchUpsertResult upsertResult = this.batchUpsertInvoicePool(rows);
|
||||
created += upsertResult.created();
|
||||
updated += upsertResult.updated();
|
||||
// 已登记票跟随刷新（39ffac1 口径）：同页发票号码批量定位登记，内存比对变更后页级事务批量落库
|
||||
stats.created += upsertResult.created();
|
||||
stats.updated += upsertResult.updated();
|
||||
stats.mirrorUnchanged += upsertResult.unchanged();
|
||||
// 已登记票跟随补绑：同页发票号码批量定位登记，绑定关系变更页级事务批量落库
|
||||
if (!upsertResult.mirrors().isEmpty()) {
|
||||
RefreshResult refreshResult = this.refreshBoundReceipts(pageNo, upsertResult.mirrors());
|
||||
bound += refreshResult.bound();
|
||||
unchanged += refreshResult.unchanged();
|
||||
failed += refreshResult.failed();
|
||||
RefreshResult refreshResult = this.refreshBoundReceipts(stats.pageCount, upsertResult.mirrors());
|
||||
stats.bound += refreshResult.bound();
|
||||
stats.unchanged += refreshResult.unchanged();
|
||||
stats.failed += refreshResult.failed();
|
||||
}
|
||||
// 远程短页：该片到底，正常终止
|
||||
if (rows.size() < POOL_PAGE_SIZE) {
|
||||
break;
|
||||
stats.sliceCount++;
|
||||
return;
|
||||
}
|
||||
truncated = true;
|
||||
}
|
||||
if (pageCount >= MAX_POOL_PAGES) {
|
||||
// 末轮仍满页说明窗口数据量已超过翻页保护，尾部数据本轮未处理（须缩小 incrementDays 或调大保护页数）
|
||||
log.warn("金蝶全票池同步翻页达最大保护 {} 页（{} 条）仍未到底，窗口尾部本轮未处理",
|
||||
MAX_POOL_PAGES, MAX_POOL_PAGES * POOL_PAGE_SIZE);
|
||||
// 触顶：该切片数据超过单查询窗口上限，从中点二分续扫（updateTime 落在中点的行归左片 [start, mid)，
|
||||
// 右片 [mid, end) 不含 mid，同 updateTime 行整体归左片，不重不漏）
|
||||
if (truncated) {
|
||||
LocalDateTime mid = sliceStart.plus(java.time.Duration.between(sliceStart, sliceEnd).dividedBy(2));
|
||||
if (!mid.isAfter(sliceStart)) {
|
||||
// 切片已细到不足 1 秒仍满页：理论不可能，防御死递归
|
||||
log.error("金蝶全票池同步切片 {} ~ {} 已达最小粒度仍满页（>{} 条），该时间片数据异常，跳过",
|
||||
sliceStart, sliceEnd, MAX_POOL_PAGES * POOL_PAGE_SIZE);
|
||||
return;
|
||||
}
|
||||
log.warn("金蝶全票池同步切片 {} ~ {} 触达翻页保护 {} 页（{} 条），二分续扫 [{} ~ {}) + [{} ~ {})",
|
||||
sliceStart, sliceEnd, MAX_POOL_PAGES, MAX_POOL_PAGES * POOL_PAGE_SIZE, sliceStart, mid, mid, sliceEnd);
|
||||
this.syncWindowSlice(sliceStart, mid, stats);
|
||||
this.syncWindowSlice(mid, sliceEnd, stats);
|
||||
return;
|
||||
}
|
||||
String summary = "金蝶全票池同步：翻页" + pageCount + "页拉取" + pulledCount + "张发票，镜像新增" + created
|
||||
+ "更新" + updated + "，补绑" + bound + "，登记无变化跳过" + unchanged
|
||||
+ "，失败" + failed + "，耗时" + (System.currentTimeMillis() - startMillis) + "ms";
|
||||
log.info("金蝶全票池同步结束，{}", summary);
|
||||
return summary;
|
||||
stats.sliceCount++;
|
||||
}
|
||||
|
||||
/** 已登记票跟随刷新结果统计（绑定/刷新/无变化/失败） */
|
||||
@@ -1345,7 +1389,7 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
||||
private BatchUpsertResult batchUpsertInvoicePool(List<BfiInvoicePoolVO> rows) {
|
||||
Map<String, KingdeeInvoicePool> mirrorMap = new LinkedHashMap<>();
|
||||
if (rows.isEmpty()) {
|
||||
return new BatchUpsertResult(mirrorMap, 0, 0);
|
||||
return new BatchUpsertResult(mirrorMap, 0, 0, 0);
|
||||
}
|
||||
List<String> invoiceNos = rows.stream()
|
||||
.map(row -> row.getInvoiceNo().trim())
|
||||
@@ -1358,16 +1402,21 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
||||
.collect(Collectors.toMap(KingdeeInvoicePool::getInvoiceNo, Function.identity(), (first, second) -> first));
|
||||
List<KingdeeInvoicePool> toInsert = new ArrayList<>();
|
||||
List<KingdeeInvoicePool> toUpdate = new ArrayList<>();
|
||||
int unchangedCount = 0;
|
||||
for (BfiInvoicePoolVO row : rows) {
|
||||
if (Func.isEmpty(row.getInvoiceNo())) {
|
||||
continue;
|
||||
}
|
||||
KingdeeInvoicePool mirror = fillInvoicePool(existingMap.get(row.getInvoiceNo().trim()), row);
|
||||
KingdeeInvoicePool existing = existingMap.get(row.getInvoiceNo().trim());
|
||||
KingdeeInvoicePool mirror = fillInvoicePool(existing, row);
|
||||
mirrorMap.put(mirror.getInvoiceNo(), mirror);
|
||||
if (mirror.getId() == null) {
|
||||
if (existing == null) {
|
||||
toInsert.add(mirror);
|
||||
} else {
|
||||
} else if (poolMirrorChanged(existing, mirror)) {
|
||||
toUpdate.add(mirror);
|
||||
} else {
|
||||
// 票面无变化跳过重写：整窗重复拉取时消掉每轮 2.5 万行无效写
|
||||
unchangedCount++;
|
||||
}
|
||||
}
|
||||
// 批量写库：失败中断本轮，由任务层记录异常、下轮重试（upsert 幂等）
|
||||
@@ -1377,12 +1426,43 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
||||
if (!toUpdate.isEmpty()) {
|
||||
Db.updateBatchById(toUpdate);
|
||||
}
|
||||
log.info("金蝶全票池镜像批量写入完成，新增 {} 条，更新 {} 条", toInsert.size(), toUpdate.size());
|
||||
return new BatchUpsertResult(mirrorMap, toInsert.size(), toUpdate.size());
|
||||
log.info("金蝶全票池镜像批量写入完成，新增 {} 条，更新 {} 条，无变化 {} 条", toInsert.size(), toUpdate.size(), unchangedCount);
|
||||
return new BatchUpsertResult(mirrorMap, toInsert.size(), toUpdate.size(), unchangedCount);
|
||||
}
|
||||
|
||||
/** 票池镜像批量 upsert 结果（invoiceNo -> 镜像，新增条数、更新条数） */
|
||||
private record BatchUpsertResult(Map<String, KingdeeInvoicePool> mirrors, int created, int updated) {
|
||||
/**
|
||||
* 镜像票面是否有变化（轻量比对，ponytail: 只比对"会变且影响业务"的字段，
|
||||
* 不搞人工字段清单——清单漏新增列即静默不同步是旧版指纹的坑）：
|
||||
* 流水号（金蝶侧标识）、价税合计、税额、剩余可用金额、金额使用状态。
|
||||
* BigDecimal 比对用 compareTo（13.00 与 13.0 等价）；空值视为变化跳过重写（与 MP 非空字段更新策略一致）
|
||||
*
|
||||
* @param existing 库中既有镜像
|
||||
* @param filled 填充后的镜像
|
||||
* @return true 表示有变化需重写
|
||||
*/
|
||||
private boolean poolMirrorChanged(KingdeeInvoicePool existing, KingdeeInvoicePool filled) {
|
||||
if (!Objects.equals(existing.getSerialNo(), filled.getSerialNo())
|
||||
|| !Objects.equals(existing.getAmountUseStatus(), filled.getAmountUseStatus())) {
|
||||
return true;
|
||||
}
|
||||
return amountChanged(existing.getInvoiceAmount(), filled.getInvoiceAmount())
|
||||
|| amountChanged(existing.getTaxAmount(), filled.getTaxAmount())
|
||||
|| amountChanged(existing.getAvailableAmount(), filled.getAvailableAmount());
|
||||
}
|
||||
|
||||
/** 金额变化比对：双方均空视为无变化；一方空一方有视为有变化；否则按数值比较 */
|
||||
private boolean amountChanged(BigDecimal before, BigDecimal after) {
|
||||
if (before == null && after == null) {
|
||||
return false;
|
||||
}
|
||||
if (before == null || after == null) {
|
||||
return true;
|
||||
}
|
||||
return before.compareTo(after) != 0;
|
||||
}
|
||||
|
||||
/** 票池镜像批量 upsert 结果（invoiceNo -> 镜像，新增条数、更新条数、无变化条数） */
|
||||
private record BatchUpsertResult(Map<String, KingdeeInvoicePool> mirrors, int created, int updated, int unchanged) {
|
||||
}
|
||||
|
||||
// ==================== 结算单认领确认重查闸门 ====================
|
||||
|
||||
Reference in New Issue
Block a user