spring - 使用reactor等待多个http响应
问题描述
我正在使用 Spring reactor 进行多个 http 调用并等待所有结果。这是我的代码:
final Map<String, WSSearchResult> reduced = new HashMap<>();
List<Mono<ClientResponse>> monos = new ArrayList<>();
for (int i=0;i<10;i++) {
log.info("Executing http call {}", i);
WSSearchRequest wsSearchRequest = WSSearchRequest.builder().param(i).build();
Mono<ClientResponse> exchange = webClient.post().uri("/search/availability")
.body(BodyInserters.fromObject(wsSearchRequest)).exchange();
monos.add(exchange);
}
final CountDownLatch latch = new CountDownLatch(monos.size());
Flux.merge(monos).subscribe(clientResponse -> {
List<WSSearchResult> partialResult = clientResponse.bodyToFlux(WSSearchResult.class).collectList().block();
List<WSSearchResult> partial =
partialResult.parallelStream().filter(w-> !Strings.isNullOrEmpty(w.getId())).
collect(Collectors.toList());
mapAndReduce(partial, reduced);
});
try {
latch.await(150, TimeUnit.SECONDS);
} catch (InterruptedException e) {
log.error(e.getMessage(), e);
}
问题是上面的代码给出了异常:
reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-epoll-6
Caused by: java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-epoll-6
at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:77) ~[reactor-core-3.2.8.RELEASE.jar!/:3.2.8.RELEASE]
at reactor.core.publisher.Mono.block(Mono.java:1494) ~[reactor-core-3.2.8.RELEASE.jar!/:3.2.8.RELEASE]
我是反应堆开发的新手,我不知道为什么它给了我这个例外。
有人可以让我走上正确的道路吗?
解决方案
感觉这段代码可以用以下方式重写:
List<Mono<ClientResponse>> clientResponses = IntStream.range(0, 10)
.mapToObj(i -> WSSearchRequest.builder().param(i).build())
.map(request -> send(request))
.collect(toList());
Mono<Map<String, WSSearchResult>> reduced = Flux.merge(clientResponses)
.flatMap(clientResponse ->
clientResponse.bodyToFlux(WSSearchResult.class)
.filter(result -> !Strings.isNullOrEmpty(result.getId()))
.collectList()
.map(listOfFilteredResults -> map(listOfFilteredResults))
).reduce(new HashMap<>(), (left, right) -> reduce(left, right));
reduced
.timeout(Duration.ofSeconds(150))
.subscribe(result -> handle(result));
private Mono<ClientResponse> send(WSSearchRequest request) {
return webClient.post().uri("/search/availability")
.body(BodyInserters.fromObject(wsSearchRequest))
.exchange();
}
推荐阅读
- angular - 读取“id”时出错
- c++ - 从 std::map 值获取密钥的有效方法
- android - Android Socket.IO
- api - 当请求对象变为 NULL 时,如何在 .Net Core API 中获取实际的请求对象?
- .net - .net mvc,为用户设置角色来自outlook
- python - 当应用在没有 for 循环的情况下无法工作时,如何迭代 DataFrame?
- google-chrome - 量角器 woant 运行我的脚本
- html - 在 Amazon S3 上显示 em-dash
- javascript - 即使目标是支持 ES 模块的现代浏览器,是否需要打包程序?
- ionic3 - 更改离子列表项单击时的复选框状态?