refactor: 金蝶全票池同步改用BfiUtil二分切片续扫
This commit is contained in:
刘泉佳
+44
-64
@@ -50,6 +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.thirdparty.bfi.util.BfiUtil;
|
||||
import org.springblade.transport.constant.HcsSyncConstant;
|
||||
import org.springblade.transport.constant.KingdeeAttachmentConstant;
|
||||
import org.springblade.transport.config.BfiProperties;
|
||||
@@ -1304,11 +1305,11 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
||||
invoiceTypes.isEmpty() ? "不限制" : invoiceTypes, customerTaxNos != null ? "开启" : "关闭",
|
||||
POOL_PAGE_SIZE, MAX_POOL_PAGES);
|
||||
if (invoiceTypes.isEmpty()) {
|
||||
this.syncWindowSlice(windowStart, windowEnd, null, customerTaxNos, stats);
|
||||
this.syncPoolWindow(windowStart, windowEnd, null, customerTaxNos, stats);
|
||||
} else {
|
||||
// 单值参数拆轮：每个票种编码独立完整切片续扫，互不挤占翻页额度
|
||||
for (String invoiceType : invoiceTypes) {
|
||||
this.syncWindowSlice(windowStart, windowEnd, invoiceType, customerTaxNos, stats);
|
||||
this.syncPoolWindow(windowStart, windowEnd, invoiceType, customerTaxNos, stats);
|
||||
}
|
||||
}
|
||||
String summary = "金蝶全票池同步：翻页" + stats.pageCount + "页切片" + stats.sliceCount + "片拉取"
|
||||
@@ -1358,72 +1359,51 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
||||
}
|
||||
|
||||
/**
|
||||
* 同步单个时间片 [sliceStart, sliceEnd)：独立翻页拉取（最多 MAX_POOL_PAGES 页），
|
||||
* 触顶说明该时间片数据 >2.5 万，从中点二分为两片递归续扫（递归深度受时间粒度自然收敛，
|
||||
* 单片 500 条以内 1 页即短页终止）
|
||||
* 同步单个票种的时间窗口：翻页与触顶二分续扫统一交给 {@link BfiUtil#forEachWindowSlice}，
|
||||
* 页级客商单位过滤、镜像批量 upsert、已登记票跟随补绑在页回调中完成
|
||||
*
|
||||
* @param sliceStart 切片起点（含）
|
||||
* @param sliceEnd 切片终点（不含，避免与下一片边界重复拉取）
|
||||
* @param customerTaxNos 客商发票信息税号集，null 表示客商单位过滤关闭（整轮共享）
|
||||
* @param stats 整轮统计累加器
|
||||
* @param windowStart 窗口起点（含）
|
||||
* @param windowEnd 窗口终点（含）
|
||||
* @param invoiceType 发票种类编码，null 不限
|
||||
* @param customerTaxNos 客商发票信息税号集，null 表示客商单位过滤关闭（整轮共享）
|
||||
* @param stats 整轮统计累加器
|
||||
*/
|
||||
private void syncWindowSlice(LocalDateTime sliceStart, LocalDateTime sliceEnd, String invoiceType,
|
||||
private void syncPoolWindow(LocalDateTime windowStart, LocalDateTime windowEnd, String invoiceType,
|
||||
Set<String> customerTaxNos, 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(sliceStart, sliceEnd, invoiceType, pageNo);
|
||||
if (rows.isEmpty()) {
|
||||
break;
|
||||
}
|
||||
stats.pageCount++;
|
||||
stats.pulledCount += rows.size();
|
||||
// 客商单位过滤：销/购方税号未同时登记在客商发票信息中的发票不入镜像；
|
||||
// 短页判定用原始条数，过滤后的空页不能提前终止翻页
|
||||
stats.sliceCount += BfiUtil.forEachWindowSlice("金蝶全票池同步", windowStart, windowEnd,
|
||||
POOL_PAGE_SIZE, MAX_POOL_PAGES,
|
||||
(sliceStart, sliceEnd, pageNo) -> this.pullPoolWindowPage(sliceStart, sliceEnd, invoiceType, pageNo),
|
||||
rows -> this.handlePoolPage(rows, customerTaxNos, stats));
|
||||
}
|
||||
|
||||
/**
|
||||
* 票池同步单页处理：客商单位过滤 + 镜像批量 upsert + 已登记票跟随补绑。
|
||||
* 客商过滤只影响入镜像行数，翻页节奏由 BfiUtil 按原始行数判断，不受过滤影响
|
||||
*
|
||||
* @param rows 本页拉取的票池发票（原始行）
|
||||
* @param customerTaxNos 客商发票信息税号集，null 表示客商单位过滤关闭
|
||||
* @param stats 整轮统计累加器
|
||||
*/
|
||||
private void handlePoolPage(List<BfiInvoicePoolVO> rows, Set<String> customerTaxNos, SyncStats stats) {
|
||||
stats.pageCount++;
|
||||
stats.pulledCount += rows.size();
|
||||
// 客商单位过滤：销/购方税号未同时登记在客商发票信息中的发票不入镜像
|
||||
if (customerTaxNos != null) {
|
||||
int rawPageSize = rows.size();
|
||||
if (customerTaxNos != null) {
|
||||
rows = rows.stream().filter(row -> this.bothUnitsRegistered(row, customerTaxNos)).toList();
|
||||
stats.customerFiltered += rawPageSize - rows.size();
|
||||
}
|
||||
BatchUpsertResult upsertResult = this.batchUpsertInvoicePool(rows);
|
||||
stats.created += upsertResult.created();
|
||||
stats.updated += upsertResult.updated();
|
||||
stats.mirrorUnchanged += upsertResult.unchanged();
|
||||
// 已登记票跟随补绑：同页发票号码批量定位登记，绑定关系变更页级事务批量落库
|
||||
if (!upsertResult.mirrors().isEmpty()) {
|
||||
RefreshResult refreshResult = this.refreshBoundReceipts(stats.pageCount, upsertResult.mirrors());
|
||||
stats.bound += refreshResult.bound();
|
||||
stats.unchanged += refreshResult.unchanged();
|
||||
stats.failed += refreshResult.failed();
|
||||
}
|
||||
// 远程短页：该片到底，正常终止（按原始拉取条数判断，客商过滤不改变翻页节奏）
|
||||
if (rawPageSize < POOL_PAGE_SIZE) {
|
||||
stats.sliceCount++;
|
||||
return;
|
||||
}
|
||||
truncated = true;
|
||||
rows = rows.stream().filter(row -> this.bothUnitsRegistered(row, customerTaxNos)).toList();
|
||||
stats.customerFiltered += rawPageSize - rows.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, invoiceType, customerTaxNos, stats);
|
||||
this.syncWindowSlice(mid, sliceEnd, invoiceType, customerTaxNos, stats);
|
||||
return;
|
||||
BatchUpsertResult upsertResult = this.batchUpsertInvoicePool(rows);
|
||||
stats.created += upsertResult.created();
|
||||
stats.updated += upsertResult.updated();
|
||||
stats.mirrorUnchanged += upsertResult.unchanged();
|
||||
// 已登记票跟随补绑：同页发票号码批量定位登记，绑定关系变更页级事务批量落库
|
||||
if (!upsertResult.mirrors().isEmpty()) {
|
||||
RefreshResult refreshResult = this.refreshBoundReceipts(stats.pageCount, upsertResult.mirrors());
|
||||
stats.bound += refreshResult.bound();
|
||||
stats.unchanged += refreshResult.unchanged();
|
||||
stats.failed += refreshResult.failed();
|
||||
}
|
||||
stats.sliceCount++;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -1523,7 +1503,7 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
||||
}
|
||||
|
||||
/** A801-1 按票池更新时间窗口分页拉取（只认正常票，不推单据使用状态；invoiceType 单值下推，null 不限） */
|
||||
private List<BfiInvoicePoolVO> pullPoolWindowPage(LocalDateTime windowStart, LocalDateTime windowEnd,
|
||||
private BfiPageVO<BfiInvoicePoolVO> pullPoolWindowPage(LocalDateTime windowStart, LocalDateTime windowEnd,
|
||||
String invoiceType, int pageNo) {
|
||||
BfiInvoicePoolQueryDataDTO data = new BfiInvoicePoolQueryDataDTO();
|
||||
data.setPageNo(pageNo);
|
||||
@@ -1536,7 +1516,7 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
||||
param.setQueryParam(data);
|
||||
BfiPageVO<BfiInvoicePoolVO> poolPage = bfiService.queryInvoicePool(param);
|
||||
log.info("金蝶全票池同步 第{}页拉取 {} 条", pageNo, poolPage.getRows() == null ? 0 : poolPage.getRows().size());
|
||||
return poolPage.getRows() == null ? List.of() : poolPage.getRows();
|
||||
return poolPage;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user