perf: 收款流水同步改为批量查询与批量保存（N+1→批量）

2026-09-28 15:59:39 +08:00
parent 4d62bb6df7
commit 978caa640a
1 changed files with 70 additions and 25 deletions
@@ -346,53 +346,98 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl<KingdeeReceiptFlowMa
 		// 同步口径：A301-2 按参与认领用户（当前登录用户真实姓名）拉取全量认领公告，
 		// 以金蝶明细流水号为幂等键 upsert 收款流水表；本地已认领金额/状态（claimedAmount/claimStatus）
 		// 由认领、作废驱动，不被公告顺带覆盖，公告侧状态单独存 kingdee_claim_status/unclaimed_amount。
-		Set<String> serialNumbers = new HashSet<>();
-		int syncedCount = 0;
-		int createdCount = 0;
-		int updatedCount = 0;
+		// 第一遍：收集有效的 detailSerialNo，跳过无效数据
+		Map<String, BfiClaimAnnounceVO> validAnnounces = new LinkedHashMap<>();
 		for (BfiClaimAnnounceVO announce : announces) {
-			// 明细流水号为幂等键，缺省依次回退金蝶交易明细编号、银行流水号
 			String detailSerialNo = firstNonBlank(announce.getDetailid(),
 				announce.getTradedetailno(), announce.getBankdetailno());
 			if (Func.isEmpty(detailSerialNo)) {
 				log.warn("收款流水同步跳过：认领公告[{}]缺少明细流水号", announce.getBillno());
 				continue;
 			}
-			if (!serialNumbers.add(detailSerialNo.trim())) {
-				throw new ServiceException("明细流水号" + detailSerialNo.trim() + "重复");
+			detailSerialNo = detailSerialNo.trim();
+			if (validAnnounces.containsKey(detailSerialNo)) {
+				throw new ServiceException("明细流水号" + detailSerialNo + "重复");
 			}
-			// 行级锁防并发：同一流水并发手动同步时避免重复落库或覆盖已认领状态
-			KingdeeReceiptFlow entity = baseMapper.selectOne(Wrappers.<KingdeeReceiptFlow>lambdaQuery()
-				.eq(KingdeeReceiptFlow::getDetailSerialNo, detailSerialNo.trim())
-				.last("FOR UPDATE"));
-			boolean created = entity == null;
+			BigDecimal receiptAmount = money(announce.getReamount());
+			if (receiptAmount.compareTo(BigDecimal.ZERO) <= 0) {
+				log.warn("收款流水同步跳过：认领公告[{}]收款金额为空或小于0", announce.getBillno());
+				continue;
+			}
+			validAnnounces.put(detailSerialNo, announce);
+		}
+		
+		if (validAnnounces.isEmpty()) {
+			record(null, null, "sync", "手动同步流水", null, null, BigDecimal.ZERO,
+				"金蝶认领公告无有效数据");
+			return 0;
+		}
+		
+		// 批量查询已存在的记录（FOR UPDATE 防并发）
+		List<KingdeeReceiptFlow> existingEntities = baseMapper.selectList(
+			Wrappers.<KingdeeReceiptFlow>lambdaQuery()
+				.in(KingdeeReceiptFlow::getDetailSerialNo, validAnnounces.keySet())
+				.last("FOR UPDATE")
+		);
+		Map<String, KingdeeReceiptFlow> existingMap = existingEntities.stream()
+			.collect(Collectors.toMap(KingdeeReceiptFlow::getDetailSerialNo, e -> e));
+		
+		// 第二遍：构建新增和更新列表
+		List<KingdeeReceiptFlow> toInsert = new ArrayList<>();
+		List<KingdeeReceiptFlow> toUpdate = new ArrayList<>();
+		int createdCount = 0;
+		int updatedCount = 0;
+		
+		for (Map.Entry<String, BfiClaimAnnounceVO> entry : validAnnounces.entrySet()) {
+			String detailSerialNo = entry.getKey();
+			BfiClaimAnnounceVO announce = entry.getValue();
+			BigDecimal receiptAmount = money(announce.getReamount());
+			
+			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;
 			}
-			BigDecimal receiptAmount = money(announce.getReamount());
-			if (receiptAmount.compareTo(BigDecimal.ZERO) <= 0) {
-				// 公告金额异常时跳过单条并告警，不阻断整批同步、不破坏已有流水
-				log.warn("收款流水同步跳过：认领公告[{}]收款金额为空或小于0", announce.getBillno());
-				continue;
-			}
-			copyAnnounceRow(entity, announce, detailSerialNo.trim(), receiptAmount);
+			
+			copyAnnounceRow(entity, announce, detailSerialNo, receiptAmount);
 			entity.setClaimStatus(amountClaimStatus(money(entity.getClaimedAmount()), receiptAmount));
+			
 			if (created) {
-				baseMapper.insert(entity);
+				toInsert.add(entity);
 				createdCount++;
 			} else {
-				baseMapper.updateById(entity);
+				toUpdate.add(entity);
 				updatedCount++;
 			}
-			record(entity.getId(), null, "sync", created ? "新增金蝶收款流水" : "更新金蝶收款流水",
-				entity.getClaimStatus(), entity.getClaimStatus(), BigDecimal.ZERO, detailSerialNo.trim());
-			syncedCount++;
 		}
+		
+		// 批量插入和更新
+		if (!toInsert.isEmpty()) {
+			saveBatch(toInsert);
+		}
+		if (!toUpdate.isEmpty()) {
+			updateBatchById(toUpdate);
+		}
+		
+		// 记录操作日志
+		for (KingdeeReceiptFlow entity : toInsert) {
+			record(entity.getId(), null, "sync", "新增金蝶收款流水",
+				entity.getClaimStatus(), entity.getClaimStatus(), BigDecimal.ZERO, entity.getDetailSerialNo());
+		}
+		for (KingdeeReceiptFlow entity : toUpdate) {
+			record(entity.getId(), null, "sync", "更新金蝶收款流水",
+				entity.getClaimStatus(), entity.getClaimStatus(), BigDecimal.ZERO, entity.getDetailSerialNo());
+		}
+		
 		log.info("金蝶收款流水同步完成 claimusers：{}，拉取 {} 条，新增 {} 条，更新 {} 条",
 			claimusers, announces.size(), createdCount, updatedCount);
-		return syncedCount;
+		return createdCount + updatedCount;
 	}

 	private Map<String, Object> candidateRow(FormalSettlement settlement) {