如何同时等待多个Mono完成并获取值

huangapple
go评论101阅读模式
英文:

How do I wait for multiple Mono's to complete at once and get the value

问题

  1. 类似于 https://stackoverflow.com/questions/49396692/waiting-for-running-reactor-mono-instances-to-complete 的问题，但我希望在另一个 Mono 中获得结果。以下是我的代码。我尝试了使用 materialize 的方法，但效果不佳。
  3. ```java
  4.     @GetMapping("/bounced")
  5.     public Mono<Map<String, Object>> bounced(
  6.         @RequestHeader("X-B3-Traceid") String traceId,
  7.         @RequestHeader(HttpHeaders.AUTHORIZATION) String authorization
  8.     ) {
  10.         final Mono<Map<String, Object>> sample = webClient.get()
  11.             .uri("http://sample:8080/")
  12.             .header(HttpHeaders.AUTHORIZATION, authorization)
  13.             .retrieve()
  14.             .bodyToMono(new ParameterizedTypeReference<>() {
  15.             });
  17.         final Mono<Map<String, Object>> httpGet = webClient.get()
  18.             .uri("http://httpbin.org/get")
  19.             .retrieve()
  20.             .bodyToMono(new ParameterizedTypeReference<>() {
  21.             });
  23.         final Mono<Map<String, Object>> anything = webClient.get()
  24.             .uri("http://httpbin.org/anything/foo")
  25.             .retrieve()
  26.             .bodyToMono(new ParameterizedTypeReference<>() {
  27.             });
  29. /*
  30.    尝试了下面的方法，它会启动执行，但会在返回块中触发另一个“download”。
  32.         Mono.when(anything, sample, httpGet)
  33.             .subscribe();
  34.             .materialize()
  35.             .block();
  36. */
  37.         return Mono.just(Map.of("traceFromBounced", traceId,
  38.             "anything", anything.block(),
  39.             "sample", sample.block(),
  40.             "httpGet", httpGet.block()));
英文:

Similar in question to https://stackoverflow.com/questions/49396692/waiting-for-running-reactor-mono-instances-to-complete but I want to get the result ideally in another Mono. Here's the code I have. I tried the materialize solution but that didn't pan out.

  1.     @GetMapping(&quot;/bounced&quot;)
  2.     public Mono&lt;Map&lt;String, Object&gt;&gt; bounced(
  3.         @RequestHeader(&quot;X-B3-Traceid&quot;) String traceId,
  4.         @RequestHeader(HttpHeaders.AUTHORIZATION) String authorization
  5.     ) {
  7.         final Mono&lt;Map&lt;String, Object&gt;&gt; sample = webClient.get()
  8.             .uri(&quot;http://sample:8080/&quot;)
  9.             .header(HttpHeaders.AUTHORIZATION, authorization)
  10.             .retrieve()
  11.             .bodyToMono(new ParameterizedTypeReference&lt;&gt;() {
  12.             });
  14.         final Mono&lt;Map&lt;String, Object&gt;&gt; httpGet = webClient.get()
  15.             .uri(&quot;http://httpbin.org/get&quot;)
  16.             .retrieve()
  17.             .bodyToMono(new ParameterizedTypeReference&lt;&gt;() {
  18.             });
  20.         final Mono&lt;Map&lt;String, Object&gt;&gt; anything = webClient.get()
  21.             .uri(&quot;http://httpbin.org/anything/foo&quot;)
  22.             .retrieve()
  23.             .bodyToMono(new ParameterizedTypeReference&lt;&gt;() {
  24.             });
  26. /*
  27.    Tried this and it does start it up, but it triggers another &quot;download&quot; in the return block.
  29.         Mono.when(anything, sample, httpGet)
  30.             .subscribe();
  31.             .materialize()
  32.             .block();
  33. */
  34.         return Mono.just(Map.of(&quot;traceFromBounced&quot;, traceId,
  35.             &quot;anything&quot;, anything.block(),
  36.             &quot;sample&quot;, sample.block(),
  37.             &quot;httpGet&quot;, httpGet.block()));

答案1

得分: 5

根据 @K.Nicholas 的评论，我让它正常运行了：

  1. @GetMapping("/bounced")
  2. public Mono<Map<String, Object>> bounced(
  3.     @RequestHeader("X-B3-Traceid") String traceId,
  4.     @RequestHeader(HttpHeaders.AUTHORIZATION) String authorization
  5. ) {
  7.     final Mono<Map<String, Object>> sample = webClient.get()
  8.         .uri("http://sample:8080/")
  9.         .header(HttpHeaders.AUTHORIZATION, authorization)
  10.         .retrieve()
  11.         .bodyToMono(new ParameterizedTypeReference<>() {
  12.         });
  14.     final Mono<Map<String, Object>> httpGet = webClient.get()
  15.         .uri("http://httpbin.org/get")
  16.         .retrieve()
  17.         .bodyToMono(new ParameterizedTypeReference<>() {
  18.         });
  20.     final Mono<Map<String, Object>> anything = webClient.get()
  21.         .uri("http://httpbin.org/anything/foo")
  22.         .retrieve()
  23.         .bodyToMono(new ParameterizedTypeReference<>() {
  24.         });
  26.     return Mono.zip(anything, sample, httpGet)
  27.         .map(t -> Map.of("traceFromBounced", traceId,
  28.             "anything", t.getT1(),
  29.             "sample", t.getT2(),
  30.             "httpGet", t.getT3()));
  32. }

这里是 Zipkin 的输出，显示它并行运行：
如何同时等待多个Mono完成并获取值

英文:

Based on @K.Nicholas' comment I got it working

  1.     @GetMapping(&quot;/bounced&quot;)
  2.     public Mono&lt;Map&lt;String, Object&gt;&gt; bounced(
  3.         @RequestHeader(&quot;X-B3-Traceid&quot;) String traceId,
  4.         @RequestHeader(HttpHeaders.AUTHORIZATION) String authorization
  5.     ) {
  7.         final Mono&lt;Map&lt;String, Object&gt;&gt; sample = webClient.get()
  8.             .uri(&quot;http://sample:8080/&quot;)
  9.             .header(HttpHeaders.AUTHORIZATION, authorization)
  10.             .retrieve()
  11.             .bodyToMono(new ParameterizedTypeReference&lt;&gt;() {
  12.             });
  14.         final Mono&lt;Map&lt;String, Object&gt;&gt; httpGet = webClient.get()
  15.             .uri(&quot;http://httpbin.org/get&quot;)
  16.             .retrieve()
  17.             .bodyToMono(new ParameterizedTypeReference&lt;&gt;() {
  18.             });
  20.         final Mono&lt;Map&lt;String, Object&gt;&gt; anything = webClient.get()
  21.             .uri(&quot;http://httpbin.org/anything/foo&quot;)
  22.             .retrieve()
  23.             .bodyToMono(new ParameterizedTypeReference&lt;&gt;() {
  24.             });
  26.         return Mono.zip(anything, sample, httpGet)
  27.             .map(t -&gt; Map.of(&quot;traceFromBounced&quot;, traceId,
  28.                 &quot;anything&quot;, t.getT1(),
  29.                 &quot;sample&quot;, t.getT2(),
  30.                 &quot;httpGet&quot;, t.getT3()));
  32.     }

Here's the zipkin output that shows it runs in parallel
如何同时等待多个Mono完成并获取值

huangapple
  • 本文由 发表于 2020年5月4日 10:53:25
  • 转载请务必保留本文链接：https://go.coder-hub.com/61584447.html
  • java
  • project-reactor
  • reactive-programming
  • spring-webflux
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定