From 0eba3f6573be545b213e3a318c954279a60cd283 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 14:29:34 +0800 Subject: [PATCH] =?UTF-8?q?perf:=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=E4=B8=BA=E6=B5=81=E5=BC=8F=E5=88=86?= =?UTF-8?q?=E6=89=B9=E5=85=A5=E5=BA=93=E6=AF=8F5000=E6=9D=A1=E4=B8=80?= =?UTF-8?q?=E6=89=B9=E7=8B=AC=E7=AB=8B=E4=BA=8B=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/impl/ReceiptFlowServiceImpl.java | 317 +++++++++--------- 1 file changed, 166 insertions(+), 151 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 1ea5162e..6eef4fae 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 @@ -39,6 +39,7 @@ import org.springblade.system.cache.SysCache; import org.springblade.system.cache.UserCache; import org.springblade.system.pojo.entity.Dept; import org.springblade.thirdparty.bfi.pojo.dto.BfiRecBillQueryDTO; +import org.springblade.thirdparty.bfi.pojo.vo.BfiPageVO; import org.springblade.thirdparty.bfi.pojo.vo.BfiRecBillFlowVO; import org.springblade.thirdparty.bfi.pojo.vo.BfiRecBillVO; import org.springblade.thirdparty.bfi.service.IBfiService; @@ -66,6 +67,7 @@ import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionTemplate; import java.math.BigDecimal; import java.math.RoundingMode; @@ -107,6 +109,14 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl recBills = queryRecBills(); - - if (recBills.isEmpty()) { - record(null, null, "sync", "手动同步流水", null, null, BigDecimal.ZERO, - "金蝶收款单未查询到数据"); - return 0; - } - - // 同步口径：A301-1 按业务日期窗口拉取金蝶收款单（收款处理），按流水分录展开， + int days = bfiProperties.getRecBillSync().getDays(); + 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); + long start = System.currentTimeMillis(); + // 同步口径：按业务日期窗口拉取金蝶收款单（收款处理），按流水分录展开， // 以金蝶明细流水号为幂等键 upsert 收款流水表；本地已认领金额/状态（claimedAmount/claimStatus） // 由认领、作废驱动，不被同步顺带覆盖，未匹配收款金额单独存 unclaimed_amount 作为认领上限。 - - // 第一遍：数据校验与过滤，主单+流水分录展开为流水行，收集有效的 detailSerialNo - log.info("收款流水同步开始数据校验，共 {} 张收款单", recBills.size()); - Map validRows = new LinkedHashMap<>(); + BfiRecBillQueryDTO queryDTO = new BfiRecBillQueryDTO(); + queryDTO.setBeginBizdate(beginBizdate); + queryDTO.setEndBizdate(endBizdate); + queryDTO.setPageSize(SYNC_PAGE_SIZE); + // 待入库缓冲：累计满 SYNC_BATCH_SIZE 即入库一批并清空 + List pending = new ArrayList<>(); + Set batchSerials = new HashSet<>(); + int totalCount = 0; + int batchNo = 0; + int pullCount = 0; int skippedCount = 0; - for (BfiRecBillVO recBill : recBills) { - List flowEntries = recBill.getInfoentry(); - if (Func.isEmpty(flowEntries)) { - log.warn("收款流水同步跳过：收款单[{}]无流水分录", recBill.getBillno()); - skippedCount++; - continue; + 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; } - for (BfiRecBillFlowVO flowEntry : flowEntries) { - // 明细流水号优先级：bei_detailid > bei_bankdetailno - String detailSerialNo = firstNonBlank(flowEntry.getBeiDetailid(), flowEntry.getBeiBankdetailno()); - if (Func.isEmpty(detailSerialNo)) { - log.warn("收款流水同步跳过：收款单[{}]流水分录缺少明细流水号", recBill.getBillno()); - skippedCount++; - continue; + pullCount += rows.size(); + for (BfiRecBillVO recBill : rows) { + skippedCount += expandRecBillRows(recBill, pending, batchSerials); + if (pending.size() >= SYNC_BATCH_SIZE) { + batchNo++; + totalCount += syncBatch(batchNo, pending); + pending.clear(); + batchSerials.clear(); } - detailSerialNo = detailSerialNo.trim(); - if (validRows.containsKey(detailSerialNo)) { - throw new ServiceException("明细流水号" + detailSerialNo + "重复"); - } - // 收款金额取主单入账金额 inneraccountamount（金蝶口径，非流水分录 bei_creditamount） - BigDecimal receiptAmount = money(recBill.getInneraccountAmount()); - if (receiptAmount.compareTo(BigDecimal.ZERO) <= 0) { - log.warn("收款流水同步跳过：收款单[{}]收款金额（inneraccountamount）为空或小于0", recBill.getBillno()); - skippedCount++; - continue; - } - validRows.put(detailSerialNo, new BfiRecBillRow(recBill, detailSerialNo, receiptAmount)); + } + // 翻页终止：末页标识、数据不满一页（防金蝶侧 lastPage 异常死循环，另有 SYNC_MAX_PAGES 兜底） + if (Boolean.TRUE.equals(page.getLastPage()) + || (page.getPageSize() != null && rows.size() < page.getPageSize())) { + break; } } - log.info("收款流水数据校验完成：有效 {} 条，跳过 {} 条", validRows.size(), skippedCount); - - if (validRows.isEmpty()) { - log.info("收款流水同步无有效数据，记录操作日志后返回"); + if (!pending.isEmpty()) { + batchNo++; + totalCount += syncBatch(batchNo, pending); + pending.clear(); + batchSerials.clear(); + } + if (pullCount == 0) { + record(null, null, "sync", "手动同步流水", null, null, BigDecimal.ZERO, + "金蝶收款单未查询到数据"); + } else if (totalCount == 0) { record(null, null, "sync", "手动同步流水", null, null, BigDecimal.ZERO, "金蝶收款单无有效数据"); - return 0; } + log.info("金蝶收款流水同步完成，拉取收款单 {} 张，新增/更新 {} 条，跳过 {} 条，分 {} 批入库，耗时 {}ms", + pullCount, totalCount, skippedCount, batchNo, System.currentTimeMillis() - start); + return totalCount; + } - // 批量查询已存在的记录（FOR UPDATE 防并发） - log.info("收款流水同步开始批量查询已存在记录，查询条件：detailSerialNo IN {} 条", validRows.size()); - List existingEntities = baseMapper.selectList( - Wrappers.lambdaQuery() - .in(KingdeeReceiptFlow::getDetailSerialNo, validRows.keySet()) - .last("FOR UPDATE") - ); - Map existingMap = existingEntities.stream() - .collect(Collectors.toMap(KingdeeReceiptFlow::getDetailSerialNo, e -> e)); - log.info("收款流水批量查询完成：已存在 {} 条，待新增 {} 条", - existingEntities.size(), validRows.size() - existingEntities.size()); - - // 第二遍：构建新增和更新列表 - log.info("收款流水同步开始构建新增/更新列表"); - List toInsert = new ArrayList<>(); - List toUpdate = new ArrayList<>(); - int createdCount = 0; - int updatedCount = 0; - - for (Map.Entry entry : validRows.entrySet()) { - String detailSerialNo = entry.getKey(); - BfiRecBillRow row = entry.getValue(); - - // 判断是新增还是更新 - KingdeeReceiptFlow existing = existingMap.get(detailSerialNo); - boolean created = (existing == null); - - KingdeeReceiptFlow entity; - if (created) { - // 新增：初始化本地认领金额和状态 - entity = new KingdeeReceiptFlow(); - entity.setClaimedAmount(BigDecimal.ZERO.setScale(2, RoundingMode.HALF_UP)); - entity.setClaimStatus(UNCLAIMED); - } else { - // 更新：保留本地已认领金额和状态 - entity = existing; + /** + * 单张收款单展开为流水行并做批内校验（无分录/缺流水号/金额非法跳过，批内流水号重复拦截） + * + * @return 跳过条数 + */ + private int expandRecBillRows(BfiRecBillVO recBill, List pending, Set batchSerials) { + int skipped = 0; + List flowEntries = recBill.getInfoentry(); + if (Func.isEmpty(flowEntries)) { + log.warn("收款流水同步跳过：收款单[{}]无流水分录", recBill.getBillno()); + return 1; + } + for (BfiRecBillFlowVO flowEntry : flowEntries) { + // 明细流水号优先级：bei_detailid > bei_bankdetailno + String detailSerialNo = firstNonBlank(flowEntry.getBeiDetailid(), flowEntry.getBeiBankdetailno()); + if (Func.isEmpty(detailSerialNo)) { + log.warn("收款流水同步跳过：收款单[{}]流水分录缺少明细流水号", recBill.getBillno()); + skipped++; + continue; } - - // 复制收款单数据到实体（不覆盖本地认领字段） - copyRecBillRow(entity, row); - // 根据本地累计认领金额与可认领上限重新计算认领状态 - entity.setClaimStatus(amountClaimStatus(money(entity.getClaimedAmount()), - claimCeiling(entity))); - - if (created) { - toInsert.add(entity); - createdCount++; - } else { - toUpdate.add(entity); - updatedCount++; + detailSerialNo = detailSerialNo.trim(); + if (!batchSerials.add(detailSerialNo)) { + throw new ServiceException("明细流水号" + detailSerialNo + "重复"); } + // 收款金额取主单入账金额 inneraccountamount（金蝶口径，非流水分录 bei_creditamount） + BigDecimal receiptAmount = money(recBill.getInneraccountAmount()); + if (receiptAmount.compareTo(BigDecimal.ZERO) <= 0) { + log.warn("收款流水同步跳过：收款单[{}]收款金额（inneraccountamount）为空或小于0", recBill.getBillno()); + skipped++; + continue; + } + pending.add(new BfiRecBillRow(recBill, detailSerialNo, receiptAmount)); } - log.info("收款流水列表构建完成：待新增 {} 条，待更新 {} 条", createdCount, updatedCount); + return skipped; + } - // 批量插入和更新 - if (!toInsert.isEmpty()) { - log.info("收款流水同步开始批量插入 {} 条", toInsert.size()); - saveBatch(toInsert); - log.info("收款流水批量插入完成"); - } - if (!toUpdate.isEmpty()) { - log.info("收款流水同步开始批量更新 {} 条", toUpdate.size()); - updateBatchById(toUpdate); - log.info("收款流水批量更新完成"); - } - - // 记录操作日志（批量插入，避免逐行 N+1） - log.info("收款流水同步开始记录操作日志，共 {} 条", toInsert.size() + toUpdate.size()); - List records = new ArrayList<>(toInsert.size() + toUpdate.size()); - for (KingdeeReceiptFlow entity : toInsert) { - records.add(buildRecord(entity.getId(), null, "sync", "新增金蝶收款流水", - entity.getClaimStatus(), entity.getClaimStatus(), BigDecimal.ZERO, entity.getDetailSerialNo())); - } - for (KingdeeReceiptFlow entity : toUpdate) { - records.add(buildRecord(entity.getId(), null, "sync", "更新金蝶收款流水", - entity.getClaimStatus(), entity.getClaimStatus(), BigDecimal.ZERO, entity.getDetailSerialNo())); - } - if (!records.isEmpty()) { - Db.saveBatch(records); - } - log.info("收款流水操作日志记录完成"); - - log.info("金蝶收款流水同步完成，拉取收款单 {} 张，新增 {} 条，更新 {} 条", - recBills.size(), createdCount, updatedCount); - return createdCount + updatedCount; + /** + * 单批入库（独立事务）：批内 FOR UPDATE 防并发，按明细流水号 upsert，操作日志批量落库 + * + * @param batchNo 批次号（日志用） + * @param rows 本批流水行 + * @return 本批新增+更新条数 + */ + private int syncBatch(int batchNo, List rows) { + Integer count = transactionTemplate.execute(status -> { + // 批量查询已存在的记录（FOR UPDATE 防并发） + Map existingMap = baseMapper.selectList( + Wrappers.lambdaQuery() + .in(KingdeeReceiptFlow::getDetailSerialNo, + rows.stream().map(BfiRecBillRow::detailSerialNo).toList()) + .last("FOR UPDATE")) + .stream().collect(Collectors.toMap(KingdeeReceiptFlow::getDetailSerialNo, e -> e)); + List toInsert = new ArrayList<>(); + List toUpdate = new ArrayList<>(); + for (BfiRecBillRow row : rows) { + KingdeeReceiptFlow existing = existingMap.get(row.detailSerialNo()); + boolean created = (existing == null); + KingdeeReceiptFlow entity; + if (created) { + // 新增：初始化本地认领金额和状态 + entity = new KingdeeReceiptFlow(); + entity.setClaimedAmount(BigDecimal.ZERO.setScale(2, RoundingMode.HALF_UP)); + entity.setClaimStatus(UNCLAIMED); + } else { + // 更新：保留本地已认领金额和状态 + entity = existing; + } + // 复制收款单数据到实体（不覆盖本地认领字段） + copyRecBillRow(entity, row); + // 根据本地累计认领金额与可认领上限重新计算认领状态 + entity.setClaimStatus(amountClaimStatus(money(entity.getClaimedAmount()), + claimCeiling(entity))); + if (created) { + toInsert.add(entity); + } else { + toUpdate.add(entity); + } + } + if (!toInsert.isEmpty()) { + saveBatch(toInsert); + } + if (!toUpdate.isEmpty()) { + updateBatchById(toUpdate); + } + // 记录操作日志（批量插入，避免逐行 N+1） + List records = new ArrayList<>(toInsert.size() + toUpdate.size()); + for (KingdeeReceiptFlow entity : toInsert) { + records.add(buildRecord(entity.getId(), null, "sync", "新增金蝶收款流水", + entity.getClaimStatus(), entity.getClaimStatus(), BigDecimal.ZERO, entity.getDetailSerialNo())); + } + for (KingdeeReceiptFlow entity : toUpdate) { + records.add(buildRecord(entity.getId(), null, "sync", "更新金蝶收款流水", + entity.getClaimStatus(), entity.getClaimStatus(), BigDecimal.ZERO, entity.getDetailSerialNo())); + } + if (!records.isEmpty()) { + Db.saveBatch(records); + } + return toInsert.size() + toUpdate.size(); + }); + log.info("收款流水同步第 {} 批入库完成，本批 {} 条", batchNo, count == null ? 0 : count); + return count == null ? 0 : count; } /** @@ -776,32 +817,6 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl queryRecBills() { - int days = bfiProperties.getRecBillSync().getDays(); - 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); - long start = System.currentTimeMillis(); - log.info("调用BFI接口 收款单查询开始，业务日期窗口：{} ~ {}（近{}天）", beginBizdate, endBizdate, days); - BfiRecBillQueryDTO queryDTO = new BfiRecBillQueryDTO(); - queryDTO.setBeginBizdate(beginBizdate); - queryDTO.setEndBizdate(endBizdate); - queryDTO.setPageSize(SYNC_PAGE_SIZE); - List recBills = BfiUtil.queryAllPagesByTotal(pageNo -> { - queryDTO.setPageNo(pageNo); - return bfiService.queryRecBill(queryDTO); - }); - log.info("调用BFI接口 收款单查询结束 耗时：{}ms，拉取 {} 条", System.currentTimeMillis() - start, recBills.size()); - return recBills; - } - /** * A301-1 收款单 → 收款流水表字段拷贝（仅覆盖金蝶侧来源字段，不覆盖本地认领金额与状态）。 * 认领上限口径：未匹配收款金额 unmatchamountrec 存 unclaimed_amount。