From 2eded6b20ff0deae4824563cfd2ac7a6547038c6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=88=98=E6=B3=89=E4=BD=B3?= <1107186916@qq.com> Date: Thu, 29 Oct 2026 17:26:29 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=94=B6=E6=AC=BE=E6=B5=81=E6=B0=B4?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=E6=94=B9=E7=94=A8=E4=BA=8C=E5=88=86=E5=88=87?= =?UTF-8?q?=E7=89=87=E7=BB=AD=E6=89=AB=E4=BF=AE=E5=A4=8D=E7=AA=97=E5=8F=A3?= =?UTF-8?q?=E8=B6=85=E9=99=90=E6=BC=8F=E5=90=8C=E6=AD=A5=E5=B0=BE=E9=83=A8?= =?UTF-8?q?=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/impl/ReceiptFlowServiceImpl.java | 202 +++++++++++------- 1 file changed, 130 insertions(+), 72 deletions(-) diff --git a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/ReceiptFlowServiceImpl.java b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/ReceiptFlowServiceImpl.java index 9ba21d92..5c6da62f 100644 --- a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/ReceiptFlowServiceImpl.java +++ b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/ReceiptFlowServiceImpl.java @@ -445,8 +445,10 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl orgNumbers = baseMapper.selectPlatformOrgNumbers(); if (Func.isEmpty(orgNumbers)) { throw new ServiceException("组织管理中未维护平台公司金蝶编码，无法同步金蝶收款流水，请先检查组织数据"); } - queryDTO.setOrgNumber(orgNumbers); log.info("收款流水同步按组织过滤，组织编码：{}", orgNumbers); - // 待入库缓冲：累计满 SYNC_BATCH_SIZE 即入库一批并清空 - List pending = new ArrayList<>(); - Set 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 page = bfiService.queryRecBill(queryDTO); - List 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 pullRecBillPage(LocalDateTime sliceStart, LocalDateTime sliceEnd, + List 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 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 pending = new ArrayList<>(); + /** + * 批内收款单号去重（含切片边界重复拉取） + */ + final Set batchSerials = new HashSet<>(); + /** + * A301-1 原始拉取收款单张数 + */ + int pullCount; + /** + * 跳过条数（缺单号/金额异常/重复拉取） + */ + int skippedCount; + /** + * 新增+更新条数 + */ + int totalCount; + /** + * 已入库批次数 + */ + int batchNo; + /** + * 实际处理的翻页总页数 + */ + int pageCount; + /** + * 实际处理的切片数 + */ + int sliceCount; + } + private Map candidateRow(FormalSettlement settlement, BigDecimal claimedAmount) { BigDecimal settlementAmount = money(settlement.getSettlementAmount()); Map row = new LinkedHashMap<>();