Spring WebFlux Web客户端-迭代分页REST API

Question

我需要从可分页REST API的所有页面获取项目。我还需要开始处理项目，只要它们可用，不需要等待所有页面加载。为了做到这一点，我使用了Spring WebFlux及其WebClient，并希望返回Flux

• 当前窗口的大小
• 当前窗口中的剩余时间
• 在窗口中请求配额
• 请求留在当前窗口中

对单个页面请求的响应如下所示：

{
    "data": [],
    "meta": {
      "pagination": {
        "total": 10,
        "current": 1
      }
    }
}

数据数组包含实际项目，而元对象包含分页信息。

我当前的解决方案首先执行一个“虚拟”请求，只是为了获取总页数和速率限制。

Mono<T> paginated = client.get()
    .uri(uri)
    .exchange()
    .flatMap(response -> {                  
        HttpHeaders headers = response.headers().asHttpHeaders();

        Limits limits = new Limits();
        limits.setWindowSize(headers.getFirst("X-Window-Size"));
        limits.setWindowRemaining(headers.getFirst("X-Window-Remaining"));
        limits.setRequestsQuota(headers.getFirst("X-Requests-Quota");
        limits.setRequestsLeft(headers.getFirst("X-Requests-Remaining");

        return response.bodyToMono(Paginated.class)
                .map(paginated -> { 
                    paginated.setLimits(limits);
                    return paginated;
                });
    });

之后，我发出一个包含页码的流量，对于每个页面，我执行一个REST API请求，每个请求都被延迟了足够长的时间，因此不会超过限制，并返回一个提取项目的流量：

return paginated.flatMapMany(paginated -> {
    return Flux.range(1, paginated.getMeta().getPagination().getTotal())
            .delayElements(Duration.ofMillis(paginated.getLimits().getWindowRemaining() / paginated.getLimits().getRequestsQuota()))
            .flatMap(page -> {
                return client.get()
                        .uri(pageUri)
                        .retrieve()
                        .bodyToMono(Item.class)
                        .flatMapMany(p -> Flux.fromIterable(p.getData()));
            });
});

这确实有效，但我对此不满意，因为：

• 它执行初始“虚拟”请求以获取页数，然后重复相同的请求以获取实际数据
• 它仅在初始请求时获取速率限制，并假设限制不会更改（例如，它是唯一使用API的限制）-这可能不是真的，在这种情况下，它将得到一个超出限制的错误

因此，我的问题是如何重构它，使其不需要初始请求（而是从第一个请求中获取限制、页码和数据，并继续遍历所有页面，同时更新（并尊重）限制）。

Answer 1

我认为这段代码可以满足您的要求。其想法是创建一个流量来调用资源服务器，但在处理响应的过程中，在该流量上添加一个新事件，以便能够调用下一页。

代码由以下部分组成：

一个简单的包装器，包含要调用的下一页和在执行调用之前要等待的延迟

private class WaitAndNext{
    private String next;
    private long delay;
}

一个将进行HTTP调用并处理响应的FLuxProcess：

FluxProcessor<WaitAndNext, WaitAndNext> processor= DirectProcessor.<WaitAndNext>create();
FluxSink<WaitAndNext> sink=processor.sink();

processor
    .flatMap(x-> Mono.just(x).delayElement(Duration.ofMillis(x.delay)))
    .map(x-> WebClient.builder()
    .baseUrl(x.next)
    .defaultHeader("Accept","application/json")
    .build())
    .flatMap(x->x.get()        
                 .exchange()
                 .flatMapMany(z->manageResponse(sink, z))
            )
    .subscribe(........);

我用一个只管理响应的方法拆分代码：它只是打开您的数据并向接收器添加一个新事件（在给定延迟后调用下一页的事件）

private Flux<Data> manageResponse(FluxSink<WaitAndNext> sink, ClientResponse resp) {

    if (resp.statusCode()!= HttpStatus.OK){
        sink.error(new IllegalStateException("Status code invalid"));
    }

    WaitAndNext wn=new WaitAndNext();
    HttpHeaders headers=resp.headers().asHttpHeaders();
    wn.delay= Integer.parseInt(headers.getFirst("X-Window-Remaining"))/ Integer.parseInt(headers.getFirst("X-Requests-Quota"));

    return resp.bodyToMono(Item.class)
        .flatMapMany(p -> {
            if (p.paginated.current==p.paginated.total){
                sink.complete();
            }else{
                wn.next="https://....?page="+(p.paginated.current+1);
                sink.next(wn);
            }
            return Flux.fromIterable(p.getData());
        });
}

现在我们只需要通过无延迟地调用第一页的检索来初始化系统：

WaitAndNext wn=new WaitAndNext();
wn.next="https://....?page=1";
wn.delay=0;
sink.next(wn);