如何基于返回Mono<Result>的API实现增量式结果获取？

问题背景 #

现有返回Mono<Result>的异步API，服务器会因数据量过大返回分页结果，需从Result中获取token并设置到下一次请求，循环请求直到Result.token为null，以此获取全部数据。同步实现的逻辑已给出，需转换为Reactor异步实现。

核心实现方案 #

利用Reactor的expand操作符实现递归分页请求，该操作符可以将每一次异步调用的结果作为下一次调用的输入，完美适配这种依赖前序结果的分页场景。

基础实现（逐页处理数据）

// 初始化初始请求
Request initialRequest = new Request();

// 构建异步分页请求序列
Mono<Void> fetchAndProcessAll = Mono.just(initialRequest)
        // 发起第一次异步请求
        .flatMap(request -> getAsync(request))
        // 递归展开分页请求：每拿到一页结果，判断是否有下一页token
        .expand(result -> {
            if (result.getToken() != null) {
                Request nextRequest = new Request();
                nextRequest.setToken(result.getToken());
                return getAsync(nextRequest);
            }
            // 无token时返回空Mono终止递归
            return Mono.empty();
        })
        // 处理每一页的结果数据
        .doOnNext(result -> {
            for (String value : result.getData()) {
                // 这里替换为你的单个数据处理逻辑
                // processSingleValue(value);
            }
        })
        // 等待所有请求和处理完成，返回完成信号
        .then();

// 订阅触发整个异步流程
fetchAndProcessAll.subscribe();

进阶实现（收集全部数据后统一处理）

如果需要将所有分页数据收集到一个列表后再批量处理，可以调整如下：

Request initialRequest = new Request();

Mono<List<String>> collectAllData = Mono.just(initialRequest)
        .flatMap(request -> getAsync(request))
        .expand(result -> {
            if (result.getToken() != null) {
                Request nextRequest = new Request();
                nextRequest.setToken(result.getToken());
                return getAsync(nextRequest);
            }
            return Mono.empty();
        })
        .map(Result::getData) // 提取每一页的数据列表
        .flatMapIterable(data -> data) // 将列表拆分为单个元素的序列
        .collectList(); // 收集所有元素到一个List

// 订阅并处理收集后的全部数据
collectAllData.subscribe(allData -> {
    // 批量处理全部数据的逻辑
    // batchProcess(allData);
});

关键说明 #

• expand操作符：是实现递归分页的核心，它会持续订阅由前一次结果生成的新Mono，直到返回Mono.empty()终止序列。
• 错误处理：如果需要处理请求异常，可以在序列中添加onErrorResume等操作符，示例：

.expand(result -> { /* 原有逻辑 */ })
.onErrorResume(exception -> {
    // 异常处理逻辑，比如日志记录
    System.err.println("分页请求失败: " + exception.getMessage());
    return Mono.empty(); // 终止递归
})

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