⚡ 金蝶全票池定时同步按变更比对落库并加 single-flight 锁

- 镜像批量 upsert 增加票面字段指纹比对,票面无变化不重写(含 attachments_json 大字段),口径对齐 MyBatis-Plus 非空字段更新策略
- 收票登记绑定前比对 kingdeeInvoicePoolId/kingdeeBillNo/kingdeeStatus,三项均无变化不更新;同步流水只在绑定关系变化或金蝶状态迁移时写入,原因文案按语义拆分为「定时绑定」与「定时刷新」
- 同步入口按租户加 RedisLockClient 分布式锁做 single-flight(零等待、租约可配 lockLeaseSeconds),未开启 blade.lock.enabled 时降级执行并告警
- 客商税号白名单明确为购、销双方税号均须命中客商档案(我方主体税号亦在档案中),抽出 taxNoArchived 做空值防护
- 日志补强:每页过滤漏斗明细、镜像无变化数、镜像未匹配登记提示、翻页触顶告警、补偿绑定耗时与远程补查次数、解析失败带发票号码
- Nacos 模板:MySQL 连接串补 rewriteBatchedStatements=true 使 JDBC 批量真正合批;dev/test 开启 blade.lock.enabled,prod 补开启前置条件说明
- SyncKingdeeTest:方法名更正为 syncKingdeeInvoicePool,测试属性关闭 syncEnabled,避免 cron 与手动触发在同一 JVM 并发跑
This commit is contained in:
刘泉佳
2026-09-21 17:31:22 +08:00
parent 9a82ed2eaf
commit b3eee6347b
7 changed files with 363 additions and 75 deletions
@@ -70,6 +70,9 @@ public interface IInvoiceReceiptService extends BaseService<InvoiceReceipt> {
/**
* 定时查询金蝶全票池发票(A801-1)并绑定收票登记
* <p>
* 按租户加分布式锁做 single-flight(需开启 {@code blade.lock.enabled}):多副本并发触发、
* 或手动触发撞上进行中的轮次时,抢不到锁的一轮直接跳过并返回提示,不重复拉取与写库。
*
* @return 执行摘要,含拉取页数、镜像新增/更新、绑定与状态刷新统计
*/
@@ -78,8 +81,8 @@ public interface IInvoiceReceiptService extends BaseService<InvoiceReceipt> {
/**
* 批量应用金蝶票池绑定结果(页级事务):登记绑定批量更新 + 同步流水批量写入,整页原子生效
*
* @param receipts 已回填绑定字段的收票登记
* @param records 同步流水
* @param receipts 变更比对后需更新、且已回填绑定字段的收票登记
* @param records 同步流水,绑定关系变化或金蝶状态迁移时才写,可为 receipts 的子集
*/
void applyInvoicePoolSyncBatch(List<InvoiceReceipt> receipts, List<InvoiceReceiptRecord> records);
@@ -33,6 +33,8 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springblade.core.log.exception.ServiceException;
import org.springblade.core.mp.base.BaseServiceImpl;
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.jackson.JsonUtil;
import org.springblade.core.tool.utils.Func;
@@ -65,6 +67,7 @@ import org.springblade.transport.pojo.entity.KingdeeInvoicePool;
import org.springblade.transport.pojo.vo.InvoiceReceiptVO;
import org.springblade.transport.service.IInvoiceReceiptService;
import org.springblade.transport.wrapper.InvoiceReceiptWrapper;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -83,6 +86,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -121,6 +125,30 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
/** A801-1 增量窗口更新时间格式 */
private static final DateTimeFormatter POOL_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
/**
* 镜像票面指纹字段:A801-1 回填且参与变更比对的字段,全部无变化则本轮不重写镜像(附件 JSON 为大字段,禁止无变化重写)
*/
private static final List<Function<KingdeeInvoicePool, Object>> POOL_FINGERPRINT_FIELDS = List.of(
KingdeeInvoicePool::getInvoiceNo,
KingdeeInvoicePool::getSerialNo,
KingdeeInvoicePool::getKingdeeBillNo,
KingdeeInvoicePool::getKingdeeStatus,
KingdeeInvoicePool::getInvoiceDate,
KingdeeInvoicePool::getInvoiceType,
KingdeeInvoicePool::getTaxRate,
KingdeeInvoicePool::getInvoiceAmount,
KingdeeInvoicePool::getTaxAmount,
KingdeeInvoicePool::getReceiverName,
KingdeeInvoicePool::getBuyerTaxNo,
KingdeeInvoicePool::getIssuerName,
KingdeeInvoicePool::getSalerTaxNo,
KingdeeInvoicePool::getCheckStatus,
KingdeeInvoicePool::getAttachmentsJson,
KingdeeInvoicePool::getSourceUpdatedTime);
/** 定时绑定 single-flight 锁键前缀,按租户区分,同租户同一时刻只允许一轮全票池同步 */
private static final String POOL_SYNC_LOCK_PREFIX = "transport:invoice-receipt:kingdee-pool-sync:";
private final KingdeeInvoicePoolMapper invoicePoolMapper;
private final InvoiceReceiptSettlementMapper settlementRelationMapper;
private final InvoiceReceiptRecordMapper recordMapper;
@@ -128,6 +156,8 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
private final CustomerArchiveMapper customerArchiveMapper;
private final CustomerContactMapper customerContactMapper;
private final IBfiService bfiService;
/** 分布式锁客户端,仅在 blade.lock.enabled=true 时存在,缺失时全票池同步降级为无 single-flight 保护 */
private final ObjectProvider<RedisLockClient> redisLockClientProvider;
/** A801-1 增量窗口天数(按票池更新时间),首轮/兜底全量比对窗口 */
@Value("${thirdParty.bfi.invoiceReceipt.incrementDays:30}")
@@ -141,6 +171,13 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
@Value("${thirdParty.bfi.invoiceReceipt.compensateLimit:200}")
private int compensateLimit;
/**
* 全票池同步 single-flight 锁租约秒数,须大于最坏一轮耗时(翻页远程 + 补偿补查),
* 否则租约到期后另一副本会并发起跑;进程异常退出时锁在租约到期后自动释放
*/
@Value("${thirdParty.bfi.invoiceReceipt.lockLeaseSeconds:1800}")
private long syncLockLeaseSeconds;
@Override
public IPage<InvoiceReceiptVO> selectPage(IPage<InvoiceReceipt> page, InvoiceReceiptVO query) {
LambdaQueryWrapper<InvoiceReceipt> wrapper = Wrappers.<InvoiceReceipt>lambdaQuery()
@@ -412,58 +449,137 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
@Override
public String syncKingdeeInvoicePool() {
RedisLockClient redisLockClient = redisLockClientProvider.getIfAvailable();
if (redisLockClient == null) {
// 未启用 blade.lock.enabled:多副本部署会各自起跑同一轮同步,联调环境可接受,生产须开启
log.warn("未启用分布式锁(blade.lock.enabled=false),金蝶全票池同步无 single-flight 保护");
return executeInvoicePoolSync();
}
String lockKey = POOL_SYNC_LOCK_PREFIX + currentTenant();
boolean locked = false;
try {
// 零等待抢锁:已有轮次在跑就立刻放弃本轮,多副本与手动触发共用这一道闸
locked = redisLockClient.tryLock(lockKey, LockType.REENTRANT, 0, syncLockLeaseSeconds, TimeUnit.SECONDS);
if (!locked) {
log.warn("金蝶全票池同步已有轮次进行中,本轮跳过,锁键={}", lockKey);
return "金蝶全票池同步进行中,本轮跳过";
}
return executeInvoicePoolSync();
} catch (InterruptedException interruptedException) {
Thread.currentThread().interrupt();
throw new ServiceException("等待金蝶全票池同步分布式锁被中断");
} finally {
if (locked) {
redisLockClient.unLock(lockKey, LockType.REENTRANT);
}
}
}
/**
* 金蝶全票池同步主体:增量窗口分页拉取 + 镜像批量 upsert + 绑定回填 + 漏绑补偿,
* 由 {@link #syncKingdeeInvoicePool()} 在分布式锁保护下调用
*
* @return 执行摘要
*/
private String executeInvoicePoolSync() {
// ① A801-1 增量窗口分页拉取(票池更新时间窗口,rim 契约不返回总数,按"本页不满页即末页"终止);
// 配置我方购方税号后按税号逐个分页拉取(buyerTaxNo 单值参数),只拉我方收票,分页精确不漏
long startMillis = System.currentTimeMillis();
LocalDateTime windowStart = LocalDateTime.now().minusDays(incrementDays);
LocalDateTime windowEnd = LocalDateTime.now();
IInvoiceReceiptService self = self();
// 以下为本轮统计口径,全部累计整轮并进入结束摘要,用于观测拉取量、过滤量与写库量
// A801-1 原始拉取张数,未经任何白名单过滤
int pulledCount = 0;
// 发票种类白名单过滤掉的张数(增量窗口内非同步范围的票)
int filteredCount = 0;
int salerFilteredCount = 0;
// 客商税号白名单过滤掉的张数(购、销双方税号任一未命中客商档案)
int taxNoFilteredCount = 0;
// 镜像新增条数(票池有、本地无)
int created = 0;
// 镜像重写条数(票面指纹比对后有变化才计,无变化不重写)
int updated = 0;
// 两道白名单过滤后进入镜像 upsert 的张数
int allowedCount = 0;
// 镜像票面无变化条数(既非新增也非重写,用于解释「过滤后有票却零写入」)
int mirrorUnchanged = 0;
// 登记首次绑定/改绑条数(绑定关系发生变化)
int bound = 0;
// 登记金蝶侧信息刷新条数(已绑定,仅流水号或同步状态变化)
int refreshed = 0;
// 页级批量落库失败条数(整页跳过,下轮幂等重试)
int failed = 0;
// 客商税号白名单:客商档案有效税号(购、销双方税号均须命中),集合为空不限制
// 登记无变化跳过条数(绑定关系与金蝶信息均一致,不写库不刷流水)
int unchanged = 0;
// 实际处理的翻页数,触达最大页数保护时配合 warn 定位截断
int pageCount = 0;
// 客商税号白名单:客商档案有效税号(购、销双方税号均须命中,我方主体税号同样维护在客商档案),集合为空不限制
Set<String> customerTaxNos = customerTaxNoSet();
log.info("定时绑定金蝶全票池开始,增量窗口 {} ~ {},发票种类:{},客商税号白名单 {} 个",
log.info("定时绑定金蝶全票池开始,增量窗口 {} ~ {},发票种类:{},客商税号白名单 {} 个,单页 {} 条最多 {} 页",
windowStart.format(POOL_TIME_FORMATTER), windowEnd.format(POOL_TIME_FORMATTER),
Func.isEmpty(invoiceTypes) ? "不限制" : invoiceTypes,
customerTaxNos.isEmpty() ? "无(不限制)" : String.valueOf(customerTaxNos.size()));
for (int pageNo = 1; pageNo <= MAX_POOL_PAGES; pageNo++) {
customerTaxNos.isEmpty() ? "无(不限制)" : String.valueOf(customerTaxNos.size()),
POOL_PAGE_SIZE, MAX_POOL_PAGES);
int pageNo = 1;
for (; pageNo <= MAX_POOL_PAGES; pageNo++) {
List<BfiInvoicePoolVO> rows = queryPoolPage(windowStart, windowEnd, pageNo);
if (rows.isEmpty()) {
break;
}
pageCount++;
pulledCount += rows.size();
// 发票种类白名单 + 客商税号白名单(购、销双方)过滤(翻页终止仍按原始页条数判断)
// 发票种类白名单 + 客商税号白名单(购、销双方税号均须命中)过滤(翻页终止仍按原始页条数判断)
List<BfiInvoicePoolVO> typeAllowedRows = filterByInvoiceTypes(rows);
filteredCount += rows.size() - typeAllowedRows.size();
int pageTypeFiltered = rows.size() - typeAllowedRows.size();
filteredCount += pageTypeFiltered;
List<BfiInvoicePoolVO> allowedRows = filterByArchiveTaxNos(customerTaxNos, typeAllowedRows);
salerFilteredCount += typeAllowedRows.size() - allowedRows.size();
// ② 镜像批量 upsert:票池为唯一事实源,无论是否已登记均刷新镜像(批量查改,避免逐条 N+1)
int pageTaxFiltered = typeAllowedRows.size() - allowedRows.size();
taxNoFilteredCount += pageTaxFiltered;
allowedCount += allowedRows.size();
// 每页过滤漏斗:定位「拉了几千张却一张都不入库」到底卡在发票种类还是客商税号白名单
log.info("定时绑定金蝶全票池 第{}页过滤明细:原始 {} 张,发票种类白名单外 {} 张,客商税号白名单外 {} 张,通过 {} 张",
pageNo, rows.size(), pageTypeFiltered, pageTaxFiltered, allowedRows.size());
// ② 镜像批量 upsert:票池为唯一事实源,无论是否已登记均以票池为准,票面字段有变化才重写(批量查改避免逐条 N+1)
BatchUpsertResult upsertResult = batchUpsertInvoicePool(allowedRows);
Map<String, KingdeeInvoicePool> mirrorMap = upsertResult.mirrors();
created += upsertResult.created();
updated += upsertResult.updated();
// ③ 同页发票号码批量定位收票登记,内存回填绑定后页级批量落库(批量失败整页跳过待下轮)
mirrorUnchanged += upsertResult.unchanged();
// ③ 同页发票号码批量定位收票登记,内存比对变更后页级批量落库(无变化不写库不刷流水;批量失败整页跳过待下轮)
if (!mirrorMap.isEmpty()) {
Map<String, List<InvoiceReceipt>> receiptMap = list(Wrappers.<InvoiceReceipt>lambdaQuery()
.in(InvoiceReceipt::getInvoiceNo, mirrorMap.keySet())
.ne(InvoiceReceipt::getApprovalStatus, VOIDED)).stream()
.filter(receipt -> !Objects.equals(receipt.getIsDeleted(), 1))
.collect(Collectors.groupingBy(InvoiceReceipt::getInvoiceNo));
// 镜像有了却没人登记:本页不产生任何绑定,属正常(票池范围大于我方登记范围)
if (receiptMap.isEmpty()) {
log.info("定时绑定金蝶全票池 第{}页 {} 张镜像未匹配到收票登记(发票号码未登记或登记已作废)",
pageNo, mirrorMap.size());
}
List<BindTarget> targets = new ArrayList<>();
for (Map.Entry<String, KingdeeInvoicePool> entry : mirrorMap.entrySet()) {
for (InvoiceReceipt receipt : receiptMap.getOrDefault(entry.getKey(), List.of())) {
KingdeeInvoicePool mirror = entry.getValue();
// A801-1 未返回流水号时保持登记原值,与镜像非空字段更新口径一致,避免空值每轮判为变更
String kingdeeBillNo = Func.isNotEmpty(mirror.getKingdeeBillNo())
? mirror.getKingdeeBillNo() : receipt.getKingdeeBillNo();
boolean unbound = receipt.getKingdeeInvoicePoolId() == null
|| !Objects.equals(receipt.getKingdeeInvoicePoolId(), entry.getValue().getId());
|| !Objects.equals(receipt.getKingdeeInvoicePoolId(), mirror.getId());
boolean statusChanged = !Objects.equals(receipt.getKingdeeStatus(), mirror.getKingdeeStatus());
boolean billNoChanged = !Objects.equals(receipt.getKingdeeBillNo(), kingdeeBillNo);
// 变更检测:绑定关系、金蝶流水号、同步状态三项均无变化则本轮跳过,避免整窗每轮重写与流水刷屏
if (!unbound && !statusChanged && !billNoChanged) {
unchanged++;
continue;
}
// 内存回填:绑定关系对齐当前发票号码对应镜像(同租户同发票号镜像唯一),票面字段不覆盖人工登记信息
receipt.setKingdeeInvoicePoolId(entry.getValue().getId());
receipt.setKingdeeBillNo(entry.getValue().getKingdeeBillNo());
receipt.setKingdeeStatus(entry.getValue().getKingdeeStatus());
targets.add(new BindTarget(receipt, entry.getValue(), unbound));
receipt.setKingdeeInvoicePoolId(mirror.getId());
receipt.setKingdeeBillNo(kingdeeBillNo);
receipt.setKingdeeStatus(mirror.getKingdeeStatus());
// 流水只在绑定关系变化或金蝶状态迁移时写,纯流水号回填不刷流水
InvoiceReceiptRecord syncRecord = unbound || statusChanged ? buildSyncRecord(receipt,
unbound ? "定时绑定金蝶票池发票" : "定时刷新金蝶票池状态") : null;
targets.add(new BindTarget(receipt, mirror, unbound, syncRecord));
}
}
if (!targets.isEmpty()) {
@@ -471,20 +587,28 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
// 页级事务批量落库:绑定批量更新 + 同步流水批量写入(JDBC batch,整页原子生效)
self.applyInvoicePoolSyncBatch(
targets.stream().map(BindTarget::receipt).toList(),
targets.stream().map(target -> buildSyncRecord(target.receipt(), "定时绑定金蝶票池发票")).toList());
targets.stream().map(BindTarget::syncRecord).filter(Objects::nonNull).toList());
int pageBound = 0;
int pageRecorded = 0;
for (BindTarget target : targets) {
if (target.syncRecord() != null) {
pageRecorded++;
}
if (target.unbound()) {
bound++;
log.info("定时绑定金蝶票池发票成功,发票号码={},登记id={},金蝶流水号={}",
target.receipt().getInvoiceNo(), target.receipt().getId(), target.mirror().getSerialNo());
pageBound++;
log.info("定时绑定金蝶票池发票成功,第{}页,发票号码={},登记id={},金蝶流水号={}",
pageNo, target.receipt().getInvoiceNo(), target.receipt().getId(), target.mirror().getSerialNo());
} else {
refreshed++;
}
}
log.info("定时绑定金蝶票池发票第{}页落库完成,变更 {} 条(绑定 {} 条、状态刷新 {} 条),同步流水 {} 条",
pageNo, targets.size(), pageBound, targets.size() - pageBound, pageRecorded);
} catch (Exception batchException) {
// 批量落库失败:整页跳过计入失败,下轮任务幂等重试,不做逐单降级(保持全批量链路)
failed += targets.size();
log.error("定时绑定金蝶票池发票批量落库异常,本页 {} 条跳过待下轮重试", targets.size(), batchException);
log.error("定时绑定金蝶票池发票批量落库异常,第{}页 {} 条跳过待下轮重试", pageNo, targets.size(), batchException);
}
}
}
@@ -492,12 +616,19 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
break;
}
}
if (pageNo > MAX_POOL_PAGES) {
// 末轮仍满页说明增量窗口数据量已超过翻页保护,尾部数据本轮未处理(下轮窗口未变仍会漏,须调整配置)
log.warn("金蝶全票池增量窗口翻页达最大保护 {} 页({} 条)仍未到底,窗口尾部本轮未处理,请缩小 incrementDays 或调大保护页数",
MAX_POOL_PAGES, MAX_POOL_PAGES * POOL_PAGE_SIZE);
}
// ④ 补偿绑定:仍为未绑定状态的登记按发票号码精确补查(不受增量窗口限制),
// 兜底增量窗口滑动或批量落库持续失败导致的漏绑——发票在金蝶票池可随时按号补查,直到绑定成功为止
int compensated = compensateUnboundReceipts(self);
String summary = "金蝶全票池同步:拉取" + pulledCount + "张发票,类型过滤" + filteredCount + "张,客商过滤"
+ salerFilteredCount + "张,镜像新增" + created + "更新" + updated + ",绑定" + bound + ",状态刷新"
+ refreshed + ",失败" + failed + ",补偿绑定" + compensated;
String summary = "金蝶全票池同步:翻页" + pageCount + "页拉取" + pulledCount + "张发票,类型过滤" + filteredCount
+ "张,客商过滤" + taxNoFilteredCount + "张,过滤后剩" + allowedCount + "张,镜像新增" + created + "更新"
+ updated + "无变化" + mirrorUnchanged + ",绑定" + bound + ",状态刷新"
+ refreshed + ",登记无变化跳过" + unchanged + ",失败" + failed + ",补偿绑定" + compensated + ",耗时"
+ (System.currentTimeMillis() - startMillis) + "ms";
log.info("定时绑定金蝶全票池结束,{}", summary);
return summary;
}
@@ -520,6 +651,7 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
if (unboundReceipts.isEmpty()) {
return 0;
}
long startMillis = System.currentTimeMillis();
log.info("定时绑定金蝶全票池补偿开始,未绑定登记 {} 条(限量 {} 条)", unboundReceipts.size(), compensateLimit);
// ① 本地镜像一次批量查询(主流程本轮刚落库的镜像一并命中,免远程补查)
List<String> invoiceNos = unboundReceipts.stream()
@@ -531,17 +663,23 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
// ② 内存回填绑定;本地镜像缺失的按发票号码精确补查,新增镜像攒批
List<BindTarget> targets = new ArrayList<>();
List<KingdeeInvoicePool> newMirrors = new ArrayList<>();
List<String> poolMissInvoiceNos = new ArrayList<>();
int localHitCount = mirrorMap.size();
int remoteQueryCount = 0;
for (InvoiceReceipt receipt : unboundReceipts) {
if (Func.isEmpty(receipt.getInvoiceNo())) {
log.warn("补偿绑定跳过,登记发票号码为空,登记id={}", receipt.getId());
continue;
}
String invoiceNo = receipt.getInvoiceNo().trim();
KingdeeInvoicePool mirror = mirrorMap.get(invoiceNo);
if (mirror == null) {
// rim 契约无批量号码参数,本地镜像缺失只能逐号远程补查,是本轮最主要的远程调用开销来源
remoteQueryCount++;
List<BfiInvoicePoolVO> poolInvoices = filterByInvoiceTypes(queryPoolByInvoiceNo(invoiceNo));
if (poolInvoices.isEmpty()) {
// 金蝶票池暂无该发票:跳过不写库,待后续轮次补查,避免流水刷屏
log.warn("补偿绑定跳过,金蝶票池未查询到发票,登记id={},发票号码={}", receipt.getId(), invoiceNo);
poolMissInvoiceNos.add(invoiceNo);
continue;
}
mirror = fillInvoicePool(null, poolInvoices.get(0));
@@ -551,9 +689,16 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
receipt.setKingdeeInvoicePoolId(mirror.getId());
receipt.setKingdeeBillNo(mirror.getKingdeeBillNo());
receipt.setKingdeeStatus(mirror.getKingdeeStatus());
targets.add(new BindTarget(receipt, mirror, true));
// 补查对象均为未绑定登记,绑定关系必然变化,流水必写
targets.add(new BindTarget(receipt, mirror, true, buildSyncRecord(receipt, "补偿绑定金蝶票池发票")));
}
if (!poolMissInvoiceNos.isEmpty()) {
log.warn("补偿绑定跳过,金蝶票池未查询到或类型白名单外发票 {} 张,待后续轮次补查,发票号码={}",
poolMissInvoiceNos.size(), poolMissInvoiceNos);
}
if (targets.isEmpty()) {
log.info("定时绑定金蝶全票池补偿结束,无可补偿登记,本地镜像命中 {} 条,远程按号补查 {} 次,票池无票 {} 张,耗时 {}ms",
localHitCount, remoteQueryCount, poolMissInvoiceNos.size(), System.currentTimeMillis() - startMillis);
return 0;
}
// ③ 批量落库:新增镜像批量写入 + 绑定页级事务批量落库(失败整轮跳过待下轮重试,绑定幂等)
@@ -563,9 +708,12 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
}
self.applyInvoicePoolSyncBatch(
targets.stream().map(BindTarget::receipt).toList(),
targets.stream().map(target -> buildSyncRecord(target.receipt(), "补偿绑定金蝶票池发票")).toList());
targets.stream().map(BindTarget::syncRecord).filter(Objects::nonNull).toList());
targets.forEach(target -> log.info("补偿绑定金蝶票池发票成功,发票号码={},登记id={},金蝶流水号={}",
target.receipt().getInvoiceNo(), target.receipt().getId(), target.mirror().getSerialNo()));
log.info("定时绑定金蝶全票池补偿结束,补偿绑定 {} 条,本地镜像命中 {} 条,远程按号补查 {} 次(新增镜像 {} 条),票池无票 {} 张,耗时 {}ms",
targets.size(), localHitCount, remoteQueryCount, newMirrors.size(), poolMissInvoiceNos.size(),
System.currentTimeMillis() - startMillis);
return targets.size();
}
@@ -659,15 +807,16 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
}
/**
* 票池镜像批量 upsert:一次查询同页既有镜像,分组后批量插入/更新,避免逐条查改 N+1
* 票池镜像批量 upsert:一次查询同页既有镜像,票面指纹比对后有变化才批量更新、无变化不重写,
* 新增批量插入,避免逐条查改 N+1,也避免每轮整窗重写大字段
*
* @param rows 白名单过滤后的 A801-1 票池发票
* @return 批量 upsert 结果(invoiceNo -&gt; 镜像,新增/更新条数)
* @return 批量 upsert 结果(invoiceNo -&gt; 镜像,新增条数、实际重写的更新条数)
*/
private BatchUpsertResult batchUpsertInvoicePool(List<BfiInvoicePoolVO> rows) {
Map<String, KingdeeInvoicePool> mirrorMap = new LinkedHashMap<>();
if (rows.isEmpty()) {
return new BatchUpsertResult(mirrorMap, 0, 0);
return new BatchUpsertResult(mirrorMap, 0, 0, 0);
}
List<String> invoiceNos = rows.stream()
.map(row -> row.getInvoiceNo().trim())
@@ -680,14 +829,23 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
.collect(Collectors.toMap(KingdeeInvoicePool::getInvoiceNo, Function.identity(), (first, second) -> first));
List<KingdeeInvoicePool> toInsert = new ArrayList<>();
List<KingdeeInvoicePool> toUpdate = new ArrayList<>();
int unchangedCount = 0;
for (BfiInvoicePoolVO row : rows) {
if (Func.isEmpty(row.getInvoiceNo())) {
continue;
}
KingdeeInvoicePool existing = existingMap.get(row.getInvoiceNo().trim());
// fillInvoicePool 原地改写实体,故先取填充前的票面指纹用于变更比对
String[] fingerprintBefore = invoicePoolFingerprint(existing);
KingdeeInvoicePool mirror = fillInvoicePool(existing, row);
(existing == null ? toInsert : toUpdate).add(mirror);
mirrorMap.put(mirror.getInvoiceNo(), mirror);
if (existing == null) {
toInsert.add(mirror);
} else if (invoicePoolFingerprintChanged(fingerprintBefore, mirror)) {
toUpdate.add(mirror);
} else {
unchangedCount++;
}
}
// 批量写库:失败时中断本轮,由任务层记录异常、下轮重试(upsert 幂等)
if (!toInsert.isEmpty()) {
@@ -696,8 +854,9 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
if (!toUpdate.isEmpty()) {
Db.updateBatchById(toUpdate);
}
log.info("定时绑定金蝶全票池镜像批量写入完成,新增 {} 条,更新 {} 条", toInsert.size(), toUpdate.size());
return new BatchUpsertResult(mirrorMap, toInsert.size(), toUpdate.size());
log.info("定时绑定金蝶全票池镜像批量写入完成,新增 {} 条,更新 {} 条,无变化 {} 条",
toInsert.size(), toUpdate.size(), unchangedCount);
return new BatchUpsertResult(mirrorMap, toInsert.size(), toUpdate.size(), unchangedCount);
}
/**
@@ -718,39 +877,85 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
mirror.setSerialNo(row.getSerialNo());
mirror.setKingdeeBillNo(row.getSerialNo());
mirror.setKingdeeStatus(poolKingdeeStatus(row));
mirror.setInvoiceDate(parseDate(row.getInvoiceDate()));
mirror.setInvoiceDate(parseDate(row.getInvoiceNo(), row.getInvoiceDate()));
mirror.setInvoiceType(BfiConstant.INVOICE_TYPE_NAMES.getOrDefault(row.getInvoiceType(), row.getInvoiceType()));
mirror.setTaxRate(resolveTaxRate(row));
mirror.setInvoiceAmount(parseAmount(row.getTotalAmount()));
mirror.setTaxAmount(parseAmount(row.getTaxAmount()));
mirror.setInvoiceAmount(parseAmount(row.getInvoiceNo(), row.getTotalAmount()));
mirror.setTaxAmount(parseAmount(row.getInvoiceNo(), row.getTaxAmount()));
mirror.setReceiverName(truncate(row.getBuyerName(), 100));
mirror.setBuyerTaxNo(row.getBuyerTaxNo());
mirror.setIssuerName(truncate(row.getSalerName(), 100));
mirror.setSalerTaxNo(row.getSalerTaxNo());
mirror.setCheckStatus(row.getCheckStatus());
mirror.setAttachmentsJson(buildAttachmentJson(row));
mirror.setSourceUpdatedTime(parseDateTime(row.getBillCreateTime()));
mirror.setSourceUpdatedTime(parseDateTime(row.getInvoiceNo(), row.getBillCreateTime()));
return mirror;
}
/**
* 票池镜像批量 upsert 结果
*
* @param mirrors invoiceNo -&gt; upsert 后镜像
* @param created 新增条数
* @param updated 更新条数
* @param mirrors invoiceNo -&gt; upsert 后镜像
* @param created 新增条数
* @param updated 票面有变化并重写的条数
* @param unchanged 票面无变化、本轮不重写的条数
*/
private record BatchUpsertResult(Map<String, KingdeeInvoicePool> mirrors, int created, int updated) {
private record BatchUpsertResult(Map<String, KingdeeInvoicePool> mirrors, int created, int updated,
int unchanged) {
}
/**
* 定时绑定目标:登记实体 + 对应镜像 + 是否首次绑定(区分绑定与状态刷新统计)
*
* @param receipt 已回填绑定字段的收票登记
* @param mirror 对应票据池镜像
* @param unbound true 表示首次绑定,false 表示仅状态刷新
* 镜像票面指纹:A801-1 回填字段的规范化取值快照,用于判断本轮是否需要重写镜像;null 镜像(新增)返回 null
*/
private record BindTarget(InvoiceReceipt receipt, KingdeeInvoicePool mirror, boolean unbound) {
private String[] invoicePoolFingerprint(KingdeeInvoicePool mirror) {
if (mirror == null) {
return null;
}
String[] fingerprint = new String[POOL_FINGERPRINT_FIELDS.size()];
for (int index = 0; index < fingerprint.length; index++) {
fingerprint[index] = fingerprintValue(POOL_FINGERPRINT_FIELDS.get(index).apply(mirror));
}
return fingerprint;
}
/**
* 镜像票面是否变化:填充后为空视为维持库中原值(MyBatis-Plus 非空字段更新策略不写空值),
* 与填充前一致则不算变化,避免源侧空字段导致每轮判为变更而无效重写
*/
private boolean invoicePoolFingerprintChanged(String[] fingerprintBefore, KingdeeInvoicePool mirror) {
if (fingerprintBefore == null) {
return true;
}
for (int index = 0; index < fingerprintBefore.length; index++) {
String fingerprintAfter = fingerprintValue(POOL_FINGERPRINT_FIELDS.get(index).apply(mirror));
if (!fingerprintAfter.isEmpty() && !fingerprintAfter.equals(fingerprintBefore[index])) {
return true;
}
}
return false;
}
/** 指纹取值规范化:空值统一为空串,金额去尾零使 13.0000 与 13 等价,其余按去首尾空白后的文本比对 */
private String fingerprintValue(Object value) {
if (value == null) {
return "";
}
if (value instanceof BigDecimal decimal) {
return decimal.stripTrailingZeros().toPlainString();
}
return value.toString().trim();
}
/**
* 定时绑定目标:登记实体 + 对应镜像 + 是否绑定关系变化 + 同步流水
*
* @param receipt 已回填绑定字段的收票登记
* @param mirror 对应票据池镜像
* @param unbound true 表示绑定关系发生变化(含首次绑定),false 表示仅金蝶侧信息刷新
* @param syncRecord 同步流水,null 表示本轮变更无需记录流水
*/
private record BindTarget(InvoiceReceipt receipt, KingdeeInvoicePool mirror, boolean unbound,
InvoiceReceiptRecord syncRecord) {
}
/**
@@ -801,19 +1006,24 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
}
/**
* 客商税号白名单过滤:购、销双方税号均须命中客商档案税号集合,集合为空不限制;白名单外发票不入镜像不绑定
* 客商税号白名单过滤:购方税号与销方税号**均须**命中客商档案税号集合才放行(我方主体税号同样维护在客商档案中,
* 故进项票需购方=我方、销方=建档客商同时成立),集合为空不限制;白名单外发票不入镜像不绑定
*/
private List<BfiInvoicePoolVO> filterByArchiveTaxNos(Set<String> customerTaxNos, List<BfiInvoicePoolVO> rows) {
if (customerTaxNos.isEmpty()) {
return rows;
}
return rows.stream()
.filter(row -> row.getBuyerTaxNo() != null && row.getSalerTaxNo() != null
&& customerTaxNos.contains(row.getBuyerTaxNo().trim())
&& customerTaxNos.contains(row.getSalerTaxNo().trim()))
.filter(row -> taxNoArchived(customerTaxNos, row.getBuyerTaxNo())
&& taxNoArchived(customerTaxNos, row.getSalerTaxNo()))
.toList();
}
/** 税号是否命中客商档案集合(空税号不命中) */
private boolean taxNoArchived(Set<String> customerTaxNos, String taxNo) {
return Func.isNotEmpty(taxNo) && customerTaxNos.contains(taxNo.trim());
}
/**
* 加载客商档案税号集合(购、销方白名单来源):status=1 有效客商的统一社会信用代码(纳税人识别号)
*/
@@ -848,61 +1058,63 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
* 税率优先取顶层小数换算为百分数(0.13 -> 13),顶层为空回退首条明细,均无效时置 0(镜像列非空约束)
*/
private BigDecimal resolveTaxRate(BfiInvoicePoolVO row) {
BigDecimal taxRate = parsePercent(row.getTaxRate());
String invoiceNo = row.getInvoiceNo();
BigDecimal taxRate = parsePercent(invoiceNo, row.getTaxRate());
if (taxRate == null) {
taxRate = Optional.ofNullable(row.getItems()).orElse(List.of()).stream()
.map(BfiInvoicePoolItemVO::getTaxRate)
.filter(StringUtil::isNotBlank)
.findFirst()
.map(this::parsePercent)
.map(value -> parsePercent(invoiceNo, value))
.orElse(null);
}
return taxRate == null ? BigDecimal.ZERO : taxRate;
}
private BigDecimal parsePercent(String value) {
private BigDecimal parsePercent(String invoiceNo, String value) {
if (StringUtil.isBlank(value)) {
return null;
}
try {
return new BigDecimal(value.trim()).multiply(BigDecimal.valueOf(100));
} catch (NumberFormatException exception) {
log.warn("金蝶票池发票税率解析失败:{}", value);
log.warn("金蝶票池发票税率解析失败,按 0 处理,发票号码={},税率={}", invoiceNo, value);
return null;
}
}
private BigDecimal parseAmount(String value) {
private BigDecimal parseAmount(String invoiceNo, String value) {
if (StringUtil.isBlank(value)) {
return BigDecimal.ZERO;
}
try {
return new BigDecimal(value.trim()).setScale(2, RoundingMode.HALF_UP);
} catch (NumberFormatException exception) {
log.warn("金蝶票池发票金额解析失败:{}", value);
log.warn("金蝶票池发票金额解析失败,按 0 处理,发票号码={},金额={}", invoiceNo, value);
return BigDecimal.ZERO;
}
}
private LocalDate parseDate(String value) {
private LocalDate parseDate(String invoiceNo, String value) {
if (StringUtil.isBlank(value)) {
return null;
}
try {
return LocalDate.parse(value.trim().substring(0, 10));
} catch (Exception exception) {
log.warn("金蝶票池发票开票日期解析失败:{}", value);
log.warn("金蝶票池发票开票日期解析失败,置空处理,发票号码={},日期={}", invoiceNo, value);
return null;
}
}
private LocalDateTime parseDateTime(String value) {
private LocalDateTime parseDateTime(String invoiceNo, String value) {
if (StringUtil.isBlank(value)) {
return null;
}
try {
return LocalDateTime.parse(value.trim().replace(" ", "T").substring(0, 19));
} catch (Exception exception) {
log.warn("金蝶票池发票源侧时间解析失败,置空处理,发票号码={},时间={}", invoiceNo, value);
return null;
}
}
@@ -915,6 +1127,12 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
return trimmed.length() > length ? trimmed.substring(0, length) : trimmed;
}
/** 租户标识:定时任务上下文无登录用户时以 default 加锁,锁粒度与数据隔离边界一致 */
private String currentTenant() {
String tenantId = AuthUtil.getTenantId();
return Func.isEmpty(tenantId) ? "default" : tenantId;
}
/** 操作人:定时任务上下文无登录用户时以「定时任务」名义记录流水 */
private String currentOperator() {
String userName = AuthUtil.getUserName();
@@ -0,0 +1,57 @@
/**
* 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;
import org.junit.jupiter.api.Test;
import org.springblade.transport.service.IInvoiceReceiptService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
/**
* 金蝶数据定时同步测试:手动触发收票管理金蝶全票池发票绑定(A801-1),打印执行摘要便于联调抽查。
* <p>
* 测试用属性显式关闭 {@code thirdParty.bfi.invoiceReceipt.syncEnabled}:{@code @SpringBootTest} 会连同
* {@code @EnableScheduling} 一起启动上下文,不关闭则 cron 会在同一 JVM 内并发再跑一轮,
* 与手动这一轮互相抢数据(未开启 blade.lock.enabled 时没有 single-flight 保护),执行摘要会对不上。
*
* @author Chill
*/
@SpringBootTest(classes = TransportApplication.class, properties = {
"spring.profiles.active=dev",
"spring.main.allow-bean-definition-overriding=true",
"thirdParty.bfi.invoiceReceipt.syncEnabled=false"
})
class SyncKingdeeTest {
@Autowired
private IInvoiceReceiptService invoiceReceiptService;
@Test
void syncKingdeeInvoicePool() {
String summary = invoiceReceiptService.syncKingdeeInvoicePool();
System.out.println("金蝶数据定时同步执行结果:" + summary);
}
}