b2894lxlx/tms-erp-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: 收款流水同步改为流式分批入库每5000条一批独立事务

Browse Source
This commit is contained in:
刘泉佳
2026-10-29 14:29:34 +08:00
parent d2bd36b4f8
commit 0eba3f6573
1 changed files with 166 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/ReceiptFlowServiceImpl.java
+166 -151
View File
@@ -39,6 +39,7 @@ import org.springblade.system.cache.SysCache;
import org.springblade.system.cache.UserCache; import org.springblade.system.cache.UserCache;
import org.springblade.system.pojo.entity.Dept; import org.springblade.system.pojo.entity.Dept;
import org.springblade.thirdparty.bfi.pojo.dto.BfiRecBillQueryDTO; 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.BfiRecBillFlowVO;
import org.springblade.thirdparty.bfi.pojo.vo.BfiRecBillVO; import org.springblade.thirdparty.bfi.pojo.vo.BfiRecBillVO;
import org.springblade.thirdparty.bfi.service.IBfiService; 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.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;
import java.math.BigDecimal; import java.math.BigDecimal;
import java.math.RoundingMode; import java.math.RoundingMode;
@@ -107,6 +109,14 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl<KingdeeReceiptFlowMa
* A301-1 收款单查询分页大小 * A301-1 收款单查询分页大小
*/ */
private static final int SYNC_PAGE_SIZE = 200; private static final int SYNC_PAGE_SIZE = 200;
/**
* 同步分批入库批次大小：累计满一批即独立事务入库，避免全量驻留内存
*/
private static final int SYNC_BATCH_SIZE = 5000;
/**
* A301-1 翻页最大页数保护，防金蝶侧 lastPage 异常死循环（200条/页 × 500页）
*/
private static final int SYNC_MAX_PAGES = 500;
/** /**
* A301-1 业务日期时间格式（请求参数与返回 bizdate 均为 yyyy-MM-dd HH:mm:ss） * A301-1 业务日期时间格式（请求参数与返回 bizdate 均为 yyyy-MM-dd HH:mm:ss）
*/ */
@@ -120,6 +130,7 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl<KingdeeReceiptFlowMa
private final IFormalSettlementService formalSettlementService; private final IFormalSettlementService formalSettlementService;
private final IBfiService bfiService; private final IBfiService bfiService;
private final BfiProperties bfiProperties; private final BfiProperties bfiProperties;
private final TransactionTemplate transactionTemplate;
private final HcsClaimSyncSupport hcsClaimSyncSupport; private final HcsClaimSyncSupport hcsClaimSyncSupport;
@Override @Override
@@ -433,148 +444,178 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl<KingdeeReceiptFlowMa
return claim.getId(); return claim.getId();
} }
/**
* 手动同步金蝶收款流水：直连 A301-1 收款单查询，按流水分录展开、明细流水号幂等 upsert；
* 数据量大，流式翻页处理，累计满一批（5000 条流水行）即独立事务入库一批，不全量驻留内存
*
* @return 本次同步条数（新增+更新）
*/
@Override @Override
@Transactional(rollbackFor = Exception.class)
public int sync() { public int sync() {
List<BfiRecBillVO> recBills = queryRecBills(); int days = bfiProperties.getRecBillSync().getDays();
LocalDate endDate = LocalDate.now();
if (recBills.isEmpty()) { LocalDate startDate = endDate.minusDays(days);
record(null, null, "sync", "手动同步流水", null, null, BigDecimal.ZERO, // 金蝶 A301-1 仅接受带时分秒的日期参数，传纯日期会报 bos.filterSqlBuilder 时区时间格式错误
"金蝶收款单未查询到数据"); String beginBizdate = startDate.atStartOfDay().format(BFI_DATE_TIME_FORMATTER);
return 0; String endBizdate = endDate.atTime(23, 59, 59).format(BFI_DATE_TIME_FORMATTER);
} log.info("收款流水同步开始，业务日期窗口：{} ~ {}（近{}天）", beginBizdate, endBizdate, days);
long start = System.currentTimeMillis();
// 同步口径：A301-1 按业务日期窗口拉取金蝶收款单（收款处理），按流水分录展开， // 同步口径：按业务日期窗口拉取金蝶收款单（收款处理），按流水分录展开，
// 以金蝶明细流水号为幂等键 upsert 收款流水表；本地已认领金额/状态（claimedAmount/claimStatus） // 以金蝶明细流水号为幂等键 upsert 收款流水表；本地已认领金额/状态（claimedAmount/claimStatus）
// 由认领、作废驱动，不被同步顺带覆盖，未匹配收款金额单独存 unclaimed_amount 作为认领上限。 // 由认领、作废驱动，不被同步顺带覆盖，未匹配收款金额单独存 unclaimed_amount 作为认领上限。
BfiRecBillQueryDTO queryDTO = new BfiRecBillQueryDTO();
// 第一遍：数据校验与过滤，主单+流水分录展开为流水行，收集有效的 detailSerialNo queryDTO.setBeginBizdate(beginBizdate);
log.info("收款流水同步开始数据校验，共 {} 张收款单", recBills.size()); queryDTO.setEndBizdate(endBizdate);
Map<String, BfiRecBillRow> validRows = new LinkedHashMap<>(); queryDTO.setPageSize(SYNC_PAGE_SIZE);
// 待入库缓冲：累计满 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; int skippedCount = 0;
for (BfiRecBillVO recBill : recBills) { for (int pageNo = 1; pageNo <= SYNC_MAX_PAGES; pageNo++) {
List<BfiRecBillFlowVO> flowEntries = recBill.getInfoentry(); queryDTO.setPageNo(pageNo);
if (Func.isEmpty(flowEntries)) { BfiPageVO<BfiRecBillVO> page = bfiService.queryRecBill(queryDTO);
log.warn("收款流水同步跳过：收款单[{}]无流水分录", recBill.getBillno()); List<BfiRecBillVO> rows = page == null ? List.of() : page.getRows();
skippedCount++; if (Func.isEmpty(rows)) {
continue; break;
} }
for (BfiRecBillFlowVO flowEntry : flowEntries) { pullCount += rows.size();
// 明细流水号优先级：bei_detailid > bei_bankdetailno for (BfiRecBillVO recBill : rows) {
String detailSerialNo = firstNonBlank(flowEntry.getBeiDetailid(), flowEntry.getBeiBankdetailno()); skippedCount += expandRecBillRows(recBill, pending, batchSerials);
if (Func.isEmpty(detailSerialNo)) { if (pending.size() >= SYNC_BATCH_SIZE) {
log.warn("收款流水同步跳过：收款单[{}]流水分录缺少明细流水号", recBill.getBillno()); batchNo++;
skippedCount++; totalCount += syncBatch(batchNo, pending);
continue; pending.clear();
batchSerials.clear();
} }
detailSerialNo = detailSerialNo.trim(); }
if (validRows.containsKey(detailSerialNo)) { // 翻页终止：末页标识、数据不满一页（防金蝶侧 lastPage 异常死循环，另有 SYNC_MAX_PAGES 兜底）
throw new ServiceException("明细流水号" + detailSerialNo + "重复"); if (Boolean.TRUE.equals(page.getLastPage())
} || (page.getPageSize() != null && rows.size() < page.getPageSize())) {
// 收款金额取主单入账金额 inneraccountamount（金蝶口径，非流水分录 bei_creditamount） break;
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));
} }
} }
log.info("收款流水数据校验完成：有效 {} 条，跳过 {} 条", validRows.size(), skippedCount); if (!pending.isEmpty()) {
batchNo++;
if (validRows.isEmpty()) { totalCount += syncBatch(batchNo, pending);
log.info("收款流水同步无有效数据，记录操作日志后返回"); 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, 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<KingdeeReceiptFlow> existingEntities = baseMapper.selectList( *
Wrappers.<KingdeeReceiptFlow>lambdaQuery() * @return 跳过条数
.in(KingdeeReceiptFlow::getDetailSerialNo, validRows.keySet()) */
.last("FOR UPDATE") private int expandRecBillRows(BfiRecBillVO recBill, List<BfiRecBillRow> pending, Set<String> batchSerials) {
); int skipped = 0;
Map<String, KingdeeReceiptFlow> existingMap = existingEntities.stream() List<BfiRecBillFlowVO> flowEntries = recBill.getInfoentry();
.collect(Collectors.toMap(KingdeeReceiptFlow::getDetailSerialNo, e -> e)); if (Func.isEmpty(flowEntries)) {
log.info("收款流水批量查询完成：已存在 {} 条，待新增 {} 条", log.warn("收款流水同步跳过：收款单[{}]无流水分录", recBill.getBillno());
existingEntities.size(), validRows.size() - existingEntities.size()); return 1;
}
// 第二遍：构建新增和更新列表 for (BfiRecBillFlowVO flowEntry : flowEntries) {
log.info("收款流水同步开始构建新增/更新列表"); // 明细流水号优先级：bei_detailid > bei_bankdetailno
List<KingdeeReceiptFlow> toInsert = new ArrayList<>(); String detailSerialNo = firstNonBlank(flowEntry.getBeiDetailid(), flowEntry.getBeiBankdetailno());
List<KingdeeReceiptFlow> toUpdate = new ArrayList<>(); if (Func.isEmpty(detailSerialNo)) {
int createdCount = 0; log.warn("收款流水同步跳过：收款单[{}]流水分录缺少明细流水号", recBill.getBillno());
int updatedCount = 0; skipped++;
continue;
for (Map.Entry<String, BfiRecBillRow> 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;
} }
detailSerialNo = detailSerialNo.trim();
// 复制收款单数据到实体（不覆盖本地认领字段） if (!batchSerials.add(detailSerialNo)) {
copyRecBillRow(entity, row); throw new ServiceException("明细流水号" + detailSerialNo + "重复");
// 根据本地累计认领金额与可认领上限重新计算认领状态
entity.setClaimStatus(amountClaimStatus(money(entity.getClaimedAmount()),
claimCeiling(entity)));
if (created) {
toInsert.add(entity);
createdCount++;
} else {
toUpdate.add(entity);
updatedCount++;
} }
// 收款金额取主单入账金额 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()) { * 单批入库（独立事务）：批内 FOR UPDATE 防并发，按明细流水号 upsert，操作日志批量落库
log.info("收款流水同步开始批量插入 {} 条", toInsert.size()); *
saveBatch(toInsert); * @param batchNo 批次号（日志用）
log.info("收款流水批量插入完成"); * @param rows 本批流水行
} * @return 本批新增+更新条数
if (!toUpdate.isEmpty()) { */
log.info("收款流水同步开始批量更新 {} 条", toUpdate.size()); private int syncBatch(int batchNo, List<BfiRecBillRow> rows) {
updateBatchById(toUpdate); Integer count = transactionTemplate.execute(status -> {
log.info("收款流水批量更新完成"); // 批量查询已存在的记录（FOR UPDATE 防并发）
} Map<String, KingdeeReceiptFlow> existingMap = baseMapper.selectList(
Wrappers.<KingdeeReceiptFlow>lambdaQuery()
// 记录操作日志（批量插入，避免逐行 N+1） .in(KingdeeReceiptFlow::getDetailSerialNo,
log.info("收款流水同步开始记录操作日志，共 {} 条", toInsert.size() + toUpdate.size()); rows.stream().map(BfiRecBillRow::detailSerialNo).toList())
List<ReceiptFlowRecord> records = new ArrayList<>(toInsert.size() + toUpdate.size()); .last("FOR UPDATE"))
for (KingdeeReceiptFlow entity : toInsert) { .stream().collect(Collectors.toMap(KingdeeReceiptFlow::getDetailSerialNo, e -> e));
records.add(buildRecord(entity.getId(), null, "sync", "新增金蝶收款流水", List<KingdeeReceiptFlow> toInsert = new ArrayList<>();
entity.getClaimStatus(), entity.getClaimStatus(), BigDecimal.ZERO, entity.getDetailSerialNo())); List<KingdeeReceiptFlow> toUpdate = new ArrayList<>();
} for (BfiRecBillRow row : rows) {
for (KingdeeReceiptFlow entity : toUpdate) { KingdeeReceiptFlow existing = existingMap.get(row.detailSerialNo());
records.add(buildRecord(entity.getId(), null, "sync", "更新金蝶收款流水", boolean created = (existing == null);
entity.getClaimStatus(), entity.getClaimStatus(), BigDecimal.ZERO, entity.getDetailSerialNo())); KingdeeReceiptFlow entity;
} if (created) {
if (!records.isEmpty()) { // 新增：初始化本地认领金额和状态
Db.saveBatch(records); entity = new KingdeeReceiptFlow();
} entity.setClaimedAmount(BigDecimal.ZERO.setScale(2, RoundingMode.HALF_UP));
log.info("收款流水操作日志记录完成"); entity.setClaimStatus(UNCLAIMED);
} else {
log.info("金蝶收款流水同步完成，拉取收款单 {} 张，新增 {} 条，更新 {} 条", // 更新：保留本地已认领金额和状态
recBills.size(), createdCount, updatedCount); entity = existing;
return createdCount + updatedCount; }
// 复制收款单数据到实体（不覆盖本地认领字段）
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<ReceiptFlowRecord> 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<KingdeeReceiptFlowMa
return flow; return flow;
} }
/**
* A301-1 收款单查询：业务日期窗口全量翻页，窗口天数取 Nacos 配置 thirdParty.bfi.recBillSync.days（默认近 60 天）。
* 按 totalCount 推导总页数，避免窗口放大后受固定页数上限截断。
* 收款流水同步专用。
*/
private List<BfiRecBillVO> 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<BfiRecBillVO> recBills = BfiUtil.queryAllPagesByTotal(pageNo -> {
queryDTO.setPageNo(pageNo);
return bfiService.queryRecBill(queryDTO);
});
log.info("调用BFI接口 收款单查询结束 耗时：{}ms，拉取 {} 条", System.currentTimeMillis() - start, recBills.size());
return recBills;
}
/** /**
* A301-1 收款单 → 收款流水表字段拷贝（仅覆盖金蝶侧来源字段，不覆盖本地认领金额与状态）。 * A301-1 收款单 → 收款流水表字段拷贝（仅覆盖金蝶侧来源字段，不覆盖本地认领金额与状态）。
* 认领上限口径：未匹配收款金额 unmatchamountrec 存 unclaimed_amount。 * 认领上限口径：未匹配收款金额 unmatchamountrec 存 unclaimed_amount。