WebFlux API未知记录数时批量调用异常：调用次数超预期

问题现象 #

尝试通过分页批量调用API：每次请求offset递增batchSize（10000），当返回结果数量小于batchSize时停止调用。但在UAT环境中，26000条记录本该调用3次，实际却触发了约33次调用，本地环境则表现正常。

问题根源 #

当前实现依赖外部AtomicBoolean makeNextCall控制takeWhile的终止条件，存在两个关键缺陷：

• Reactor背压预取行为：Flux.fromStream(Stream.iterate(...))是同步无限流，Reactor会基于背压策略预取多个元素。在UAT环境API响应较慢时，takeWhile会在makeNextCall被更新为false之前，提前通过多次判断，触发大量不必要的API调用。
• 异步状态竞争：makeNextCall的更新是在API调用完成后的flatMap中执行，而takeWhile的判断是在流的上游，两者执行时序无法保证同步，导致状态判断滞后。

修复方案 #

使用Reactor的expand操作符替代takeWhile+外部变量的方案。expand会基于前一次请求的结果决定是否生成下一次请求，完全适配异步场景，不会预取多余请求。

修改后的代码示例（严谨版） #

为避免offset计算错误，用Tuple2传递当前offset和结果列表：

// 初始请求：包装offset和结果
Mono<Tuple2<Integer, List<BBTransaction>>> initialRequest = bbTransactionRepository.accountTransactions(transactionContext, 0, batchSize)
        .map(list -> Tuples.of(0, list))
        .doOnNext(tuple -> {
            int offset = tuple.getT1();
            List<BBTransaction> list = tuple.getT2();
            LOGGER.withTask(GET_TRANSACTIONS)
                    .withAttribute(RESULT, list.size())
                    .withAttribute(HIST_OFFSET, offset)
                    .withAttribute(HIST_LIMIT, batchSize)
                    .withAttribute(BATCH, offset / batchSize)
                    .withMessage("fetching bb transactions in batches")
                    .info();
        });

initialRequest.expand(tuple -> {
    int currentOffset = tuple.getT1();
    List<BBTransaction> currentList = tuple.getT2();
    // 结果数量小于batchSize时终止迭代
    if (currentList.size() < batchSize) {
        return Mono.empty();
    }
    int nextOffset = currentOffset + batchSize;
    return bbTransactionRepository.accountTransactions(transactionContext, nextOffset, batchSize)
            .map(nextList -> Tuples.of(nextOffset, nextList))
            .doOnNext(nextTuple -> {
                int batch = nextTuple.getT1() / batchSize;
                LOGGER.withTask(GET_TRANSACTIONS)
                        .withAttribute(RESULT, nextTuple.getT2().size())
                        .withAttribute(HIST_OFFSET, nextTuple.getT1())
                        .withAttribute(HIST_LIMIT, batchSize)
                        .withAttribute(BATCH, batch)
                        .withMessage("fetching bb transactions in batches")
                        .info();
            });
})
.map(Tuple2::getT2) // 提取结果列表
.flatMap(Flux::fromIterable)
.collectList();

关键改进点 #

• 无外部状态依赖：所有控制逻辑基于流的内部数据，避免异步状态竞争。
• 按需生成请求：expand仅在当前请求完成并返回结果后，才决定是否发起下一次请求，完全符合分页逻辑的时序要求。
• 避免无效调用：不会因背压预取导致大量无效API调用，在慢响应环境下表现稳定。

内容的提问来源于stack exchange，提问作者Punit Singh