1概要

このチュートリアルでは、RxJava2 ** 演算子を使用して、従来の同期および非同期APIを

Observables

に変換する方法を学習します。

これらの演算子について詳細に説明するのに役立つ簡単な関数をいくつか作成します。

2. Mavenの依存関係

まず、https://search.maven.org/search?q=g:io.reactivex.rxjava2%20AND%20a:rxjavaを追加する必要があります。 

<dependency>
    <groupId>io.reactivex.rxjava2</groupId>
    <artifactId>rxjava</artifactId>
    <version>2.2.2</version>
</dependency>
<dependency>
    <groupId>com.github.akarnokd</groupId>
    <artifactId>rxjava2-extensions</artifactId>
    <version>0.20.4</version>
</dependency>


3オペレータ

RxJava2では、リアクティブプログラミングのさまざまなユースケースに対してhttps://github.com/ReactiveX/RxJava/wiki/Alphabetical-List-of-Observable-Operators[多くの演算子]を定義しています。

しかし、同期または非同期メソッドをその性質に基づいて

Observables

に変換するために一般的に使用される少数の演算子についてのみ説明します。これらの演算子は引数として関数を取り、その関数から返された値を出力します。

通常の演算子に加えて、RxJava2では拡張機能用にさらにいくつかの演算子が定義されています。

同期メソッドと非同期メソッドを変換するためにこれらの演算子をどのように利用できるかを探りましょう。


4同期メソッド変換


4.1.

fromCallable（）


を使う

この演算子は

Observable

を返します。サブスクライバがそれをサブスクライブすると、引数として渡された関数が呼び出され、その関数から返された値が発行されます。整数を返す関数を作成して変換しましょう。 

AtomicInteger counter = new AtomicInteger();
Callable<Integer> callable = () -> counter.incrementAndGet();

それでは、それを

Observable

に変換して、購読してテストしましょう。 

Observable<Integer> source = Observable.fromCallable(callable);

for (int i = 1; i < 5; i++) {
    source.test()
      .awaitDone(5, TimeUnit.SECONDS)
      .assertResult(i);
    assertEquals(i, counter.get());
}


  • fromCallable（）

    演算子は、ラップされた

    Observable

    がサブスクライブされるたびに、指定された関数を遅延的に実行します。** この動作をテストするために、ループを使用して複数のサブスクライバを作成しました。

リアクティブストリームはデフォルトで非同期であるため、サブスクライバはただちに戻ります。ほとんどの実用的なシナリオでは、呼び出し可能関数はその実行を完了するためにある種の遅延があります。そのため、呼び出し可能な関数の結果をテストする前に、

最大待機時間を5秒

追加しました。


Observable

s


test（）


メソッドを使用したことにも注意してください。

このメソッドは便利ですhttps://www.baeldung.com/rxjava-testing[

Observables

]をテストするとき

それは

__TestObserver


を作成し、私達の

Observable.__を購読する


4.2.

start（）


を使う


start（）

演算子は

RxJava2Extension

モジュールの一部です。それは指定された関数を非同期的に呼び出し、結果を出す

Observable

を返します。 

Observable<Integer> source = AsyncObservable.start(callable);

for (int i = 1; i < 5; i++) {
    source.test()
      .awaitDone(5, TimeUnit.SECONDS)
      .assertResult(1);
    assertEquals(1, counter.get());
}

この関数は、サブスクライバが結果のObservableをサブスクライブするたびにではなく、ただちに呼び出されます。このオブザーバブルへの複数のサブスクリプションは同じ戻り値を観察します。


5非同期メソッド変換


5.1.

fromFuture（）


を使う

私たちが知っているように、Javaで非同期メソッドを作成する最も一般的な方法は

Future

実装を使うことです。

fromFuture

メソッドは引数として

Future

を取り、__Future.get（）メソッドから取得した値を出力します。

まず、先ほど作成した関数を非同期にしましょう。 

ExecutorService executor = Executors.newSingleThreadExecutor();
Future<Integer> future = executor.submit(callable);

次に、テストを変換してみましょう。 

Observable<Integer> source = Observable.fromFuture(future);

for (int i = 1; i < 5; i++) {
    source.test()
      .awaitDone(5, TimeUnit.SECONDS)
      .assertResult(1);
    assertEquals(1, counter.get());
}
executor.shutdown();

また、各サブスクリプションは同じ戻り値を遵守することに注意してください。

さて、

Observable

の

dispose（）

メソッドは、メモリリーク防止に関しては本当に役に立ちます。しかし、この場合、

Future.get（）

のブロック性のために将来をキャンセルすることはありません。

そのため、

source

observableの

doOnDispose（）

** 関数と

future

の

cancel

メソッドを組み合わせることで、確実に将来を取り消すことができます。 

source.doOnDispose(() -> future.cancel(true));


5.2.

startFuture（）


を使う

名前が示すように、この演算子は指定された

Future

をただちに開始し、サブスクライバがそれをサブスクライブすると戻り値を発行します。

次回の使用のために結果をキャッシュする

fromFuture

演算子とは異なり、この演算子はサブスクライブされるたびに非同期メソッドを実行します。 

ExecutorService executor = Executors.newSingleThreadExecutor();
Observable<Integer> source = AsyncObservable.startFuture(() -> executor.submit(callable));

for (int i = 1; i < 5; i++) {
    source.test()
      .awaitDone(5, TimeUnit.SECONDS)
      .assertResult(i);
    assertEquals(i, counter.get());
}
executor.shutdown();


5.3.

deferFuture（）


を使う

  • この演算子は、

    Future

    メソッドから返された複数の

    Observables

    を集約し、各__Observableから取得した戻り値のストリームを返します。


    新しいサブスクライバがサブスクライブするたびに、渡された非同期ファクトリ関数が開始されます。

それでは、まず非同期ファクトリ関数を作成しましょう。 

List<Integer> list = Arrays.asList(new Integer[]{ counter.incrementAndGet(),
  counter.incrementAndGet(), counter.incrementAndGet() });
ExecutorService exec = Executors.newSingleThreadExecutor();
Callable<Observable<Integer>> callable = () -> Observable.fromIterable(list);

それから簡単なテストを行います。 

Observable<Integer> source = AsyncObservable.deferFuture(() -> exec.submit(callable));
for (int i = 1; i < 4; i++) {
    source.test()
      .awaitDone(5, TimeUnit.SECONDS)
      .assertResult(1,2,3);
}
exec.shutdown();


6. 結論

このチュートリアルでは、同期メソッドと非同期メソッドをRxJava2観測量に変換する方法を学びました。

もちろん、ここに示した例は基本的な実装です。

しかし、ビデオストリーミングのようなより複雑なアプリケーションや、大量のデータを少しずつ送信する必要があるアプリケーションには、RxJava2を使用できます。

いつものように、ここで説明した短い例はすべてhttps://github.com/eugenp/tutorials/tree/master/rxjava-2[Githubプロジェクト]にあります。