DevelopRx

RxJavaの基本的な構成

• RxJavaの基本的な構成は生産者(Flowable/Observable)が消費者(Subscriber/Observer)に購読(subscribe)される形で成り立っている
• Flowable/Observable
  • データを生産し通知するクラス
  • FlowableとObservableの違いは、Flowableにはバックプレッシャーの機能があり、Observableには無い
• Subscriber/Observer
  • 通知されたデータを受け取りそのデータを使った処理を行うインターフェース
  • Subscriberにはバックプレッシャーの機能があり通知するデータ数をリクエストしないといけないのに対し、Observerはバックプレッシャーの機能が無いため、データ数の制限無しにデータが通知されることがある
• Subscription
  • Reactive Streamsで定義されているインターフェースで通知するデータの数をリクエストするrequestメソッドと処理の途中でも購読を解除するcancelメソッドを持っています
• Disposable
  • 購読を解除するためのメソッドを持つインターフェース
  • ObservableとObserver間の購読の際にonSubscribeメソッド経由で購読を処理の途中でも解除することができます
• RxJavaの拡張モジュール
  • RxAndroid:Androidで使うスレッドを管理するSchedulerを持つ
  • RxLifecycle:AndroidのActivityやFragmentをRxJavaのFlowableやObservableのライフサイクルと同期する

オペレータ内で生成される非同期のFLowable/Observable

• flatMap()
  • データを受け取ると、新たにFlowable/Observableを生成して実行し、そこで通知されたデータを結果として通知するオペレータ
  • 受け取るデータが連続してくるような場合、生成したFlowable/Observableが同時に実行され、それぞれがデータを通知したタイミングで結果としてデータを通知するので、最終的に通知されるデータは受け取った順に実行されるとは限らない
• concatMap()
  • flatMapと違う点は、受け取るデータが連続してくるような場合、最終的に通知されるデータは順序通り
  • １つのデータの処理に時間がかかる場合は、処理が終わるまで次のFLowable/Observableを生成しない
• concatMapEager()
  • flatMapと違う点は、受け取るデータが連続してくるような場合、最終的に通知されるデータは順序通り
  • 1つのデータの処理に時間がかかる場合は、全て違うスレッドで実行の後に生成されたデータがバッファされる

リソース管理

• using()
  • Observable.using()
  • リソースの取得、リソースから取得したデータを使ったFLowable/Observableの生成、リソースの解放を行うインターフェースが用意されている

RxJavaのバックプレッシャー

• Flowableにのみ用意されており、Observableにはない
• この機能が必要になるケースはFlowableとデータを受け取る側が異なるスレッド上で処理を行い、通知する処理スピードがデータを受け取る側の処理スピードより速い場合
• request()
  • SubscriberがonSubscribe()で受け取るSubscriptionの、Flowableがデータを何件まで通知するかをリクエストするメソッド
  • 1回目はonSubscribe()で通知されるデータ数のリクエストを行い、2回目以降はonNext()で処理を行った後に再度データ数のリクエストを行う
  • 1回目のonSubscribe()でrequest(Long.MAX_VALUE)とした場合は、通知するデータ数の制限がなくなり、生成されたデータは全て通知されるようになるため、onNext()でリクエストは不要となる
• observeOn()
  • scheduler:Scheduler, delayError:boolean, bufferSize:intを設定することが出来る
  • デフォルトでbufferSizeは128となっているため、バッファされるデータが128個を超えるとMissingBackpressureExceptionが起きる
• BackpressureStrategy
  • Flowableがどのように通知待ちのデータを扱うかのバックプレッシャーの種類
  • create()でFlowableを生成する際の第2引数にBackpressureStrategyを指定する
  • 種類
    • BUFFER:通知されるまで、全てのデータをバッファする
    • DROP:データを通知できるようになるまで、新たに生成されたデータを破棄する
    • LATEST:生成した最新のデータのみをバッファし、生成される度にバッファするデータを置き換える
    • 通知待ちのデータのバッファサイズを超す場合はMissingBackpressureExceptionのエラーを通知する
    • 特定の処理を行わない。主にonBackpressureで始まるメソッドを使ってバックプレッシャーのモードを設定する場合に使われる

通知する際のルール

• nullを通知してはならない
• データの通知は行っても行わなくても良い
• Flowable/Observableの処理を終了する際は完了かエラーの通知のどちらか一方を行わないといけない。両方を通知することはない
• 完了かエラーの通知を通知をした後は他の通知を行ってはいけない
• 通知をする際は1つずつ順番に行い同時に行ってはいけない