java - 使用 webclient 和 Flux 进行多个异步剩余分页调用

Question

我需要使用 webclient 和 Flux 调用其余分页 api。我尝试过以阻塞方式(一个接一个)，但我想让它并行。假设一次 10 个并行调用。每次调用都会获取 1000 条记录。
我已经在调用第 0 个请求以从 header 获取总记录数。请求完成后，我需要调用 POST api 发送此响应(1000 条记录)。

如果任何请求完成，则将发送第 11 个请求，依此类推。我已经看到了 asyncRestTemplate 和可监听 futures 的其他示例，但是 asyncRestTemplate 已经被弃用，替代方案是 spring-webflux

还有
如rest template即将被弃用

我做了什么。

1. 除以总计数/1000 -> 得出总页数
2. 循环直到 5(如果我更改为 totalpages 计数，那么它会给我 500 内部服务器错误)
3. 调用返回 Mono 的服务>
4. 订阅每个请求

ObjectMapper objmapper = new ObjectMapper();

HttpHeaders headers = partsService.getHeaders();
long totalCount = Long.parseLong(headers.get("total-count").get(0));
log.info(totalCount);
long totalPages = (long) Math.ceil((double) totalCount / 1000);
log.info(totalPages);
// List<Mono<List<Parts>>> parts = new ArrayList<>();
for (long i = 1; i <= 5; i++) {
    partsService.fetchAllParts(1000L, i).log().subscribe(partList -> {
        try {
            // post each request response to another API
            log.info(objmapper.writeValueAsString(partList));
        } catch (JsonProcessingException ex) {
            ex.printStackTrace();
        }

    });
    log.info("Page Number:" + i);
}

我希望并行执行而没有任何 outOfmemoryerror 并且不给调用 api 带来太多负担。另外，我尝试一次获取所有页面，但收到 500 内部服务器错误。

我是 Flux(项目 react 器)的新手

Implemented below solution

它不是并行运行的，单个请求大约需要 2 分钟时间，这意味着所有 10 个(并发级别)应该同时完成。

try {
        fetchTotalCount().log()
                .flatMapMany(totalCount -> createPageRange(totalCount, 1000)).log()
                .flatMap(pageNumber -> fetch(1000, pageNumber), 10).log()
                .flatMap(response -> create(response))
                .subscribe();

        } catch (Exception e) {
            e.printStackTrace();
        }

Logs

2019-07-29T09:00:14,477 INFO  [scheduling-1] r.u.Loggers$Slf4JLogger: request(10)
2019-07-29T09:00:14,478 INFO  [scheduling-1] r.u.Loggers$Slf4JLogger: request(10)
2019-07-29T09:00:14,479 INFO  [scheduling-1] r.u.Loggers$Slf4JLogger: request(unbounded)
2019-07-29T09:00:14,679 INFO  [scheduling-1] c.o.q.l.Logging: fetch() execution time: 546 ms
2019-07-29T09:00:17,028 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(74577)
2019-07-29T09:00:17,042 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(1)
2019-07-29T09:00:17,068 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,1) execution time: 24 ms
2019-07-29T09:00:17,078 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(2)
2019-07-29T09:00:17,080 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,2) execution time: 2 ms
2019-07-29T09:00:17,083 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(3)
2019-07-29T09:00:17,087 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,3) execution time: 2 ms
2019-07-29T09:00:17,096 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(4)
2019-07-29T09:00:17,098 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,4) execution time: 1 ms
2019-07-29T09:00:17,100 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(5)
2019-07-29T09:00:17,101 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,5) execution time: 1 ms
2019-07-29T09:00:17,103 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(6)
2019-07-29T09:00:17,106 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,6) execution time: 3 ms
2019-07-29T09:00:17,108 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(7)
2019-07-29T09:00:17,110 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,7) execution time: 2 ms
2019-07-29T09:00:17,113 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(8)
2019-07-29T09:00:17,115 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,8) execution time: 1 ms
2019-07-29T09:00:17,116 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(9)
2019-07-29T09:00:17,118 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,9) execution time: 1 ms
2019-07-29T09:00:17,119 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(10)
2019-07-29T09:00:17,121 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,10) execution time: 1 ms
2019-07-29T09:00:17,123 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onComplete()
2019-07-29T09:09:03,295 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: request(1)
2019-07-29T09:09:03,296 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(11)
2019-07-29T09:09:03,296 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,11) execution time: 0 ms
2019-07-29T09:09:03,730 INFO  [reactor-http-nio-1] c.o.q.s.Scheduler: 200 OK
2019-07-29T09:09:03,730 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: request(1)
2019-07-29T09:09:05,106 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(// data print)
2019-07-29T09:09:05,196 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: request(1)
2019-07-29T09:09:05,196 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(12)
2019-07-29T09:09:05,198 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,12) execution time: 1 ms
2019-07-29T09:09:05,466 INFO  [reactor-http-nio-1] c.o.q.s.Scheduler: 200 OK
2019-07-29T09:09:05,466 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: request(1)
2019-07-29T09:09:09,565 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(//  data print)
2019-07-29T09:09:09,730 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: request(1)
2019-07-29T09:09:09,730 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(13)
2019-07-29T09:09:09,731 INFO  [reactor-http-nio-1] c.o.q.l.Logging: fetch(1000,13) execution time: 0 ms
2019-07-29T09:09:10,049 INFO  [reactor-http-nio-1] c.o.q.s.Scheduler: 200 OK

Update

更正调用 API 后，记录即将到来，但在获取最后一页(75)后，我收到 404 未找到错误。

2019-07-30T14:07:50,071 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(75)
2019-07-30T14:07:50,075 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onComplete()
2019-07-30T14:07:50,322 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(200 OK)
2019-07-30T14:07:50,323 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: request(1)
2019-07-30T14:07:51,973 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(//data)
2019-07-30T14:07:52,440 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(200 OK)
2019-07-30T14:07:52,440 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: request(1)
2019-07-30T14:07:54,522 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(//data)
2019-07-30T14:07:54,699 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(//data)
2019-07-30T14:07:55,075 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(200 OK)
2019-07-30T14:07:55,076 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: request(1)
2019-07-30T14:07:55,371 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onNext(200 OK)
2019-07-30T14:07:55,371 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: request(1)
2019-07-30T14:07:55,471 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: cancel()
2019-07-30T14:07:55,472 INFO  [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: cancel()
2019-07-30T14:07:55,473 ERROR [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: onError(java.lang.Exception: 4XX received from API)
2019-07-30T14:07:55,473 ERROR [reactor-http-nio-1] r.u.Loggers$Slf4JLogger: 
java.lang.Exception: 4XX received from API

Answer 1

Flux.flatMap有一个参数来设置并发级别，让您可以协调并行化。

在下面的示例中，我使用了虚拟 URL、示例中的一些片段以及一些其他简单代码来演示如何实现此目的:

public static void main(String[] args)
{
    fetchTotalCount()
            .flatMapMany(totalCount -> createPageRange(totalCount))
            .flatMap(pageNumber -> fetch(pageNumber), 5) // 5 is the concurrency level = how many pages we query concurrently
            .flatMap(response -> process(response))
            .subscribe();
}

private static Mono<Integer> fetchTotalCount()
{
    return webClient.get()
                    .uri("http://www.example.com/get-total-count")
                    .exchange()
                    .map(ClientResponse::headers)
                    .map(headers -> headers.asHttpHeaders().get("total-count").get(0))
                    .map(Integer::valueOf);
}

private static Flux<Integer> createPageRange(int totalCount)
{
    int totalPages = (int) Math.ceil((double) totalCount / 1000);

    return Flux.range(1, totalPages);
}

private static Mono<Response> fetch(int pageNumber)
{
    return webClient.get()
                    .uri("http://www.example.com/fetch?page=" + pageNumber)
                    .retrieve()
                    .bodyToMono(Response.class);
}

private static Mono<Response> process(Response response)
{
    // todo send other http request for the post api here
    return Mono.just(response);
}

private static class Response
{
}