⚡ 优化金蝶客商定时同步消除N+1查询与逐条写入,快照纳入金蝶字段修复变更不落变更记录

This commit is contained in:
刘泉佳
2026-09-20 19:44:57 +08:00
parent 9a35d408dc
commit c5d65f5678
@@ -28,6 +28,7 @@ import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.toolkit.Db;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springblade.core.log.exception.ServiceException;
@@ -36,7 +37,6 @@ import org.springblade.core.secure.utils.AuthUtil;
import org.springblade.core.tool.jackson.JsonUtil;
import org.springblade.core.tool.utils.BeanUtil;
import org.springblade.core.tool.utils.Func;
import org.springblade.core.tool.utils.SpringUtil;
import org.springblade.system.pojo.entity.UserCustomerScope;
import org.springblade.thirdparty.bfi.constant.BfiConstant;
import org.springblade.thirdparty.bfi.pojo.dto.BfiCustomerQueryDTO;
@@ -46,6 +46,7 @@ import org.springblade.thirdparty.bfi.pojo.vo.BfiBillSaveResultVO;
import org.springblade.thirdparty.bfi.pojo.vo.BfiCustomerVO;
import org.springblade.thirdparty.bfi.pojo.vo.BfiPageVO;
import org.springblade.thirdparty.bfi.service.IBfiService;
import org.springblade.thirdparty.bfi.util.BfiUtil;
import org.springblade.transport.excel.CustomerArchiveExcel;
import org.springblade.transport.mapper.CreditRatingStandardMapper;
import org.springblade.transport.mapper.CreditScoreCategoryMapper;
@@ -104,6 +105,7 @@ import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* 客商档案 服务实现类
@@ -370,17 +372,18 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
updateById(after);
return after;
}
pushCustomerToKingdee(id, customer, after);
pushCustomerToKingdee(id, customer, after, detail(id));
updateById(after);
return after;
}
/**
* 组装 A902-2 报文推送金蝶并回写协同状态
* 组装 A902-2 报文推送金蝶并回写协同状态;detail 由调用方准备,
* 定时同步批量组装报文数据,避免逐条 detail 查询的 N+1 问题
*/
private void pushCustomerToKingdee(Long id, CustomerArchive customer, CustomerArchive after) {
private void pushCustomerToKingdee(Long id, CustomerArchive customer, CustomerArchive after, CustomerArchiveVO detail) {
try {
BfiBillSaveResultVO result = bfiService.saveCustomer(customerBfiConverter.buildSaveDTO(detail(id)));
BfiBillSaveResultVO result = bfiService.saveCustomer(customerBfiConverter.buildSaveDTO(detail));
BfiBillSaveResultItemVO item = firstSaveItem(result);
if (item != null && Boolean.TRUE.equals(item.getBillStatus())) {
after.setKingdeeStatus(KINGDEE_STATUS_SYNCED);
@@ -447,36 +450,116 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
}
/**
* 自代理:跨方法调用确保事务注解生效(单条推送、停用各自独立事务)
* 定时同步推送报文组装:主表字段 + 批量查询的收款账户,不含联系人/发票/评分等推送无关明细,
* 替代逐条 detail 查询,避免 N+1
*/
private ICustomerArchiveService self() {
return SpringUtil.getBean(ICustomerArchiveService.class);
private CustomerArchiveVO buildSyncArchiveVO(CustomerArchive customer, List<CustomerReceiptAccountVO> receiptAccounts) {
CustomerArchiveVO detail = Objects.requireNonNull(BeanUtil.copyProperties(customer, CustomerArchiveVO.class));
detail.setReceiptAccounts(receiptAccounts);
return detail;
}
/**
* 定时同步变更集中落库:档案批量更新 + 变更记录批量插入,失败仅记录日志,不影响汇总返回
*/
private void flushSyncChanges(List<CustomerArchive> updateList, List<CustomerChangeRecord> changeRecords) {
if (!updateList.isEmpty()) {
try {
updateBatchById(updateList);
} catch (Exception exception) {
log.error("定时同步金蝶客商批量更新档案异常,size={}", updateList.size(), exception);
}
}
if (!changeRecords.isEmpty()) {
try {
Db.saveBatch(changeRecords);
} catch (Exception exception) {
log.error("定时同步金蝶客商批量写入变更记录异常,size={}", changeRecords.size(), exception);
}
}
}
@Override
public String syncKingdeeCustomerData() {
log.info("金蝶客商数据定时同步开始");
// ① 待推送客商:审批通过且未推送/推送失败,复用单条推送(A902-1 复查 + A902-2 保存),单条失败不中断批次
List<CustomerArchive> pendingList = list(Wrappers.<CustomerArchive>lambdaQuery()
// 客商数据量不大,一次查询审批通过客商,内存按金蝶协同状态分组
List<CustomerArchive> approvedList = list(Wrappers.<CustomerArchive>lambdaQuery()
.eq(CustomerArchive::getApprovalStatus, APPROVAL_APPROVED)
.in(CustomerArchive::getKingdeeStatus, KINGDEE_STATUS_NONE, KINGDEE_STATUS_FAILED)
.eq(CustomerArchive::getCustomerCode, "KS000067")
.orderByAsc(CustomerArchive::getId));
log.info("金蝶客商数据定时同步查询到待推送客商 {} 条", pendingList.size());
ICustomerArchiveService proxy = self();
// ① 待推送客商:未推送/推送失败,单条失败不中断批次
List<CustomerArchive> pendingList = approvedList.stream()
.filter(customer -> KINGDEE_STATUS_NONE.equals(customer.getKingdeeStatus())
|| KINGDEE_STATUS_FAILED.equals(customer.getKingdeeStatus()))
.toList();
// ② 停用联动候选:已同步/金蝶已有且启用的客商
List<CustomerArchive> enabledList = approvedList.stream()
.filter(customer -> (KINGDEE_STATUS_SYNCED.equals(customer.getKingdeeStatus())
|| KINGDEE_STATUS_EXIST.equals(customer.getKingdeeStatus()))
&& Objects.equals(customer.getStatus(), STATUS_ENABLED))
.toList();
// ③ 已推送客商:A902-2 成功仅代表金蝶受理(status=B),复查 status=C 已审核才是最终成功
List<CustomerArchive> syncedList = approvedList.stream()
.filter(customer -> KINGDEE_STATUS_SYNCED.equals(customer.getKingdeeStatus()))
.toList();
log.info("金蝶客商数据定时同步查询到审批通过客商 {} 条(待推送 {},已推送 {},可停用联动 {})",
approvedList.size(), pendingList.size(), syncedList.size(), enabledList.size());
int pushed = 0;
int existed = 0;
int pushFailed = 0;
// 变更集中落库:档案批量更新、变更记录批量插入,避免逐条 update/insert 的 N+1 写入
List<CustomerArchive> updateList = new ArrayList<>();
List<CustomerChangeRecord> changeRecords = new ArrayList<>();
// 批量 A902-1 复查待推送客商:金蝶已审核直接建档(exist)、审核中置 failed,仅不存在金蝶档案的才逐条推送,避免逐条复查的 N+1 调用
Map<String, BfiCustomerVO> pendingKingdeeMap = queryKingdeeCustomerMap(pendingList.stream()
.map(CustomerArchive::getUnifiedCreditCode)
.filter(Func::isNotEmpty)
.map(code -> code.trim().toUpperCase())
.distinct()
.toList());
List<CustomerArchive> pushList = new ArrayList<>();
for (CustomerArchive customer : pendingList) {
BfiCustomerVO kingdeeCustomer = pendingKingdeeMap.get(customer.getUnifiedCreditCode().trim().toUpperCase());
if (kingdeeCustomer == null) {
pushList.add(customer);
continue;
}
CustomerArchive after = copyCustomer(customer);
after.setKingdeeSyncTime(LocalDateTime.now());
if (BfiConstant.BILL_STATUS_AUDITED.equals(kingdeeCustomer.getStatus())) {
// 金蝶已存在已审核档案:跳过推送,直接建档
after.setKingdeeStatus(KINGDEE_STATUS_EXIST);
after.setKingdeeFailReason(null);
existed++;
collectChangeRecord(changeRecords, customer.getId(), "金蝶已存在该客商档案,跳过推送", customer, after);
} else {
// 金蝶侧暂存/已提交:审核中,置失败待重推
after.setKingdeeStatus(KINGDEE_STATUS_FAILED);
after.setKingdeeFailReason("金蝶侧该客商审核中,请稍后重推");
pushFailed++;
collectChangeRecord(changeRecords, customer.getId(), "推送金蝶失败:金蝶侧客商审核中", customer, after);
}
log.info("金蝶客商数据定时同步复查处理完成,id={},unifiedCreditCode={},kingdeeStatus={}",
customer.getId(), customer.getUnifiedCreditCode(), after.getKingdeeStatus());
updateList.add(after);
}
// 金蝶不存在档案的逐条推送(A902-2 保存):
// 报文仅需主表字段 + 收款账户,按客商ID批量一次查询收款账户组装报文,避免逐条 detail 的 N+1 查询
Map<Long, List<CustomerReceiptAccountVO>> receiptAccountsMap = loadReceiptAccountsMap(pushList.stream()
.map(CustomerArchive::getId)
.toList());
for (CustomerArchive customer : pushList) {
try {
CustomerArchive after = proxy.syncKingdee(customer.getId());
CustomerArchiveVO detail = buildSyncArchiveVO(customer,
receiptAccountsMap.getOrDefault(customer.getId(), new ArrayList<>()));
CustomerArchive after = copyCustomer(customer);
after.setKingdeeSyncTime(LocalDateTime.now());
pushCustomerToKingdee(customer.getId(), customer, after, detail);
updateList.add(after);
if (Objects.equals(after.getKingdeeStatus(), KINGDEE_STATUS_SYNCED)) {
pushed++;
log.info("金蝶客商数据定时同步推送成功,id={},unifiedCreditCode={}",
customer.getId(), customer.getUnifiedCreditCode());
} else if (Objects.equals(after.getKingdeeStatus(), KINGDEE_STATUS_EXIST)) {
existed++;
log.info("金蝶客商数据定时同步金蝶已有档案跳过推送,id={},unifiedCreditCode={}",
customer.getId(), customer.getUnifiedCreditCode());
} else {
pushFailed++;
log.warn("金蝶客商数据定时同步推送失败,id={},unifiedCreditCode={},kingdeeStatus={},failReason={}",
@@ -488,34 +571,62 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
customer.getId(), customer.getUnifiedCreditCode(), exception);
}
}
// ② 已同步客商使用状态回写:金蝶禁用(enable=0)→ TMS 自动停用;金蝶恢复可用不自动启用(TMS 人工控制)
List<CustomerArchive> enabledList = list(Wrappers.<CustomerArchive>lambdaQuery()
.eq(CustomerArchive::getApprovalStatus, APPROVAL_APPROVED)
.in(CustomerArchive::getKingdeeStatus, KINGDEE_STATUS_SYNCED, KINGDEE_STATUS_EXIST)
.eq(CustomerArchive::getStatus, STATUS_ENABLED)
.orderByAsc(CustomerArchive::getId));
// ②③ 共用:对已推送(synced)/金蝶已有(exist)客商按信用代码批量复查金蝶档案(审核状态 + 使用状态),仅调用一次 A902-1
int disabled = 0;
int disableFailed = 0;
if (!enabledList.isEmpty()) {
log.info("金蝶客商数据定时同步查询到已同步可用客商 {} 条,开始联动金蝶使用状态", enabledList.size());
List<String> creditCodes = enabledList.stream()
int confirmed = 0;
int auditing = 0;
int unknown = 0;
if (!enabledList.isEmpty() || !syncedList.isEmpty()) {
List<String> reviewCreditCodes = Stream.concat(syncedList.stream(), enabledList.stream())
.map(CustomerArchive::getUnifiedCreditCode)
.filter(Func::isNotEmpty)
.map(code -> code.trim().toUpperCase())
.distinct()
.toList();
Map<String, BfiCustomerVO> kingdeeMap = queryKingdeeCustomerMap(creditCodes);
Map<String, BfiCustomerVO> kingdeeMap = reviewCreditCodes.isEmpty()
? Map.of()
: queryKingdeeCustomerMap(reviewCreditCodes);
// ③ 审核确认:A902-1 复查 status=C 已审核才是最终成功,synced 升级为 exist(终态);
// A/B 审核中保持 synced 等待下轮复查(不重推,避免金蝶侧重复建档)
for (CustomerArchive customer : syncedList) {
BfiCustomerVO kingdeeCustomer = kingdeeMap.get(customer.getUnifiedCreditCode().trim().toUpperCase());
if (kingdeeCustomer == null) {
unknown++;
log.warn("金蝶客商审核状态复查未查询到档案,id={},unifiedCreditCode={}",
customer.getId(), customer.getUnifiedCreditCode());
continue;
}
if (!BfiConstant.BILL_STATUS_AUDITED.equals(kingdeeCustomer.getStatus())) {
auditing++;
log.info("金蝶客商审核中,等待下轮复查,id={},unifiedCreditCode={},kingdeeStatus={}",
customer.getId(), customer.getUnifiedCreditCode(), kingdeeCustomer.getStatus());
continue;
}
confirmed++;
CustomerArchive after = copyCustomer(customer);
after.setKingdeeStatus(KINGDEE_STATUS_EXIST);
after.setKingdeeFailReason(null);
updateList.add(after);
collectChangeRecord(changeRecords, customer.getId(), "金蝶客商已审核,推送最终成功", customer, after);
log.info("金蝶客商已审核,推送最终成功并升级为金蝶已有档案,id={},unifiedCreditCode={}",
customer.getId(), customer.getUnifiedCreditCode());
}
// ② 停用联动:金蝶禁用(enable=0)→ TMS 自动停用;金蝶恢复可用不自动启用(TMS 人工控制);
// 直接基于已查询档案内存处理,避免逐条自代理调用重复查询档案与权限
for (CustomerArchive customer : enabledList) {
BfiCustomerVO kingdeeCustomer = kingdeeMap.get(customer.getUnifiedCreditCode().trim().toUpperCase());
if (kingdeeCustomer == null || !"0".equals(kingdeeCustomer.getEnable())) {
continue;
}
try {
if (proxy.disableCustomerFromKingdee(customer.getId())) {
disabled++;
log.info("金蝶侧客商已停用,联动停用客商档案成功,id={},unifiedCreditCode={}",
customer.getId(), customer.getUnifiedCreditCode());
}
CustomerArchive after = copyCustomer(customer);
after.setStatus(STATUS_DISABLED);
collectChangeRecord(changeRecords, customer.getId(), "金蝶侧客商已停用,同步停用客商档案", customer, after);
updateList.add(after);
disabled++;
log.info("金蝶侧客商已停用,联动停用客商档案成功,id={},unifiedCreditCode={}",
customer.getId(), customer.getUnifiedCreditCode());
} catch (Exception exception) {
disableFailed++;
log.error("定时同步金蝶客商停用异常,id={},unifiedCreditCode={}",
@@ -523,8 +634,11 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
}
}
}
flushSyncChanges(updateList, changeRecords);
String summary = "待推送 " + pendingList.size() + " 条(推送成功 " + pushed + ",金蝶已有 " + existed
+ ",失败 " + pushFailed + "),停用联动 " + disabled + " 条(失败 " + disableFailed + ")";
+ ",失败 " + pushFailed + "),停用联动 " + disabled + " 条(失败 " + disableFailed + ")"
+ ",已推送复查 " + syncedList.size() + " 条(金蝶已审核 " + confirmed + ",审核中 " + auditing
+ ",未查询到 " + unknown + ")";
log.info("金蝶客商数据定时同步结束:{}", summary);
return summary;
}
@@ -543,7 +657,12 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
}
/**
* 按统一社会信用代码分批查询金蝶客商,键为大写信用代码
* 按统一社会信用代码分批批量查询金蝶客商,键为大写信用代码;
* societycreditcode 支持数组一次查询多个,单批使用 BfiUtil.queryAllPages 自动翻页取全,
* 避免同一信用代码多条记录时漏查,单批查询异常不中断其余批次
*
* @param creditCodes 统一社会信用代码集合
* @return 大写信用代码 -> 金蝶客商
*/
private Map<String, BfiCustomerVO> queryKingdeeCustomerMap(List<String> creditCodes) {
Map<String, BfiCustomerVO> result = new HashMap<>();
@@ -554,16 +673,15 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
}
try {
BfiCustomerQueryDTO param = new BfiCustomerQueryDTO();
param.setPageNo(1);
param.setPageSize(batch.size());
param.setSocietycreditcode(batch);
BfiPageVO<BfiCustomerVO> page = bfiService.queryCustomer(param);
List<BfiCustomerVO> rows = page == null ? null : page.getRows();
if (rows != null) {
for (BfiCustomerVO vo : rows) {
if (Func.isNotEmpty(vo.getSocietycreditcode())) {
result.put(vo.getSocietycreditcode().trim().toUpperCase(), vo);
}
List<BfiCustomerVO> rows = BfiUtil.queryAllPages(pageNo -> {
param.setPageNo(pageNo);
param.setPageSize(KINGDEE_CUSTOMER_QUERY_BATCH);
return bfiService.queryCustomer(param);
});
for (BfiCustomerVO vo : rows) {
if (Func.isNotEmpty(vo.getSocietycreditcode())) {
result.put(vo.getSocietycreditcode().trim().toUpperCase(), vo);
}
}
} catch (Exception exception) {
@@ -983,12 +1101,26 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
}
private List<CustomerReceiptAccountVO> loadReceiptAccounts(Long customerId) {
return loadReceiptAccountsMap(List.of(customerId)).getOrDefault(customerId, new ArrayList<>());
}
/**
* 按客商ID集合批量查询收款账户(键为客商ID),供定时同步批量组装金蝶报文等场景使用,避免逐条查询
*
* @param customerIds 客商ID集合
* @return 客商ID -> 收款账户列表(默认一条扣款排序)
*/
private Map<Long, List<CustomerReceiptAccountVO>> loadReceiptAccountsMap(List<Long> customerIds) {
if (Func.isEmpty(customerIds)) {
return Map.of();
}
return receiptAccountMapper.selectList(Wrappers.<CustomerReceiptAccount>lambdaQuery()
.eq(CustomerReceiptAccount::getCustomerId, customerId)
.in(CustomerReceiptAccount::getCustomerId, customerIds)
.eq(CustomerReceiptAccount::getIsDeleted, 0)
.orderByDesc(CustomerReceiptAccount::getIsDefault)
.orderByAsc(CustomerReceiptAccount::getCreateTime))
.stream().map(account -> Objects.requireNonNull(BeanUtil.copyProperties(account, CustomerReceiptAccountVO.class))).toList();
.stream().map(account -> Objects.requireNonNull(BeanUtil.copyProperties(account, CustomerReceiptAccountVO.class)))
.collect(Collectors.groupingBy(CustomerReceiptAccountVO::getCustomerId));
}
private List<CustomerInvoiceInfoVO> loadInvoices(Long customerId) {
@@ -1309,11 +1441,10 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
}
private void addChangeRecord(Long customerId, String content, CustomerArchive before, CustomerArchive after) {
Map<String, Object> beforeSnapshot = customerSnapshot(before);
Map<String, Object> afterSnapshot = customerSnapshot(after);
addDetailSnapshot(beforeSnapshot, null);
addDetailSnapshot(afterSnapshot, null);
writeChangeRecord(customerId, content, beforeSnapshot, afterSnapshot);
CustomerChangeRecord record = buildChangeRecord(customerId, content, before, after);
if (record != null) {
changeRecordMapper.insert(record);
}
}
private void addChangeRecord(Long customerId, String content, CustomerArchive before, CustomerArchive after,
@@ -1322,10 +1453,35 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
Map<String, Object> afterSnapshot = customerSnapshot(after);
addDetailSnapshot(beforeSnapshot, beforeDetail);
addDetailSnapshot(afterSnapshot, afterDetail);
writeChangeRecord(customerId, content, beforeSnapshot, afterSnapshot);
CustomerChangeRecord record = buildChangeRecord(customerId, content, beforeSnapshot, afterSnapshot);
if (record != null) {
changeRecordMapper.insert(record);
}
}
private void writeChangeRecord(Long customerId, String content, Map<String, Object> beforeSnapshot,
/**
* 收集变更记录供定时同步集中批量插入(无实质变更不收集),避免逐条 insert
*/
private void collectChangeRecord(List<CustomerChangeRecord> records, Long customerId, String content,
CustomerArchive before, CustomerArchive after) {
CustomerChangeRecord record = buildChangeRecord(customerId, content, before, after);
if (record != null) {
records.add(record);
}
}
/**
* 构建变更记录(无实质变更返回 null),供单条写入与定时同步批量收集共用
*/
private CustomerChangeRecord buildChangeRecord(Long customerId, String content, CustomerArchive before, CustomerArchive after) {
Map<String, Object> beforeSnapshot = customerSnapshot(before);
Map<String, Object> afterSnapshot = customerSnapshot(after);
addDetailSnapshot(beforeSnapshot, null);
addDetailSnapshot(afterSnapshot, null);
return buildChangeRecord(customerId, content, beforeSnapshot, afterSnapshot);
}
private CustomerChangeRecord buildChangeRecord(Long customerId, String content, Map<String, Object> beforeSnapshot,
Map<String, Object> afterSnapshot) {
for (String field : afterSnapshot.keySet()) {
beforeSnapshot.putIfAbsent(field, null);
@@ -1346,7 +1502,7 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
}
}
if (changedFields.isEmpty()) {
return;
return null;
}
CustomerChangeRecord record = new CustomerChangeRecord();
record.setId(IdWorker.getId());
@@ -1358,7 +1514,7 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
record.setAfterData(JsonUtil.toJson(afterData));
record.setChangeUserName(AuthUtil.getUserName());
record.setStatus(STATUS_ENABLED);
changeRecordMapper.insert(record);
return record;
}
private boolean isSameSnapshotValue(Object beforeValue, Object afterValue) {
@@ -1488,6 +1644,10 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
snapshot.put("当前节点", customer.getCurrentNode());
snapshot.put("当前处理人", customer.getCurrentProcessor());
snapshot.put("审核通过时间", customer.getApprovedTime());
// 金蝶协同字段纳入快照:金蝶推送/审核确认/失败重推等状态变更需要落变更记录,否则前后快照一致会被判定为无实质变更而跳过
snapshot.put("金蝶协同状态", customer.getKingdeeStatus());
snapshot.put("金蝶失败原因", customer.getKingdeeFailReason());
snapshot.put("金蝶同步时间", customer.getKingdeeSyncTime());
snapshot.put("状态", customer.getStatus());
return snapshot;
}