我想進(jìn)行一個(gè)異步休息調(diào)用，我正在使用 spring webclient 并取回 Mono。我也在并行進(jìn)行一些數(shù)據(jù)庫(kù)調(diào)用，但由于某種原因無(wú)法被動(dòng)完成。 Map<String, Object> models = new HashMap<>(); Mono<User> users = this.webClient...; users.map(resp -> new UserState(userRequest, resp)) .subscribe(response -> { models.put("userState", response); }); Iterable<Product> messages = this.productRepository.findAll(); models.put("products", messages); //Wait for users.subscribe to finish <<<<<<<<<<<<<HERE return new ModelAndView("messages/list", models);如何在返回 ModelAndView 之前等待訂閱完成。Future如果我使用一個(gè)我可以get()隨時(shí)隨地做的地方，這會(huì)很容易。
1 回答
搖曳的薔薇
TA貢獻(xiàn)1793條經(jīng)驗(yàn) 獲得超6個(gè)贊
您可以將阻塞調(diào)用包裝在Mono一個(gè)單獨(dú)的調(diào)度程序上執(zhí)行，將其與Mono包含的UserState數(shù)據(jù)一起壓縮并將它們的組合轉(zhuǎn)換為一個(gè)Mono<ModelAndView>（可以從 Spring 控制器方法返回）。調(diào)用將并行執(zhí)行，兩個(gè)調(diào)用完成后將合并結(jié)果。
您可以為每個(gè)應(yīng)用程序定義一個(gè)專門用于阻塞調(diào)用的有界調(diào)度程序，并將其作為構(gòu)造函數(shù)參數(shù)提供給任何進(jìn)行阻塞調(diào)用的類。
代碼如下所示：
@Configuration
class SchedulersConfig {
@Bean
Scheduler parallelScheduler(@Value("${blocking-thread-pool-size}") int threadsCount) {
return Schedulers.parallel(threadsCount);
}
}
@RestController
class Controller {
final Scheduler parallelScheduler;
...
Mono<User> userResponse = // webClient...
Mono<Iterable<Product>> productsResponse = Mono.fromSupplier(productRepository::findAll)
.subscribeOn(parallelScheduler);
return Mono.zip(userResponse, productsResponse, (user, products) ->
new ModelAndView("messages/list",
ImmutableMap.of(
"userState", new UserState(userRequest, user),
"products", products
))
);
}
根據(jù)評(píng)論更新：
如果您只需要異步執(zhí)行 HTTP 調(diào)用，然后將其與數(shù)據(jù)庫(kù)結(jié)果結(jié)合起來(lái)，您可以執(zhí)行以下操作
Map<String, Object> models = new HashMap<>();
Mono<User> userMono = webClient...;
CompletableFuture<User> userFuture = userMono.toFuture();
Iterable<Product> messages = productRepository.findAll();
User user = userFuture.join();
models.put("products", messages);
models.put("userState", new UserState(userRequest, user));
return new ModelAndView("messages/list", models);
添加回答
舉報(bào)
0/150
提交
取消