feat: 金蝶定时同步加single-flight分布式锁,多副本与手动触发共用一道闸(付款结果回写/应收勾稽/开票结果/收票结果/客商同步)

- 新增 KingdeeSyncLockSupport:零等待tryLock+租约可配+租户隔离,未启用blade.lock.enabled时降级并告警
- 各同步方法拆为锁壳+executeXxxSync主体,锁在service层保证手动触发端点与定时任务行为一致
- 租约配置:thirdParty.bfi.{payResult|arResult|invoiceResult|receivedInvoice|customer}.lockLeaseSeconds,默认1800秒
This commit is contained in:
刘泉佳
2026-09-28 00:52:12 +08:00
parent 50573e7263
commit 7e76964a76
6 changed files with 210 additions and 0 deletions
@@ -88,6 +88,8 @@ import org.springblade.transport.pojo.vo.CustomerInvoiceContactVO;
import org.springblade.transport.pojo.vo.CustomerInvoiceInfoVO;
import org.springblade.transport.pojo.vo.CustomerReceiptAccountVO;
import org.springblade.transport.service.ICustomerArchiveService;
import org.springblade.transport.support.KingdeeSyncLockSupport;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -213,6 +215,17 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
private final PaymentApplicationMapper paymentApplicationMapper;
private final IBfiService bfiService;
private final CustomerBfiConverter customerBfiConverter;
private final KingdeeSyncLockSupport kingdeeSyncLockSupport;
/** 金蝶客商数据同步 single-flight 锁键前缀(多副本定时任务与手动触发共用) */
private static final String CUSTOMER_SYNC_LOCK_PREFIX = "transport:customer-archive:kingdee-customer-sync:";
/**
* 金蝶客商数据同步 single-flight 锁租约秒数,须大于最坏一轮耗时(复查 + 批量推送 + 停用联动),
* 否则租约到期后另一副本会并发起跑;进程异常退出时锁在租约到期后自动释放
*/
@Value("${thirdParty.bfi.customer.lockLeaseSeconds:1800}")
private long customerSyncLockLeaseSeconds;
@Override
public IPage<CustomerArchiveVO> selectCustomerArchivePage(IPage<CustomerArchiveVO> page, CustomerArchiveVO customer) {
@@ -622,6 +635,17 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl<CustomerArchiveM
@Override
public String syncKingdeeCustomerData() {
return kingdeeSyncLockSupport.runSingleFlight(CUSTOMER_SYNC_LOCK_PREFIX, customerSyncLockLeaseSeconds,
"金蝶客商数据同步", this::executeCustomerDataSync);
}
/**
* 金蝶客商数据同步主体:A902-1 复查 + A902-2 推送 + 金蝶禁用联动停用客商档案,
* 由 {@link #syncKingdeeCustomerData()} 在分布式锁保护下调用
*
* @return 执行摘要
*/
private String executeCustomerDataSync() {
log.info("金蝶客商数据定时同步开始");
// 客商数据量不大,一次查询审批通过客商,内存按金蝶协同状态分组
List<CustomerArchive> approvedList = list(Wrappers.<CustomerArchive>lambdaQuery()
@@ -80,11 +80,13 @@ import org.springblade.transport.service.IWaybillService;
import org.springblade.transport.service.KingdeeSyncContext;
import org.springblade.transport.support.KingdeeApBillSyncSupport;
import org.springblade.transport.support.KingdeeArBillSyncSupport;
import org.springblade.transport.support.KingdeeSyncLockSupport;
import org.springblade.transport.wrapper.PreSettlementWrapper;
import org.springblade.core.tool.jackson.JsonUtil;
import org.springblade.transport.wrapper.FormalSettlementWrapper;
import org.springblade.transport.wrapper.PaymentApplicationWrapper;
import org.springblade.system.cache.UserCache;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -114,6 +116,9 @@ import java.util.stream.Collectors;
public class FormalSettlementServiceImpl extends BaseServiceImpl<FormalSettlementMapper, FormalSettlement>
implements IFormalSettlementService {
/** 金蝶应收单勾稽 single-flight 锁键前缀(多副本定时任务与手动触发共用) */
private static final String AR_RESULT_SYNC_LOCK_PREFIX = "transport:formal-settlement:kingdee-ar-result-sync:";
private final FormalSettlementSourceMapper sourceMapper;
private final FormalSettlementSummaryFeeMapper summaryFeeMapper;
private final FormalSettlementPaymentMapper paymentMapper;
@@ -139,6 +144,14 @@ public class FormalSettlementServiceImpl extends BaseServiceImpl<FormalSettlemen
private final IWaybillService waybillService;
private final KingdeeApBillSyncSupport kingdeeApBillSyncSupport;
private final KingdeeArBillSyncSupport kingdeeArBillSyncSupport;
private final KingdeeSyncLockSupport kingdeeSyncLockSupport;
/**
* 金蝶应收单勾稽 single-flight 锁租约秒数,须大于最坏一轮耗时(批量远程查询 + 逐单纠偏回写),
* 否则租约到期后另一副本会并发起跑;进程异常退出时锁在租约到期后自动释放
*/
@Value("${thirdParty.bfi.arResult.lockLeaseSeconds:1800}")
private long arResultLockLeaseSeconds;
@Override
public IPage<FormalSettlementVO> selectPage(IPage<FormalSettlement> page, FormalSettlementVO query) {
@@ -523,6 +536,17 @@ public class FormalSettlementServiceImpl extends BaseServiceImpl<FormalSettlemen
@Override
public String syncKingdeeArResult() {
return kingdeeSyncLockSupport.runSingleFlight(AR_RESULT_SYNC_LOCK_PREFIX, arResultLockLeaseSeconds,
"金蝶应收单勾稽", this::executeArResultSync);
}
/**
* 金蝶应收单勾稽主体:A302-4 批量查询财务应收单回填单据编号、生成失败纠偏为已同步,
* 由 {@link #syncKingdeeArResult()} 在分布式锁保护下调用
*
* @return 执行摘要
*/
private String executeArResultSync() {
// 待勾稽应收结算单:审批通过 + 结算类型应收 + 金蝶状态为 已同步/生成失败(已同步一并轮询为补齐单号,生成失败纳入勾稽纠偏:
// 金蝶侧已生成说明上次推送实际成功,查到即回填纠正为已同步,避免人工重新推送产生重复单据)
List<FormalSettlement> settlements = list(Wrappers.<FormalSettlement>lambdaQuery()
@@ -93,7 +93,9 @@ import org.springblade.transport.pojo.vo.InvoiceApplicationVO;
import org.springblade.transport.service.IInvoiceApplicationService;
import org.springblade.transport.support.KingdeeApBillSyncSupport;
import org.springblade.transport.support.KingdeeArBillSyncSupport;
import org.springblade.transport.support.KingdeeSyncLockSupport;
import org.springblade.transport.wrapper.InvoiceApplicationWrapper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -127,6 +129,8 @@ public class InvoiceApplicationServiceImpl extends BaseServiceImpl<InvoiceApplic
implements IInvoiceApplicationService {
/** A303-5 开票结果批量查询单批单据编号上限(接口限制 50) */
private static final int INVOICE_RESULT_QUERY_BATCH = 50;
/** 金蝶开票结果回写 single-flight 锁键前缀(多副本定时任务与手动触发共用) */
private static final String INVOICE_RESULT_SYNC_LOCK_PREFIX = "transport:invoice-application:kingdee-invoice-result-sync:";
private final InvoiceApplicationSettlementMapper settlementRelationMapper;
private final InvoiceApplicationSheetMapper sheetMapper;
private final WaybillMapper waybillMapper;
@@ -145,6 +149,14 @@ public class InvoiceApplicationServiceImpl extends BaseServiceImpl<InvoiceApplic
private final BfiProperties bfiProperties;
private final KingdeeApBillSyncSupport kingdeeApBillSyncSupport;
private final KingdeeArBillSyncSupport kingdeeArBillSyncSupport;
private final KingdeeSyncLockSupport kingdeeSyncLockSupport;
/**
* 金蝶开票结果回写 single-flight 锁租约秒数,须大于最坏一轮耗时(分批远程查询 + 逐单回写),
* 否则租约到期后另一副本会并发起跑;进程异常退出时锁在租约到期后自动释放
*/
@Value("${thirdParty.bfi.invoiceResult.lockLeaseSeconds:1800}")
private long invoiceResultLockLeaseSeconds;
@Override
public IPage<InvoiceApplicationVO> selectPage(IPage<InvoiceApplication> page, InvoiceApplicationVO query) {
@@ -780,6 +792,17 @@ public class InvoiceApplicationServiceImpl extends BaseServiceImpl<InvoiceApplic
@Override
public String syncKingdeeInvoiceResult() {
return kingdeeSyncLockSupport.runSingleFlight(INVOICE_RESULT_SYNC_LOCK_PREFIX, invoiceResultLockLeaseSeconds,
"金蝶开票结果回写", this::executeInvoiceResultSync);
}
/**
* 金蝶开票结果回写主体:A303-5 分批查询开票申请发票结果落库并推进开票完成,
* 由 {@link #syncKingdeeInvoiceResult()} 在分布式锁保护下调用
*
* @return 执行摘要
*/
private String executeInvoiceResultSync() {
// 待回写开票申请:审批通过 + 已同步(invoiced 为终态不再轮询;failed 无真实金蝶单号不参与查询,纠偏走重推预检)
List<InvoiceApplication> applications = list(Wrappers.<InvoiceApplication>lambdaQuery()
.eq(InvoiceApplication::getApprovalStatus, TransportBusiConstant.APPROVED)
@@ -75,6 +75,7 @@ import org.springblade.transport.service.IInvoiceReceiptService;
import org.springblade.transport.service.KingdeeSyncContext;
import org.springblade.transport.support.KingdeeApBillSyncSupport;
import org.springblade.transport.support.KingdeeReceivedInvoiceSyncSupport;
import org.springblade.transport.support.KingdeeSyncLockSupport;
import org.springblade.transport.wrapper.InvoiceReceiptWrapper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
@@ -122,6 +123,10 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
private final IBfiService bfiService;
private final KingdeeApBillSyncSupport kingdeeApBillSyncSupport;
private final KingdeeReceivedInvoiceSyncSupport kingdeeReceivedInvoiceSyncSupport;
private final KingdeeSyncLockSupport kingdeeSyncLockSupport;
/** 金蝶收票单结果回写 single-flight 锁键前缀(多副本定时任务与手动触发共用) */
private static final String RECEIVED_RESULT_SYNC_LOCK_PREFIX = "transport:invoice-receipt:kingdee-received-result-sync:";
/** A801-3 推送幂等预检回看窗口天数(金蝶修改时间窗口,契约约束不超过 1 个月) */
@Value("${thirdParty.bfi.receivedInvoice.queryWindowDays:30}")
@@ -131,6 +136,13 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
@Value("${thirdParty.bfi.receivedInvoice.resultWindowDays:30}")
private int receivedResultWindowDays;
/**
* 金蝶收票单结果回写 single-flight 锁租约秒数,须大于最坏一轮耗时(批量远程查询 + 逐单回写),
* 否则租约到期后另一副本会并发起跑;进程异常退出时锁在租约到期后自动释放
*/
@Value("${thirdParty.bfi.receivedInvoice.lockLeaseSeconds:1800}")
private long receivedResultLockLeaseSeconds;
@Override
public IPage<InvoiceReceiptVO> selectPage(IPage<InvoiceReceipt> page, InvoiceReceiptVO query) {
LambdaQueryWrapper<InvoiceReceipt> wrapper = Wrappers.<InvoiceReceipt>lambdaQuery()
@@ -869,6 +881,17 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
@Override
public String syncKingdeeReceivedInvoiceResult() {
return kingdeeSyncLockSupport.runSingleFlight(RECEIVED_RESULT_SYNC_LOCK_PREFIX, receivedResultLockLeaseSeconds,
"金蝶收票单结果回写", this::executeReceivedInvoiceResultSync);
}
/**
* 金蝶收票单结果回写主体:A801-3 查询收票单结果推进异常回收/指定补偿/单号补登,
* 由 {@link #syncKingdeeReceivedInvoiceResult()} 在分布式锁保护下调用
*
* @return 执行摘要
*/
private String executeReceivedInvoiceResultSync() {
List<InvoiceReceipt> pendingReceipts = list(Wrappers.<InvoiceReceipt>lambdaQuery()
.in(InvoiceReceipt::getKingdeePushStatus, KingdeeBfiConstant.RECEIVED_PUSH_PUSHED,
KingdeeBfiConstant.RECEIVED_PUSH_ASSIGNED)
@@ -94,7 +94,9 @@ import org.springblade.transport.service.IFormalSettlementService;
import org.springblade.transport.service.IPaymentApplicationService;
import org.springblade.transport.service.KingdeeSyncContext;
import org.springblade.transport.support.KingdeeApBillSyncSupport;
import org.springblade.transport.support.KingdeeSyncLockSupport;
import org.springblade.transport.wrapper.PaymentApplicationWrapper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -124,6 +126,8 @@ public class PaymentApplicationServiceImpl extends BaseServiceImpl<PaymentApplic
private static final int PAY_RESULT_PAGE_SIZE = 200;
/** 历史模拟付款记录单号前缀 */
private static final String MOCK_PAYMENT_RECORD_PREFIX = "MOCK-PAY-";
/** 金蝶付款结果回写 single-flight 锁键前缀(多副本定时任务与手动触发共用) */
private static final String PAY_RESULT_SYNC_LOCK_PREFIX = "transport:payment-application:kingdee-pay-result-sync:";
private final PaymentApplicationInvoiceMapper invoiceMapper;
private final PaymentApplicationRecordMapper recordMapper;
private final PaymentApplicationSettlementMapper settlementRelationMapper;
@@ -144,6 +148,14 @@ private final CustomerReceiptAccountMapper receiptAccountMapper;
private final IBfiService bfiService;
private final KingdeeApBillSyncSupport kingdeeApBillSyncSupport;
private final BfiProperties bfiProperties;
private final KingdeeSyncLockSupport kingdeeSyncLockSupport;
/**
* 金蝶付款结果回写 single-flight 锁租约秒数,须大于最坏一轮耗时(分批远程查询 + 逐单回写),
* 否则租约到期后另一副本会并发起跑;进程异常退出时锁在租约到期后自动释放
*/
@Value("${thirdParty.bfi.payResult.lockLeaseSeconds:1800}")
private long payResultLockLeaseSeconds;
@Override
public IPage<PaymentApplicationVO> selectPage(IPage<PaymentApplication> page, PaymentApplicationVO query) {
@@ -491,6 +503,17 @@ private final CustomerReceiptAccountMapper receiptAccountMapper;
@Override
public String syncKingdeePaymentResult() {
return kingdeeSyncLockSupport.runSingleFlight(PAY_RESULT_SYNC_LOCK_PREFIX, payResultLockLeaseSeconds,
"金蝶付款结果回写", this::executePaymentResultSync);
}
/**
* 金蝶付款结果回写主体:A103-2 付款申请单查询回填 + A103-4 付款处理查询补付款记录,
* 由 {@link #syncKingdeePaymentResult()} 在分布式锁保护下调用
*
* @return 执行摘要
*/
private String executePaymentResultSync() {
// ① 待回写付款申请:审批通过且金蝶状态为 已生成/付款中/生成失败(终态已付款/已关闭不再轮询)。
// 生成失败单据一并纳入勾稽:金蝶侧已生成说明上次推送实际成功(如下推请求超时但远端已落单、本地事务回滚),
// 查到即回填纠正为已生成,避免人工重新推送产生重复单据;未查到则保持生成失败等待异常处理后重新推送
@@ -0,0 +1,93 @@
/**
* BladeX Commercial License Agreement
* Copyright (c) 2018-2099, https://bladex.cn. All rights reserved.
* <p>
* Use of this software is governed by the Commercial License Agreement obtained after purchasing a license from BladeX.
* <p>
* 1. This software is for development use only under a valid license from BladeX.
* <p>
* 2. Redistribution of this software's source code to any third party without a commercial license is strictly prohibited.
* <p>
* 3. Licensees may copyright their own code but cannot use segments from this software for such purposes.
* Copyright of this software remains with BladeX.
* <p>
* Using this software signifies agreement to this License, and the software must not be used for illegal purposes.
* <p>
* THIS SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY. The author is not liable for any claims arising from secondary or illegal development.
* <p>
* Author: Chill Zhuang (bladejava@qq.com)
*/
package org.springblade.transport.support;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springblade.core.log.exception.ServiceException;
import org.springblade.core.redis.lock.LockType;
import org.springblade.core.redis.lock.RedisLockClient;
import org.springblade.core.secure.utils.AuthUtil;
import org.springblade.core.tool.utils.Func;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
/**
* 金蝶定时同步 single-flight 分布式锁支撑组件(模式同全票池同步 syncKingdeeInvoicePool):
* 多副本部署的定时任务与手动触发端点共用一道闸,同一轮同步全局最多一份在跑。
* <p>零等待抢锁:已有轮次进行中则立刻放弃本轮并返回跳过摘要(定时任务下轮幂等重试,手动触发稍后再试);
* 租约须大于最坏一轮耗时(含翻页远程查询),否则租约到期后另一副本会并发起跑;进程异常退出时锁在租约到期后自动释放;
* 未启用 blade.lock.enabled 时降级为直接执行(多副本各自起跑,联调环境可接受,生产须开启)。
*
* @author Chill
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class KingdeeSyncLockSupport {
/** 分布式锁客户端,仅在 blade.lock.enabled=true 时存在,缺失时同步降级为无 single-flight 保护 */
private final ObjectProvider<RedisLockClient> redisLockClientProvider;
/**
* 在 single-flight 分布式锁保护下执行一轮同步
*
* @param lockPrefix 锁键前缀(业务侧隔离,组件内自动追加租户标识)
* @param leaseSeconds 锁租约秒数,须大于最坏一轮耗时
* @param taskName 同步任务名,用于日志与跳过摘要
* @param action 同步主体
* @return 同步摘要;已有轮次进行中时返回跳过摘要
*/
public String runSingleFlight(String lockPrefix, long leaseSeconds, String taskName, Supplier<String> action) {
RedisLockClient redisLockClient = redisLockClientProvider.getIfAvailable();
if (redisLockClient == null) {
// 未启用 blade.lock.enabled:多副本部署会各自起跑同一轮同步,联调环境可接受,生产须开启
log.warn("未启用分布式锁(blade.lock.enabled=false),{}无 single-flight 保护", taskName);
return action.get();
}
String lockKey = lockPrefix + currentTenant();
boolean locked = false;
try {
// 零等待抢锁:已有轮次在跑就立刻放弃本轮,多副本与手动触发共用这一道闸
locked = redisLockClient.tryLock(lockKey, LockType.REENTRANT, 0, leaseSeconds, TimeUnit.SECONDS);
if (!locked) {
log.warn("{}已有轮次进行中,本轮跳过,锁键={}", taskName, lockKey);
return taskName + "进行中,本轮跳过";
}
return action.get();
} catch (InterruptedException interruptedException) {
Thread.currentThread().interrupt();
throw new ServiceException("等待" + taskName + "分布式锁被中断");
} finally {
if (locked) {
redisLockClient.unLock(lockKey, LockType.REENTRANT);
}
}
}
/** 租户标识:定时任务上下文无登录用户时以 default 加锁,锁粒度与数据隔离边界一致 */
private String currentTenant() {
String tenantId = AuthUtil.getTenantId();
return Func.isEmpty(tenantId) ? "default" : tenantId;
}
}