From 7e76964a76afb10ea259e9ff6e0b4a21ffff225f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=88=98=E6=B3=89=E4=BD=B3?= <1107186916@qq.com> Date: Mon, 28 Sep 2026 00:52:12 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=87=91=E8=9D=B6=E5=AE=9A=E6=97=B6?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=E5=8A=A0single-flight=E5=88=86=E5=B8=83?= =?UTF-8?q?=E5=BC=8F=E9=94=81=EF=BC=8C=E5=A4=9A=E5=89=AF=E6=9C=AC=E4=B8=8E?= =?UTF-8?q?=E6=89=8B=E5=8A=A8=E8=A7=A6=E5=8F=91=E5=85=B1=E7=94=A8=E4=B8=80?= =?UTF-8?q?=E9=81=93=E9=97=B8=EF=BC=88=E4=BB=98=E6=AC=BE=E7=BB=93=E6=9E=9C?= =?UTF-8?q?=E5=9B=9E=E5=86=99/=E5=BA=94=E6=94=B6=E5=8B=BE=E7=A8=BD/?= =?UTF-8?q?=E5=BC=80=E7=A5=A8=E7=BB=93=E6=9E=9C/=E6=94=B6=E7=A5=A8?= =?UTF-8?q?=E7=BB=93=E6=9E=9C/=E5=AE=A2=E5=95=86=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 KingdeeSyncLockSupport：零等待tryLock+租约可配+租户隔离，未启用blade.lock.enabled时降级并告警 - 各同步方法拆为锁壳+executeXxxSync主体，锁在service层保证手动触发端点与定时任务行为一致 - 租约配置：thirdParty.bfi.{payResult|arResult|invoiceResult|receivedInvoice|customer}.lockLeaseSeconds，默认1800秒 --- .../impl/CustomerArchiveServiceImpl.java | 24 +++++ .../impl/FormalSettlementServiceImpl.java | 24 +++++ .../impl/InvoiceApplicationServiceImpl.java | 23 +++++ .../impl/InvoiceReceiptServiceImpl.java | 23 +++++ .../impl/PaymentApplicationServiceImpl.java | 23 +++++ .../support/KingdeeSyncLockSupport.java | 93 +++++++++++++++++++ 6 files changed, 210 insertions(+) create mode 100644 blade-service/blade-transport/src/main/java/org/springblade/transport/support/KingdeeSyncLockSupport.java diff --git a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/CustomerArchiveServiceImpl.java b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/CustomerArchiveServiceImpl.java index ae13b4b8..b4fa5c35 100644 --- a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/CustomerArchiveServiceImpl.java +++ b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/CustomerArchiveServiceImpl.java @@ -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 selectCustomerArchivePage(IPage page, CustomerArchiveVO customer) { @@ -622,6 +635,17 @@ public class CustomerArchiveServiceImpl extends BaseServiceImpl approvedList = list(Wrappers.lambdaQuery() diff --git a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/FormalSettlementServiceImpl.java b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/FormalSettlementServiceImpl.java index 9c8d0f86..f942ae31 100644 --- a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/FormalSettlementServiceImpl.java +++ b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/FormalSettlementServiceImpl.java @@ -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 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 selectPage(IPage page, FormalSettlementVO query) { @@ -523,6 +536,17 @@ public class FormalSettlementServiceImpl extends BaseServiceImpl settlements = list(Wrappers.lambdaQuery() diff --git a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/InvoiceApplicationServiceImpl.java b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/InvoiceApplicationServiceImpl.java index e27d0c3c..7c2f82e3 100644 --- a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/InvoiceApplicationServiceImpl.java +++ b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/InvoiceApplicationServiceImpl.java @@ -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 selectPage(IPage page, InvoiceApplicationVO query) { @@ -780,6 +792,17 @@ public class InvoiceApplicationServiceImpl extends BaseServiceImpl applications = list(Wrappers.lambdaQuery() .eq(InvoiceApplication::getApprovalStatus, TransportBusiConstant.APPROVED) diff --git a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/InvoiceReceiptServiceImpl.java b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/InvoiceReceiptServiceImpl.java index a6262440..97bcb710 100644 --- a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/InvoiceReceiptServiceImpl.java +++ b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/InvoiceReceiptServiceImpl.java @@ -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 selectPage(IPage page, InvoiceReceiptVO query) { LambdaQueryWrapper wrapper = Wrappers.lambdaQuery() @@ -869,6 +881,17 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl pendingReceipts = list(Wrappers.lambdaQuery() .in(InvoiceReceipt::getKingdeePushStatus, KingdeeBfiConstant.RECEIVED_PUSH_PUSHED, KingdeeBfiConstant.RECEIVED_PUSH_ASSIGNED) diff --git a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/PaymentApplicationServiceImpl.java b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/PaymentApplicationServiceImpl.java index 38ea696e..b9a8fe04 100644 --- a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/PaymentApplicationServiceImpl.java +++ b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/PaymentApplicationServiceImpl.java @@ -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 selectPage(IPage 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() { // ① 待回写付款申请：审批通过且金蝶状态为 已生成/付款中/生成失败（终态已付款/已关闭不再轮询）。 // 生成失败单据一并纳入勾稽：金蝶侧已生成说明上次推送实际成功（如下推请求超时但远端已落单、本地事务回滚）， // 查到即回填纠正为已生成，避免人工重新推送产生重复单据；未查到则保持生成失败等待异常处理后重新推送 diff --git a/blade-service/blade-transport/src/main/java/org/springblade/transport/support/KingdeeSyncLockSupport.java b/blade-service/blade-transport/src/main/java/org/springblade/transport/support/KingdeeSyncLockSupport.java new file mode 100644 index 00000000..684068a2 --- /dev/null +++ b/blade-service/blade-transport/src/main/java/org/springblade/transport/support/KingdeeSyncLockSupport.java @@ -0,0 +1,93 @@ +/** + * BladeX Commercial License Agreement + * Copyright (c) 2018-2099, https://bladex.cn. All rights reserved. + *

+ * Use of this software is governed by the Commercial License Agreement obtained after purchasing a license from BladeX. + *

+ * 1. This software is for development use only under a valid license from BladeX. + *

+ * 2. Redistribution of this software's source code to any third party without a commercial license is strictly prohibited. + *

+ * 3. Licensees may copyright their own code but cannot use segments from this software for such purposes. + * Copyright of this software remains with BladeX. + *

+ * Using this software signifies agreement to this License, and the software must not be used for illegal purposes. + *

+ * THIS SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY. The author is not liable for any claims arising from secondary or illegal development. + *

+ * 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）： + * 多副本部署的定时任务与手动触发端点共用一道闸，同一轮同步全局最多一份在跑。 + *

零等待抢锁：已有轮次进行中则立刻放弃本轮并返回跳过摘要（定时任务下轮幂等重试，手动触发稍后再试）； + * 租约须大于最坏一轮耗时（含翻页远程查询），否则租约到期后另一副本会并发起跑；进程异常退出时锁在租约到期后自动释放； + * 未启用 blade.lock.enabled 时降级为直接执行（多副本各自起跑，联调环境可接受，生产须开启）。 + * + * @author Chill + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class KingdeeSyncLockSupport { + + /** 分布式锁客户端，仅在 blade.lock.enabled=true 时存在，缺失时同步降级为无 single-flight 保护 */ + private final ObjectProvider redisLockClientProvider; + + /** + * 在 single-flight 分布式锁保护下执行一轮同步 + * + * @param lockPrefix 锁键前缀（业务侧隔离，组件内自动追加租户标识） + * @param leaseSeconds 锁租约秒数，须大于最坏一轮耗时 + * @param taskName 同步任务名，用于日志与跳过摘要 + * @param action 同步主体 + * @return 同步摘要；已有轮次进行中时返回跳过摘要 + */ + public String runSingleFlight(String lockPrefix, long leaseSeconds, String taskName, Supplier 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; + } +}