feat: 定时推送按认领人串行处理——查完该认领人公告并推完名下流水再轮到下一个

This commit is contained in:
刘泉佳
2026-09-29 11:52:42 +08:00
parent fe7a940201
commit 331ea0466f
2 changed files with 57 additions and 36 deletions
@@ -76,6 +76,7 @@ import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -690,10 +691,11 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl<KingdeeReceiptFlowMa
}
/**
* 定时推送金蝶:扫描认领完成且未推送过的流水,按认领人姓名分组重查 A301-2 最新公告后逐条推送
* 定时推送金蝶:扫描认领完成且未推送过的流水,按认领人姓名串行处理——
* 先重查该认领人的 A301-2 最新公告并推完其名下全部流水,再到下一个认领人。
* <p>
* A301-2 的 claimusers 必填且为单个用户姓名,定时任务无登录人上下文,
* 故取候选流水的有效认领人姓名去重后逐人查询公告,按公告id(bill_id)匹配流水。
* 故按认领人逐人查询公告,按公告id(bill_id)匹配流水。
* 单条流水推送失败独立消化,不影响同批其他流水。
*/
@Override
@@ -708,19 +710,37 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl<KingdeeReceiptFlowMa
if (candidates.isEmpty()) {
return "无待推送流水";
}
List<Long> flowIds = candidates.stream().map(KingdeeReceiptFlow::getId).toList();
Set<String> claimerNames = claimMapper.selectList(Wrappers.<ReceiptClaim>lambdaQuery()
.in(ReceiptClaim::getReceiptFlowId, flowIds)
.eq(ReceiptClaim::getClaimStatus, CLAIMED)).stream()
.map(ReceiptClaim::getClaimerName)
.filter(Func::isNotEmpty)
.collect(Collectors.toSet());
if (claimerNames.isEmpty()) {
Map<Long, KingdeeReceiptFlow> candidateById = candidates.stream()
.collect(Collectors.toMap(KingdeeReceiptFlow::getId, Function.identity(),
(first, second) -> first, LinkedHashMap::new));
// 认领人 → 其有效认领的流水id
Map<String, Set<Long>> claimedFlowIdsByClaimer = claimMapper.selectList(
Wrappers.<ReceiptClaim>lambdaQuery()
.in(ReceiptClaim::getReceiptFlowId, candidateById.keySet())
.eq(ReceiptClaim::getClaimStatus, CLAIMED)).stream()
.filter(claim -> Func.isNotEmpty(claim.getClaimerName()) && claim.getReceiptFlowId() != null)
.collect(Collectors.groupingBy(ReceiptClaim::getClaimerName,
Collectors.mapping(ReceiptClaim::getReceiptFlowId, Collectors.toSet())));
if (claimedFlowIdsByClaimer.isEmpty()) {
return "待推送流水 " + candidates.size() + " 条,但无有效认领人,跳过";
}
// 公告按公告id归集(多个认领人查询到同一公告时取首条)
Map<Long, BfiClaimAnnounceVO> announceByBillId = new HashMap<>();
for (String claimerName : claimerNames) {
int successCount = 0;
int failedCount = 0;
int skippedCount = 0;
// 本轮已推送成功的流水:同一流水被多个认领人认领时,只在首个覆盖它的认领人处推送一次
Set<Long> pushedFlowIds = new HashSet<>();
// 按认领人逐人串行:查该认领人公告 → 推完其名下流水 → 下一个认领人
for (String claimerName : claimedFlowIdsByClaimer.keySet().stream().sorted().toList()) {
Set<Long> claimerFlowIds = claimedFlowIdsByClaimer.get(claimerName);
List<KingdeeReceiptFlow> claimerFlows = candidateById.values().stream()
.filter(flow -> claimerFlowIds.contains(flow.getId()))
.filter(flow -> !pushedFlowIds.contains(flow.getId()))
.toList();
if (claimerFlows.isEmpty()) {
continue;
}
// 该认领人的公告一次查询,按公告id归集
Map<Long, BfiClaimAnnounceVO> announceByBillId = new HashMap<>();
try {
for (BfiClaimAnnounceVO announce : queryClaimAnnounces(claimerName)) {
if (announce.getId() != null) {
@@ -728,29 +748,30 @@ public class ReceiptFlowServiceImpl extends BaseServiceImpl<KingdeeReceiptFlowMa
}
}
} catch (Exception exception) {
log.error("定时推送收款认领:认领人[{}]公告查询失败", claimerName, exception);
}
}
int successCount = 0;
int failedCount = 0;
int skippedCount = 0;
for (KingdeeReceiptFlow flow : candidates) {
BfiClaimAnnounceVO announce = announceByBillId.get(flow.getBillId());
if (announce == null) {
skippedCount++;
log.info("定时推送收款认领:流水[{}]未在认领人公告查询结果中命中,跳过", flow.getId());
log.error("定时推送收款认领:认领人[{}]公告查询失败,其名下流水全部跳过", claimerName, exception);
skippedCount += claimerFlows.size();
continue;
}
try {
ReceiptClaimPushResult result = pushClaimLocked(flow.getId(), announce);
if (result.isSuccess()) {
successCount++;
} else {
failedCount++;
for (KingdeeReceiptFlow flow : claimerFlows) {
BfiClaimAnnounceVO announce = announceByBillId.get(flow.getBillId());
if (announce == null) {
skippedCount++;
log.info("定时推送收款认领:认领人[{}]流水[{}]未在公告查询结果中命中,跳过",
claimerName, flow.getId());
continue;
}
try {
ReceiptClaimPushResult result = pushClaimLocked(flow.getId(), announce);
if (result.isSuccess()) {
successCount++;
pushedFlowIds.add(flow.getId());
} else {
failedCount++;
}
} catch (Exception exception) {
failedCount++;
log.error("定时推送收款认领:认领人[{}]流水[{}]推送异常", claimerName, flow.getId(), exception);
}
} catch (Exception exception) {
failedCount++;
log.error("定时推送收款认领:流水[{}]推送异常", flow.getId(), exception);
}
}
String summary = "扫描 " + candidates.size() + " 条,推送成功 " + successCount
@@ -145,9 +145,9 @@ claimed 累计有效认领 == unclaimed_amount(且 > 0) ← 「认领
**A301-2 重查的认领人问题**:A301-2 的 `claimusers` 必填且为单个用户姓名(`BfiClaimAnnounceQueryDTO` L50-52),定时任务没有登录用户上下文。处理:
1. 取候选流水的有效认领主单 `claimer_name` 去重集合;
2. 逐人调 `BfiUtil.queryAllPages(queryClaimAnnounce)`(时间窗沿用 `thirdParty.bfi.claimAnnounce.months`,默认 6 个月);
3. 按公告 `id == 流水.bill_id` 匹配刷新,再走 §4.2 ③-⑦;
4. 单流水推送失败 try/catch 独立消化,不影响同批其他流水。
2. **按认领人串行处理**:先调 `BfiUtil.queryAllPages(queryClaimAnnounce)` 查该认领人的公告(时间窗沿用 `thirdParty.bfi.claimAnnounce.months`,默认 6 个月),推完其名下全部流水后再处理下一个认领人;
3. 该认领人名下流水按公告 `id == 流水.bill_id` 匹配刷新,再走 §4.2 ③-⑦;
4. 单流水推送失败 try/catch 独立消化,不影响同批其他流水;同一流水被多个认领人认领时只在首个覆盖它的认领人处推送一次。
> 联调确认项:`claimusers` 若支持逗号分隔多人,可合并为一次查询(见 §8)。