java - 如何链接多个并发 rx-java Single

Question

我正在使用 Vert.x 学习 Rx-Java2，我想将成功的配置检索与一些并行任务链接起来。

我创建了一个方法来搜索配置并返回单个订阅，并且效果很好。但我不确定在哪里以及如何调用后续任务:

   public void start(Future<Void> startFuture) throws Exception {
      Single<JsonObject> configSingle = prepareConfigurationAsync();
      configSingle.subscribe(onSuccess -> {
        System.out.println(onSuccess);
        -->   Single<Boolean> task1 = prepareLongAsyncTask1(onSuccess).subscribe(...); 
        -->   Completable task2 = prepareLongAsyncTask2(onSuccess)..subscribe(...); 

    }, onError -> {
        startFuture.fail(onError);
    }));

我的方法似乎有效，但没有并行性。我怎样才能实现它？

我应该如何以及在哪里处理这些订阅？

Answer 1

继续使用其他来源通常是通过flatMap完成的。并行执行操作通常通过 zip 或 merge 来完成。就您而言，我认为您不需要内部 Single 的值作为输出的一部分，因此您可以尝试以下操作:

 Completable config = prepareConfigurationAsync()
     .flatMapCompletable(success ->
         System.out.println(success);
         return Completable.mergeArray (
             prepareLongAsyncTask1(success)
                 .doOnSuccess(innerSuccess -> /* ... */)
                 .toCompletable(),
             prepareLongAsyncTask2(success)
                 .doOnComplete(() ->  /* ... */)
         )
      );

  config
  .subscribe( () -> /* completed */, error -> /* error'd */);