需要帮助来理解所提到的代码示例的Flat Map执行

Question

我试图通过使用Flux和Webclient来调用rest api和累积结果。我想了解下面代码执行flatMap运算符。

• list - 包含50000个元素。
• restClient - 这是一个WebClient。 WebClient.builder().baseUrl(url).filter(ExchangeFilterFunctions.basicAuthentication(userId, pwd)).build()
• restClient.getResult - 返回一个Mono。 client.get().uri(uri).retrieve().bodyToMono(returnType)

第一个日志是打印线程名称作为主要的特定号码。然后它开始打印reactor-http-nio-（thread-no）。第二个日志是将线程名称打印为reactor-http-nio-（thread-no）。请解释一下线程执行和切换是如何发生的。

        Flux.fromIterable(list)
            .log()
            .map(e -> e.trim())
            .flatMap(e -> restClient.getResult(client, e, String.class))//e is being used to create uri
            .log()
            .subscribe(e -> result.add(e));

Answer 1

flatMap有一个concurrency参数，默认为256.一旦订阅，它就会从源（fromIterable）请求256个元素并将它们映射到内部发布者（WebClient调用）。

这是在订阅线程中完成的，这里是main线程。

一旦它有“飞行中”的256个元素，flatMap将在内部完成时根据需要补充，通过从迭代中请求一个或多个源元素，然后将其映射到新的WebClient内部。

这是在刚完成的内部线程中完成的，这里是netty http-nio-x线程。