fix: 收款流水同步改用二分切片续扫修复窗口超限漏同步尾部数据
This commit is contained in:
刘泉佳
+130
-72
@@ -445,8 +445,10 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl<KingdeeReceiptFlowMa
|
||||
}
|
||||
|
||||
/**
|
||||
* 手动同步金蝶收款流水：直连 A301-1 收款单查询，按流水分录展开、明细流水号幂等 upsert；
|
||||
* 数据量大，流式翻页处理，累计满一批（5000 条流水行）即独立事务入库一批，不全量驻留内存
|
||||
* 手动同步金蝶收款流水：直连 A301-1 收款单查询，收款单主单粒度、以金蝶收款单号为幂等键 upsert；
|
||||
* 数据量大，时间窗口二分切片续扫（BfiUtil.forEachWindowSlice）：单片触顶自动二分续扫，
|
||||
* 总覆盖不受单查询窗口上限约束，尾部不漏；累计满一批（5000 条流水行）即独立事务入库一批，
|
||||
* 不全量驻留内存
|
||||
*
|
||||
* @return 本次同步条数（新增+更新）
|
||||
*/
|
||||
@@ -456,87 +458,105 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl<KingdeeReceiptFlowMa
|
||||
LocalDate endDate = LocalDate.now();
|
||||
LocalDate startDate = endDate.minusDays(days);
|
||||
// 金蝶 A301-1 仅接受带时分秒的日期参数，传纯日期会报 bos.filterSqlBuilder 时区时间格式错误
|
||||
String beginBizdate = startDate.atStartOfDay().format(BFI_DATE_TIME_FORMATTER);
|
||||
String endBizdate = endDate.atTime(23, 59, 59).format(BFI_DATE_TIME_FORMATTER);
|
||||
log.info("收款流水同步开始，业务日期窗口：{} ~ {}（近{}天）", beginBizdate, endBizdate, days);
|
||||
LocalDateTime windowStart = startDate.atStartOfDay();
|
||||
LocalDateTime windowEnd = endDate.atTime(23, 59, 59);
|
||||
log.info("收款流水同步开始，业务日期窗口：{} ~ {}（近{}天）",
|
||||
windowStart.format(BFI_DATE_TIME_FORMATTER), windowEnd.format(BFI_DATE_TIME_FORMATTER), days);
|
||||
long start = System.currentTimeMillis();
|
||||
// 同步口径：按业务日期窗口拉取金蝶收款单（收款处理），按流水分录展开，
|
||||
// 以金蝶明细流水号为幂等键 upsert 收款流水表；本地已认领金额/状态（claimedAmount/claimStatus）
|
||||
// 由认领、作废驱动，不被同步顺带覆盖，未匹配收款金额单独存 unmatch_amount_rec 作为认领上限。
|
||||
BfiRecBillQueryDTO queryDTO = new BfiRecBillQueryDTO();
|
||||
queryDTO.setBeginBizdate(beginBizdate);
|
||||
queryDTO.setEndBizdate(endBizdate);
|
||||
queryDTO.setPageSize(SYNC_PAGE_SIZE);
|
||||
// 同步口径：按业务日期窗口拉取金蝶收款单（收款处理），以金蝶明细流水号为幂等键 upsert 收款流水表；
|
||||
// 本地已认领金额/状态（claimedAmount/claimStatus）由认领、作废驱动，不被同步顺带覆盖，
|
||||
// 未匹配收款金额单独存 unclaimed_amount 作为认领上限。
|
||||
// 按组织过滤：组织管理（blade_dept）中平台公司（是否平台公司=是）且已维护金蝶编码的组织
|
||||
List<String> orgNumbers = baseMapper.selectPlatformOrgNumbers();
|
||||
if (Func.isEmpty(orgNumbers)) {
|
||||
throw new ServiceException("组织管理中未维护平台公司金蝶编码，无法同步金蝶收款流水，请先检查组织数据");
|
||||
}
|
||||
queryDTO.setOrgNumber(orgNumbers);
|
||||
log.info("收款流水同步按组织过滤，组织编码：{}", orgNumbers);
|
||||
// 待入库缓冲：累计满 SYNC_BATCH_SIZE 即入库一批并清空
|
||||
List<BfiRecBillRow> pending = new ArrayList<>();
|
||||
Set<String> batchSerials = new HashSet<>();
|
||||
int totalCount = 0;
|
||||
int batchNo = 0;
|
||||
int pullCount = 0;
|
||||
int skippedCount = 0;
|
||||
for (int pageNo = 1; pageNo <= SYNC_MAX_PAGES; pageNo++) {
|
||||
queryDTO.setPageNo(pageNo);
|
||||
BfiPageVO<BfiRecBillVO> page = bfiService.queryRecBill(queryDTO);
|
||||
List<BfiRecBillVO> rows = page == null ? List.of() : page.getRows();
|
||||
if (Func.isEmpty(rows)) {
|
||||
break;
|
||||
}
|
||||
pullCount += rows.size();
|
||||
for (BfiRecBillVO recBill : rows) {
|
||||
// 主单粒度：一张收款单一条流水，不展开流水分录；明细流水号取收款单号（幂等键，C104-2 按此定位合规流水）
|
||||
String detailSerialNo = trimToEmpty(recBill.getBillno());
|
||||
if (Func.isEmpty(detailSerialNo)) {
|
||||
log.warn("收款流水同步跳过：收款单缺少单号");
|
||||
skippedCount++;
|
||||
continue;
|
||||
}
|
||||
if (!batchSerials.add(detailSerialNo)) {
|
||||
throw new ServiceException("收款单号" + detailSerialNo + "重复");
|
||||
}
|
||||
// 收款金额取主单 actrecamt（金蝶收款金额口径）
|
||||
BigDecimal receiptAmount = money(recBill.getActrecamt());
|
||||
if (receiptAmount.compareTo(BigDecimal.ZERO) <= 0) {
|
||||
// log.warn("收款流水同步跳过：收款单[{}]收款金额（actrecamt）为空或小于0", recBill.getBillno());
|
||||
skippedCount++;
|
||||
continue;
|
||||
}
|
||||
pending.add(new BfiRecBillRow(recBill, detailSerialNo, receiptAmount));
|
||||
if (pending.size() >= SYNC_BATCH_SIZE) {
|
||||
batchNo++;
|
||||
totalCount += syncBatch(batchNo, pending);
|
||||
pending.clear();
|
||||
batchSerials.clear();
|
||||
}
|
||||
}
|
||||
// 翻页终止：末页标识、数据不满一页（防金蝶侧 lastPage 异常死循环，另有 SYNC_MAX_PAGES 兜底）
|
||||
if (Boolean.TRUE.equals(page.getLastPage())
|
||||
|| (page.getPageSize() != null && rows.size() < page.getPageSize())) {
|
||||
break;
|
||||
}
|
||||
RecBillSyncState state = new RecBillSyncState();
|
||||
state.sliceCount += BfiUtil.forEachWindowSlice("收款流水同步", windowStart, windowEnd,
|
||||
SYNC_PAGE_SIZE, SYNC_MAX_PAGES,
|
||||
(sliceStart, sliceEnd, pageNo) -> this.pullRecBillPage(sliceStart, sliceEnd, orgNumbers, pageNo),
|
||||
rows -> this.handleRecBillPage(rows, state));
|
||||
// 尾批不足 SYNC_BATCH_SIZE 的剩余入库
|
||||
if (!state.pending.isEmpty()) {
|
||||
state.batchNo++;
|
||||
state.totalCount += this.syncBatch(state.batchNo, state.pending);
|
||||
state.pending.clear();
|
||||
state.batchSerials.clear();
|
||||
}
|
||||
if (!pending.isEmpty()) {
|
||||
batchNo++;
|
||||
totalCount += syncBatch(batchNo, pending);
|
||||
pending.clear();
|
||||
batchSerials.clear();
|
||||
}
|
||||
if (pullCount == 0) {
|
||||
record(null, null, "sync", "手动同步流水", null, null, BigDecimal.ZERO,
|
||||
if (state.pullCount == 0) {
|
||||
this.record(null, null, "sync", "手动同步流水", null, null, BigDecimal.ZERO,
|
||||
"金蝶收款单未查询到数据");
|
||||
} else if (totalCount == 0) {
|
||||
record(null, null, "sync", "手动同步流水", null, null, BigDecimal.ZERO,
|
||||
} else if (state.totalCount == 0) {
|
||||
this.record(null, null, "sync", "手动同步流水", null, null, BigDecimal.ZERO,
|
||||
"金蝶收款单无有效数据");
|
||||
}
|
||||
log.info("金蝶收款流水同步完成，拉取收款单 {} 张，新增/更新 {} 条，跳过 {} 条，分 {} 批入库，耗时 {}ms",
|
||||
pullCount, totalCount, skippedCount, batchNo, System.currentTimeMillis() - start);
|
||||
return totalCount;
|
||||
log.info("金蝶收款流水同步完成，翻页{}页切片{}片，拉取收款单 {} 张，新增/更新 {} 条，跳过 {} 条，"
|
||||
+ "分 {} 批入库，耗时 {}ms",
|
||||
state.pageCount, state.sliceCount, state.pullCount, state.totalCount, state.skippedCount,
|
||||
state.batchNo, System.currentTimeMillis() - start);
|
||||
return state.totalCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* A301-1 按业务日期时间片分页拉取（组织编码过滤整轮共享，切片起止由 BfiUtil 二分传入）
|
||||
*
|
||||
* @param sliceStart 切片起点（含）
|
||||
* @param sliceEnd 切片终点（含）
|
||||
* @param orgNumbers 平台公司组织编码集
|
||||
* @param pageNo 页码，从 1 开始
|
||||
* @return 分页结果
|
||||
*/
|
||||
private BfiPageVO<BfiRecBillVO> pullRecBillPage(LocalDateTime sliceStart, LocalDateTime sliceEnd,
|
||||
List<String> orgNumbers, int pageNo) {
|
||||
BfiRecBillQueryDTO queryDTO = new BfiRecBillQueryDTO();
|
||||
queryDTO.setPageNo(pageNo);
|
||||
queryDTO.setPageSize(SYNC_PAGE_SIZE);
|
||||
queryDTO.setBeginBizdate(sliceStart.format(BFI_DATE_TIME_FORMATTER));
|
||||
queryDTO.setEndBizdate(sliceEnd.format(BFI_DATE_TIME_FORMATTER));
|
||||
queryDTO.setOrgNumber(orgNumbers);
|
||||
return bfiService.queryRecBill(queryDTO);
|
||||
}
|
||||
|
||||
/**
|
||||
* A301-1 单页数据处理：收款单主单粒度逐行校验入缓冲，累计满一批即独立事务入库。
|
||||
* 切片边界同秒的收款单可能在相邻两片各拉一次，按批内收款单号去重幂等跳过，不重不漏
|
||||
*
|
||||
* @param rows 本页拉取的收款单
|
||||
* @param state 整轮同步状态（待入库缓冲与统计累加）
|
||||
*/
|
||||
private void handleRecBillPage(List<BfiRecBillVO> rows, RecBillSyncState state) {
|
||||
state.pageCount++;
|
||||
state.pullCount += rows.size();
|
||||
for (BfiRecBillVO recBill : rows) {
|
||||
// 主单粒度：一张收款单一条流水，不展开流水分录；明细流水号取收款单号（幂等键，C104-2 按此定位合规流水）
|
||||
String detailSerialNo = trimToEmpty(recBill.getBillno());
|
||||
if (Func.isEmpty(detailSerialNo)) {
|
||||
log.warn("收款流水同步跳过：收款单缺少单号");
|
||||
state.skippedCount++;
|
||||
continue;
|
||||
}
|
||||
if (!state.batchSerials.add(detailSerialNo)) {
|
||||
// 切片边界重复拉取或同批脏数据重复单号，幂等跳过（upsert 本就按单号幂等）
|
||||
log.warn("收款流水同步跳过：收款单[{}]本批已拉取，跳过重复处理", detailSerialNo);
|
||||
state.skippedCount++;
|
||||
continue;
|
||||
}
|
||||
// 收款金额取主单 actrecamt（金蝶收款金额口径）
|
||||
BigDecimal receiptAmount = money(recBill.getActrecamt());
|
||||
if (receiptAmount.compareTo(BigDecimal.ZERO) <= 0) {
|
||||
// log.warn("收款流水同步跳过：收款单[{}]收款金额（actrecamt）为空或小于0", recBill.getBillno());
|
||||
state.skippedCount++;
|
||||
continue;
|
||||
}
|
||||
state.pending.add(new BfiRecBillRow(recBill, detailSerialNo, receiptAmount));
|
||||
if (state.pending.size() >= SYNC_BATCH_SIZE) {
|
||||
state.batchNo++;
|
||||
state.totalCount += this.syncBatch(state.batchNo, state.pending);
|
||||
state.pending.clear();
|
||||
state.batchSerials.clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -613,6 +633,44 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl<KingdeeReceiptFlowMa
|
||||
String detailSerialNo, BigDecimal receiptAmount) {
|
||||
}
|
||||
|
||||
/**
|
||||
* 收款流水整轮同步状态：待入库缓冲与统计累加（各切片共享，结束进日志摘要）
|
||||
*/
|
||||
private static final class RecBillSyncState {
|
||||
/**
|
||||
* 待入库缓冲：累计满 SYNC_BATCH_SIZE 即入库一批并清空
|
||||
*/
|
||||
final List<BfiRecBillRow> pending = new ArrayList<>();
|
||||
/**
|
||||
* 批内收款单号去重（含切片边界重复拉取）
|
||||
*/
|
||||
final Set<String> batchSerials = new HashSet<>();
|
||||
/**
|
||||
* A301-1 原始拉取收款单张数
|
||||
*/
|
||||
int pullCount;
|
||||
/**
|
||||
* 跳过条数（缺单号/金额异常/重复拉取）
|
||||
*/
|
||||
int skippedCount;
|
||||
/**
|
||||
* 新增+更新条数
|
||||
*/
|
||||
int totalCount;
|
||||
/**
|
||||
* 已入库批次数
|
||||
*/
|
||||
int batchNo;
|
||||
/**
|
||||
* 实际处理的翻页总页数
|
||||
*/
|
||||
int pageCount;
|
||||
/**
|
||||
* 实际处理的切片数
|
||||
*/
|
||||
int sliceCount;
|
||||
}
|
||||
|
||||
private Map<String, Object> candidateRow(FormalSettlement settlement, BigDecimal claimedAmount) {
|
||||
BigDecimal settlementAmount = money(settlement.getSettlementAmount());
|
||||
Map<String, Object> row = new LinkedHashMap<>();
|
||||
|
||||
Reference in New Issue
Block a user