使用 Project Reactor 并行调用其他 Web 服务？答案

【问题标题】：Call in parallel rest o web services with Project Reactor?使用 Project Reactor 并行调用其他 Web 服务？
【发布时间】：2018-07-24 22:54:05
【问题描述】：

我想知道如何对 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()

【问题讨论】：

我认为 fork 和 join 符合您的要求？或者您可以使用循环障碍并等待完成？

标签： java spring spring-boot project-reactor

【解决方案1】：

您可以使用 subscribeOn 在新线程和 zip 运算符中运行

 /**
     * SubscribeOn just like in Rx make the pipeline run asynchronously, from the beginning to the end.
     * <p>
     * In this example we get the three Flux pipelines and we run async all of them.
     * Once they are finish we zip the results in the order we establish in the operator.
     */
    @Test
    public void subscribeOn() throws InterruptedException {
        Scheduler scheduler = Schedulers.newElastic("thread");
        Scheduler scheduler1 = Schedulers.newElastic("thread");
        Scheduler scheduler2 = Schedulers.newElastic("thread");

        Flux<String> flux1 = Flux.just("hello ")
                .doOnNext(value -> System.out.println("Value " + value + " on :" + Thread.currentThread().getName()))
                .subscribeOn(scheduler);
        Flux<String> flux2 = Flux.just("reactive")
                .doOnNext(value -> System.out.println("Value " + value + " on :" + Thread.currentThread().getName()))
                .subscribeOn(scheduler1);
        Flux<String> flux3 = Flux.just(" world")
                .doOnNext(value -> System.out.println("Value " + value + " on :" + Thread.currentThread().getName()))
                .subscribeOn(scheduler2);
        Flux.zip(flux1, flux2, flux3)
                .map(tuple3 -> tuple3.getT1().concat(tuple3.getT2()).concat(tuple3.getT3()))
                .map(String::toUpperCase)
                .subscribe(value -> System.out.println("zip result:" + value));
        Thread.sleep(1000);

    }

您可以在此处查看更多 Reactor 示例 https://github.com/politrons/reactive

【讨论】：

【解决方案2】：

你可以从 callable 创建两个 Mono 然后压缩它们。如果你想并行执行 callable，你还需要显式地将 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

【讨论】：