使用 Project Reactor 并行调用其余 Web 服务？

Question

我想知道如何对 REST 或 Web 服务进行多个并行调用，然后加入响应并在调用 @RestController 的响应中发送它。

类似于使用 ComparableFuture 构建的以下代码，但使用 Reactor(Flux, Mono)。

CompletableFuture<Company> companyCompletableFuture = CompletableFuture.supplyAsync(() -> {
     return  Company.find.where().eq("id", id).findUnique();  
});

CompletableFuture<List<Domain>> domainsCompletableFuture = CompletableFuture.supplyAsync(() -> {
     return Domain.find.where().eq("company_id", id).findList();
});

// wait for all the data
CompletableFuture allDoneFuture = CompletableFuture.allOf(companyCompletableFuture, domainsCompletableFuture);

allDoneFuture.get(); // wait for all done
company = companyCompletableFuture.get();
domain = domainsCompletableFuture.get()

Answer 1

您可以从 callable 创建两个 Mono，然后压缩它们。如果你想并行执行可调用的，你还需要显式添加subscribeOn(Schedulers.parallel())到每个Mono：

Mono<Integer> mono1 = Mono.fromCallable(() -> {
    System.out.println(Thread.currentThread().getName());
    return 123;
}).subscribeOn(Schedulers.parallel());

Mono<Integer> mono2 = Mono.fromCallable(() -> {
    System.out.println(Thread.currentThread().getName());
    return 321;
}).subscribeOn(Schedulers.parallel());

Tuple2<Integer, Integer> result = mono1.zipWith(mono2).block();

System.out.println(result.getT1());
System.out.println(result.getT2());

结果会是这样的：

parallel-1
parallel-2
123
321