通量-带webclient的并行平面图-固定批处理速率限制

Question

我的代码是这样的：

return Flux.fromIterable(new Generator()).log()
        .flatMap(
            s ->
                webClient
                    .head()
                    .uri(
                        MessageFormat.format(
                            "/my-{2,number,#00}.xml",
                            channel, timestamp, s))
                    .exchangeToMono(r -> Mono.just(r.statusCode()))
                    .filter(HttpStatus::is2xxSuccessful)
                    .map(r -> s),
            6)  //only request 6 segments in parallel via webClient
        .take(6) //we need only 6 200 OK responses
        .sort();

它只请求HEAD，直到前6个请求成功。

并行化在这里起作用，但问题是在一个请求完成后，它会立即触发下一个请求（以保持并行化级别为6）。这里我需要的是并行化级别为6，但要分批进行。所以-触发6个请求，等待所有完成，再次触发6个请求。。。

这是上面的log（）的输出：

: | request(6)
: | onNext(7)
: | onNext(17)
: | onNext(27)
: | onNext(37)
: | onNext(47)
: | onNext(57)
: | request(1) <---- from here NOT OK; wait until all complete and request(6)
: | onNext(8)
: | request(1)
: | onNext(18)
: | request(1)
: | onNext(28)
: | request(1)
: | onNext(38)
: | request(1)
: | onNext(48)
: | request(1)
: | onNext(58)
: | cancel()

使现代化

这是我用缓冲区尝试的：

return Flux.fromIterable(new Generator())
        .buffer(6)
        .flatMap(Flux::fromIterable)
        .log()
        .flatMap(
            s ->
                webClient
                    .head()
                    .uri(
                        MessageFormat.format(
                            "/my-{2,number,#00}.xml",
                            channel, timestamp, s))
                    .exchangeToMono(r -> Mono.just(r.statusCode()))
                    .filter(HttpStatus::is2xxSuccessful)
                    .map(r -> s),
            6)  //only request 6 segments in parallel via webClient
        .take(6)
        .sort();

Answer 1

好的，看来我有代码了。这里我使用窗口：

return Flux.fromIterable(new Generator())
        .window(6) //group 1,2,3,4,5,6,7... into [0,1,2,3,4,5],[6,7..,11],[12,..,17]
        .log()
        .flatMap(
            s -> s.log().flatMap(x -> webClient
                .head()
                .uri(
                    MessageFormat.format(
                        "/my-{2,number,#00}.xml",
                        channel, timestamp, x))
                .exchangeToMono(r -> Mono.just(r.statusCode()))
                .filter(HttpStatus::is2xxSuccessful)
                .map(r -> x), 6), 1)  //1 means take only 1 array ([0,1,2,3,4,5]). 6 means take in parallel all from array (0,1,2,3,4,5)
        .take(6, true) //pass through only 6 elements (cancel afterwards)
        .sort();