用Project Reactor并行调用rest o web服务？

Question

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

类似于下面的代码构建与比较未来，但Reactor（通量，单声道）。

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

您可以使用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

Answer 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