feat: BfiUtil新增时间窗口二分切片续扫通用翻页方法

2026-10-29 17:26:10 +08:00
parent d69c112eff
commit 99b48be34f
1 changed files with 88 additions and 0 deletions
@@ -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 时间窗口二分切片续扫翻页，逐页回调处理，窗口数据超过单查询翻页上限时尾部不漏同步
+	 * <p>
+	 * 适用按时间窗口增量同步且数据量可能超过单查询翻页上限的场景
+	 * （A801-1 票池同步按 updateTime、A301-1 收款单同步按业务日期）：
+	 * <pre>{@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));
+	 * }</pre>
+	 * <p>
+	 * 单片独立翻页（最多 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 <T>         数据行类型
+	 * @return 实际处理完的切片数（含空片，异常跳过的不计）
+	 */
+	public static <T> int forEachWindowSlice(String bizName, LocalDateTime windowStart, LocalDateTime windowEnd,
+		int pageSize, int maxPages, WindowPageFetcher<T> pageFetcher, Consumer<List<T>> pageHandler) {
+		// 二分收敛保护：切片区间为空说明时间粒度异常，放弃该片
+		if (!windowStart.isBefore(windowEnd)) {
+			log.error("{}切片区间为空（{} ~ {}），时间粒度异常，放弃该切片", bizName, windowStart, windowEnd);
+			return 0;
+		}
+		for (int pageNo = 1; pageNo <= maxPages; pageNo++) {
+			BfiPageVO<T> page = pageFetcher.fetch(windowStart, windowEnd, pageNo);
+			List<T> 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 <T> 数据行类型
+	 */
+	@FunctionalInterface
+	public interface WindowPageFetcher<T> {
+
+		/**
+		 * 拉取单个时间片的一页数据
+		 *
+		 * @param sliceStart 切片起点（含）
+		 * @param sliceEnd   切片终点（含）
+		 * @param pageNo     页码，从 1 开始
+		 * @return 分页结果
+		 */
+		BfiPageVO<T> fetch(LocalDateTime sliceStart, LocalDateTime sliceEnd, int pageNo);
+	}
 }