diff --git a/blade-third-party-api/blade-bfi-api/src/main/java/org/springblade/thirdparty/bfi/util/BfiUtil.java b/blade-third-party-api/blade-bfi-api/src/main/java/org/springblade/thirdparty/bfi/util/BfiUtil.java index 6eb35686..685da253 100644 --- a/blade-third-party-api/blade-bfi-api/src/main/java/org/springblade/thirdparty/bfi/util/BfiUtil.java +++ b/blade-third-party-api/blade-bfi-api/src/main/java/org/springblade/thirdparty/bfi/util/BfiUtil.java @@ -31,10 +31,13 @@ import org.springblade.core.tool.utils.Func; import org.springblade.thirdparty.bfi.pojo.dto.BfiBaseDTO; import org.springblade.thirdparty.bfi.pojo.vo.BfiPageVO; +import java.time.Duration; +import java.time.LocalDateTime; import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.UUID; +import java.util.function.Consumer; import java.util.function.IntFunction; /** @@ -220,4 +223,89 @@ public class BfiUtil { } return totalPages; } + + /** + * BFI 时间窗口二分切片续扫翻页,逐页回调处理,窗口数据超过单查询翻页上限时尾部不漏同步 + *

+ * 适用按时间窗口增量同步且数据量可能超过单查询翻页上限的场景 + * (A801-1 票池同步按 updateTime、A301-1 收款单同步按业务日期): + *

{@code
+	 * int sliceCount = BfiUtil.forEachWindowSlice("金蝶全票池同步", windowStart, windowEnd,
+	 * 	PAGE_SIZE, MAX_PAGES,
+	 * 	(sliceStart, sliceEnd, pageNo) -> {
+	 * 		data.setUpdateTime(sliceStart.format(FORMATTER));
+	 * 		data.setUpdateTimeEnd(sliceEnd.format(FORMATTER));
+	 * 		data.setPageNo(pageNo);
+	 * 		data.setPageSize(PAGE_SIZE);
+	 * 		return bfiService.queryInvoicePool(param);
+	 * 	},
+	 * 	rows -> batchUpsert(rows));
+	 * }
+ *

+ * 单片独立翻页(最多 maxPages 页):空页或末页不满一页即该片到底;触达页数保护说明该时间片 + * 数据超过单查询窗口上限(maxPages × pageSize),从中点二分为 [start, mid] + [mid, end] 两片递归续扫, + * 总覆盖不受单查询上限约束。切片边界同刻度的数据可能在相邻两片各拉一次,pageHandler 必须幂等; + * 切片细到不足 1 秒仍满页视为远端异常,告警跳过防御死递归。 + * 终止只认空页与短页,不依赖 lastPage(末页不满一页天然触发短页终止,lastPage 异常提前终止反而漏拉) + * + * @param bizName 业务名称(日志前缀) + * @param windowStart 窗口起点(含) + * @param windowEnd 窗口终点(含) + * @param pageSize 分页大小(须与 pageFetcher 内实际设置的页大小一致) + * @param maxPages 单片最大页数保护 + * @param pageFetcher 时间片起止 + 页码 → 分页结果,lambda 内负责设置时间窗口与页码并调用接口 + * @param pageHandler 单页数据行处理(过滤、upsert、统计等),必须幂等 + * @param 数据行类型 + * @return 实际处理完的切片数(含空片,异常跳过的不计) + */ + public static int forEachWindowSlice(String bizName, LocalDateTime windowStart, LocalDateTime windowEnd, + int pageSize, int maxPages, WindowPageFetcher pageFetcher, Consumer> pageHandler) { + // 二分收敛保护:切片区间为空说明时间粒度异常,放弃该片 + if (!windowStart.isBefore(windowEnd)) { + log.error("{}切片区间为空({} ~ {}),时间粒度异常,放弃该切片", bizName, windowStart, windowEnd); + return 0; + } + for (int pageNo = 1; pageNo <= maxPages; pageNo++) { + BfiPageVO page = pageFetcher.fetch(windowStart, windowEnd, pageNo); + List rows = page == null || page.getRows() == null ? List.of() : page.getRows(); + if (rows.isEmpty()) { + return 1; + } + pageHandler.accept(rows); + // 短页:该片到底,正常终止(不依赖 lastPage,末页不满一页天然触发) + if (rows.size() < pageSize) { + return 1; + } + } + // 触顶:该时间片数据超过单查询窗口上限,从中点二分续扫(边界同刻度数据由 pageHandler 幂等兜底) + LocalDateTime mid = windowStart.plus(Duration.between(windowStart, windowEnd).dividedBy(2)); + if (!mid.isAfter(windowStart)) { + log.error("{}切片 {} ~ {} 已达最小粒度仍满页(>{} 条),该时间片数据异常,跳过", + bizName, windowStart, windowEnd, (long) maxPages * pageSize); + return 0; + } + log.warn("{}切片 {} ~ {} 触达单片翻页保护 {} 页(>{} 条),二分续扫 [{} ~ {}] + [{} ~ {}]", + bizName, windowStart, windowEnd, maxPages, (long) maxPages * pageSize, windowStart, mid, mid, windowEnd); + return forEachWindowSlice(bizName, windowStart, mid, pageSize, maxPages, pageFetcher, pageHandler) + + forEachWindowSlice(bizName, mid, windowEnd, pageSize, maxPages, pageFetcher, pageHandler); + } + + /** + * 时间窗口切片翻页取数函数:按切片起止时间 + 页码拉取一页数据 + * + * @param 数据行类型 + */ + @FunctionalInterface + public interface WindowPageFetcher { + + /** + * 拉取单个时间片的一页数据 + * + * @param sliceStart 切片起点(含) + * @param sliceEnd 切片终点(含) + * @param pageNo 页码,从 1 开始 + * @return 分页结果 + */ + BfiPageVO fetch(LocalDateTime sliceStart, LocalDateTime sliceEnd, int pageNo); + } }