前回はリアクティブプログラムミングの概要についてまとめました（記事はこちら）。今回はRxJavaの概要をまとめてみます。
| バージョン | パッケージ |
|---|---|
| 1.x | rx |
| 2.x | io.reactivex |
| Reactive Streams | 生産者(-able; 購読される) | 消費者(-er; 購読する) | 生産者&消費者(-erであり、-ableでもある) |
|---|---|---|---|
| Reactive Streams対応あり(バックプレッシャー機能あり） | Flowable | Subscriber | Processor |
| Reactive Streams対応なし(バックプレッシャー機能なし） | Observable | Observer | Subject |
RxJavaの大まかな流れ
Flowable（Reactive Streams 対応）を使ったサンプル
public void basicExample() {
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("Hello");
e.onNext("Welcome");
e.onNext("This is your first RxJava example");
e.onComplete();
}
});
Observer<String> observer = new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
LOGGER.info("observer subscribed to observable - on subscribe");
}
@Override
public void onNext(String value) {
LOGGER.info("observer - onNext " + value);
}
@Override
public void onError(Throwable e) {
LOGGER.info("observer - onError " + e.toString());
}
@Override
public void onComplete() {
LOGGER.info("observer - on complete");
}
};
observable.subscribe(observer);
}
Observable.from(new Integer[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10})
.filter(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer i) {
return (i % 2) == 0;
}
})
.map(new Func1<Integer, Integer>() {
@Override
public Integer call(Integer i) {
return i * 10;
}
})
.subscribe(new Observer<Integer>() {
@Override
public void onNext(Integer integer) {
Log.d("Hoge", integer.toString());
}
@Override
public void onCompleted() {
Log.d("Hoge", "completed");
}
@Override
public void onError(Throwable e) {}
});
public static void main(String[] args) throws Exception {
// あいさつの言葉を通知するFlowableの生成
Flowable < String > flowable =
Flowable.create(new FlowableOnSubscribe<String>() {
@Override
public void subscribe(FlowableEmitter<String> emitter)
throws Exception {
String[] datas = {
"Hello, World!",
"こんにちは、世界！"
};
for (String data: datas) {
// 購読解除されている場合は処理をやめる
if (emitter.isCancelled()) {
return;
}
// データを通知する
emitter.onNext(data);
}
// 完了したことを通知する
emitter.onComplete();
}
}, BackpressureStrategy.BUFFER); // 超過したデータはバッファする
flowable
// Subscriberの処理を別スレッドで行うようにする
.observeOn(Schedulers.computation())
// 購読する
.subscribe(new Subscriber<String>() {
// データ数のリクエストおよび購読の解除を行うオブジェクト
private Subscription subscription;
// 購読が開始された際の処理
@Override
public void onSubscribe(Subscription subscription) {
// SubscriptionをSubscriber内で保持する
this.subscription = subscription;
// 受け取るデータ数をリクエストする
this.subscription.request(1 L);
}
// データを受け取った際の処理
@Override
public void onNext(String data) {
// 実行しているスレッド名の取得
String threadName = Thread.currentThread().getName();
// 受け取ったデータを出力する
System.out.println(threadName + ": " + data);
// 次に受け取るデータ数をリクエストする
this.subscription.request(1 L);
}
// 完了を通知された際の処理
@Override
public void onComplete() {
// 実行しているスレッド名の取得
String threadName = Thread.currentThread().getName();
System.out.println(threadName + ": 完了しました");
}
// エラーを通知された際の処理
@Override
public void onError(Throwable error) {
error.printStackTrace();
}
});
// しばらく待つ
Thread.sleep(500 L);
}
次回は「【Programming】RxJava リアクティブプログラミング vol.3 / RxJavaの構成～前編～」についてまとめてみます。
written by tamito0201
プログラミングとのご縁結びならプロマリへ。
オンラインプログラミング学習スクールのプロマリは、プログラミングの初学者の皆様を応援しています。プログラミング講師と一緒に面白いアプリを作りませんか。
The programming school "Promari" will help you learn programming. "Promari" is supporting the first scholars of programming. Let's develop an application with our programming instructor.