1. 概念
リアクティブプログラミングは、データストリームと変化の伝播に焦点を当てたプログラミングパラダイムです。これには3つのキーワードが含まれ、以下のように説明できます。
【データストリーム】あらかじめ決められた順序で一度だけ読み取られるデータのシーケンスです。
コンピュータでは受け渡されるデータを指し、現実世界では任意のオブジェクトからなる順序付きのキューとして考えられます。たとえば、順番待ちをして一人ずつチケットを受け取るようなものです。
【変化の伝播】オブザーバーパターンに似ており、変化が生じたら観察者に通知します。
たとえばレストランで食事をするとき、注文する料理が異なれば、厨房が作る料理も異なります。これはドミノ効果に似ています。変化が伝播していくのです。
【プログラミングパラダイム】コンピュータプログラミングの基本的なスタイルや典型的なパターンであり、たとえば一般的なオブジェクト指向プログラミングや手続き型プログラミングなどがあります。
これらを結びつける例として、工場の生産ラインが挙げられます。異なる作業場を移動する加工対象の物品はデータストリームに似ており、各作業場が物品に対して異なる加工工程を持ち、絶えず伝播していく様子は変化の伝播に例えられます。
1つの工場に複数の生産ラインがある場合、各ラインで加工される物品は異なり、それぞれの生産ラインは異なるプログラミングパラダイムに似ています。
【RxJava】RxJavaライブラリのリポジトリは https://github.com/ReactiveX/RxJava です。公式の説明を引用します。
RxJavaはReactive ExtensionsのJava VM実装です。観測可能なシーケンスを使って、非同期かつイベントベースのプログラムを構成するためのライブラリです。
【asynchronous】RxJavaは非同期を基盤とし、コールバック特性を持つライブラリです。
【event-based】RxJavaはイベント配信機能とメッセージ伝達機能を備えています。
2. RxJava 1の要素と関係
RxJava 1のコード例
Subscription subscription = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> sub) {
if (!sub.isUnsubscribed()) {
sub.onNext("alpha");
sub.onNext("beta");
sub.onCompleted();
}
}
}).subscribe(new Observer<String>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable error) {
error.printStackTrace();
}
@Override
public void onNext(String value) {
System.out.println("onNext: " + value);
}
});5つの要素があります。
【1】【Observable】被観測者です。Observable.createで観測可能なシーケンスを生成し、subscribeで観察者を登録します。
【2】【Observer】観察者です。データを受信し、Observableのsubscribeメソッドの引数として渡されます。
【3】【Subscription】購読です。被観測者と観察者の関係を表し、購読の解除や現在の購読状態を扱う2つのメソッドを持ちます。
【4】【OnSubscribe】購読アクションが発生したときに呼び出されるインターフェースです。Observableのソースコードでは購読者にデータを発行します。
【5】【Subscriber】ObserverとSubscriptionを実装しており、自身だけが自身を停止できます。
3. RxJava 2
3.1 バックプレッシャー問題の理解
バックプレッシャー問題は非同期環境で発生し、データ送信速度とデータ処理速度が異なることによって生じる問題です。バックプレッシャー戦略は流量制御の解決策です。
生産者と消費者の問題で説明できます。生産が速すぎて生産過剰になり、消費者に負荷がかかります。
3.2 RxJava 2の例と構成要素
RxJava 2ではバックプレッシャー問題の処理が分離され、Observableはバックプレッシャーをまったく処理せず、Flowableが処理を担当します。Flowableはバックプレッシャー戦略を持ちます。
RxJava 2の例:
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
if (!emitter.isDisposed()) {
emitter.onNext("A");
emitter.onNext("B");
emitter.onNext("C");
emitter.onNext("D");
emitter.onNext("E");
emitter.onComplete();
}
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("onSubscribe");
}
@Override
public void onNext(String value) {
System.out.println("onNext: " + value);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
Flowable.create(new FlowableOnSubscribe<String>() {
@Override
public void subscribe(FlowableEmitter<String> emitter) throws Exception {
if (!emitter.isCancelled()) {
emitter.onNext("payload");
emitter.onComplete();
}
}
}, BackpressureStrategy.DROP).subscribe(new Subscriber<String>() {
@Override
public void onSubscribe(Subscription subscription) {
subscription.request(Long.MAX_VALUE); // リアクティブな取得
System.out.println("onSubscribe");
}
@Override
public void onNext(String value) {
System.out.println("onNext: " + value);
}
@Override
public void onError(Throwable t) {
t.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});5つの要素があります。
【1】【ObservableとFlowable】
【2】【ObserverとSubscriber】
【3】【DisposableとSubscription】
【4】【ObservableのOnSubscribeとFlowableのOnSubscribe】
【5】【Emitter】
3.3 バックプレッシャーなしのRxJava 2
バックプレッシャーなしのRxJava 2は本質的にRxJava 1とほとんど違いがなく、新しいクラスでラップされているだけです。
バックプレッシャーなしのRxJava 2の要素
【1】【Observable】バックプレッシャーをサポートしません。被観測者であり、Observable.createで観測可能なシーケンスを生成し、subscribeで観察者を登録します。
【2】【Observer】観察者です。データを受信し、Observableのsubscribeメソッドの引数として渡されます。
【3】【Disposable】RxJava 1のSubscriptionと同じ役割を持ち、購読の解除や現在の購読状態を扱います。
【4】【ObservableのOnSubscribe】購読アクションが発生したときに呼び出されるインターフェースです。Observableのソースコードでは購読者にデータを発行します。
【5】【Emitter】データを発行するインターフェースです。Observerのメソッドに似ており、本質的にはObserverとSubscriberのラッパーです。
3.4 バックプレッシャーありのRxJava 2
【1】【Flowable】バックプレッシャーをサポートします。被観測者であり、Flowable.createで観測可能なシーケンスを生成し、subscribeで購読者を登録します。
【2】【Subscriber】独立したインターフェースで、Observerのメソッドに似ており、Flowableのsubscribeメソッドの引数として渡されます。
【3】【Subscription】購読です。RxJava 1とは異なり、バックプレッシャーをサポートし、バックプレッシャー用のrequestメソッドが存在します。
【4】【FlowableOnSubscribe】購読アクションが発生したときに呼び出されるインターフェースです。Flowableのソースコードでは購読者/観察者にデータを発行します。
【5】【Emitter】データを発行するインターフェースです。Observerのメソッドに似ており、本質的にはObserverとSubscriberのラッパーです。
Flowable.create(new FlowableOnSubscribe<String>() {
@Override
public void subscribe(FlowableEmitter<String> emitter) throws Exception {
if (!emitter.isCancelled()) {
emitter.onNext("first");
emitter.onNext("second");
emitter.onComplete();
}
}
}, BackpressureStrategy.DROP).subscribe(new Subscriber<String>() {
@Override
public void onSubscribe(Subscription subscription) {
subscription.request(Long.MAX_VALUE); // バックプレッシャー問題を解決するリアクティブな取得
System.out.println("onSubscribe");
}
@Override
public void onNext(String value) {
System.out.println("onNext: " + value);
}
@Override
public void onError(Throwable t) {
t.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});バックプレッシャー戦略のソースコード
@Override
public void subscribeActual(Subscriber<? super T> downstream) {
BaseEmitter<T> emitter;
switch (backpressure) {
case MISSING:
emitter = new MissingEmitter<T>(downstream);
break;
case ERROR:
emitter = new ErrorAsyncEmitter<T>(downstream);
break;
case DROP:
emitter = new DropAsyncEmitter<T>(downstream);
break;
case LATEST:
emitter = new LatestAsyncEmitter<T>(downstream);
break;
default:
emitter = new BufferAsyncEmitter<T>(downstream, bufferSize());
break;
}
}