複数の観測可能な呼び出しを非同期的に呼び出すにはどうすればよいですか？

Question

私はRxJavaを使い慣れました。 Observablesを返すJersey RxJavaクライアントがいくつかあります。いくつかのデータを取得するために1回の呼び出しを行う必要があり、そのデータは次の3回の呼び出しの入力になります。これらの呼び出しを並行して実行したい。最後に、すべてのデータを必要とするすべての呼び出しが完了したら、計算を行いたいと思います。ここでは、それがどのように見えるかです：

interface Service { 
    Observable<ResultA> callServiceA(InitialData input); 
    Observable<ResultB> callServiceB(ResultA resultA); 
    Observable<ResultC> callServiceC(ResultA resultA); 
    Observable<ResultD> callServiceD(ResultA resultA); 
    FinalResult simpleCalculation(ResultA a, ResultB b, ResultC c, ResultD d); 
} 

class MyClass{ 

    @Autowired 
    ExecutorService myExecutorService; 

    Observable<FinalResult> myMethod(InitialData initialData){ 
    /* Make call to ServiceA, get the results, then make calls to services B, C, and D in parallel (on different threads), finally perform simpleCalculation, and emit the result */ 
    } 
} 

Answer 1

あなたはその後、flatMapそれが再び完了時に、その後、複数の呼び出しを送信するために郵便番号を使用するか、マージ、同期呼び出しを行うためにflatMapを使用することができます。

Answer 2

flatMap()およびzip()は、この状況のあなたの友人です。観察可能なリターンを使用して

Observable<FinalResult> myMethod(InitialData initialData) { 
    return service 
      .callServiceA(initialData) 
      .flatMap(resultA -> Observable.zip(
        service.callServiceB(resultA), 
        service.callServiceC(resultA), 
        service.callServiceD(resultA), 
        (resultB, resultC, resultD) -> 
         service.simpleCalculation(resultA, resultB, resultC, resultD)) 
      ); 
} 

は、次のようになります。

Subscription subscription = 
     myMethod(new InitialData()) 
       .subscribe(finalResult -> { 
          // FinalResult will end up here. 
         }, 
         throwable -> { 
          // Handle all errors here. 
         });