update.なんでこの記事書いたのか今開発中のプロダクトにおいて、RxJavaの導入をやってみたので、実際に使った箇所とその例、調べないとわからなかったことを載せておきました。(retrolambdaのためにjdk8を投入)までして導入したかったメリットを話してくれ、サンプルコードがないとわからん、といった声を頂いているので、実際に何が解決されたのか、どんなコードで解決したのかということと、そのために勉強しなくてはならなかった点について書いています。続報あり次第追記します。追記:リリースして安定運用しています!最近まで監視に難がありましたがそれも修正しました。この記事の監視スニペットも更新済みです!なぜRxJavaを導入したのか次の課題をまとめて解決できるのがRxJavaだと思い、リファクタの勢いも相まって導入(勢いですみません・・・)。チーム内でAndroidの非同期処理やそのエラーハンドリングをするのが面倒だという話がずっと前から出ていたリスト操作やlambdaで書いたほうが可読性の高い箇所が見受けられていたリファクタで、データソースが変更された時の表示の更新をイベントによって実現しようという方針に決まったCookpadでは最初はリスト操作から導入したとのことです:Android開発でRxJavaをチームに導入した話。このコードだとメリットが良くわからないと煽られたwのもあり、LoaderとEventBusで実装されていた箇所をRxJavaに置き換えることになりました。Programmingの概念については参考記事を書くにとどめ、この記事では実際の使用例について扱います。Programmingについては理解がないことをご容赦ください。共通する使い方だけ下記に記載しておきます。コードサンプルは続きをご覧ください。Observableからはitemが同期的or非同期に流れてくる(ストリーム)Observable.from(List)を使うと、Listの中身が1つずつ流れてくるObservableを作ることができるmap()やfilter()を始めとするOperatorを使うとストリームに流れてくるitemを加工できる原則subscribe()で処理を開始(またはobserve開始)し、コールバックに順次itemが流れてくるitemの到着を知らせるonNext()、ストリームの終了を伝えるonComplete()、エラー発生による終了を伝えるonError()がある同期的に結果を得たい場合は、toBlocking()を使用するobserveOn()にSchedulerを指定すると、subscribe()のコールバックを指定したスレッド(メインスレッドなど)で呼び出すことができるキャンセルしたい場合はunsubscribe()する(と、適切に実装されたObservableではActivity/Fragmentがリークしない)メインのドキュメントはRxJavaのrepoのWikiを見るのが良さげでした。挙動の確認はrxmarbles.comが多分一番便利です。RxJSで書かれているので、名前が違うメソッドとかあるかもしれません。RxJavaはただ単にループしているわけじゃない為、zip()など一部の処理が低速なようです。APIなどの(小さな?)バックポートであるLightweight-Stream-APIをおすすめします。メソッド数は350程度です。APIを使ってmapやfilterのような、Ruby、Python、underscore.js/EcmaScript5でおなじみの、forループを使わないリスト操作ができるようになりました。8が使えるのはずっと先になりそうな感じです・・。全部Listにまとめ、同期処理するよう指定し、1つしかない結果(List)を取り出す非同期処理を前提としているので、同期処理に変えるためのtoBlocking()の呼び出しが必要でちょっと癖がありますが、そこさえクリアすれば十分実用的かなと思いました。loopsをRxJavaでやる(Androidの環境で)非同期(バックグラウンド)処理が面倒くさい問題Androidのフレームワークで(Viewに結果を反映するような)バックグラウンド処理を実行する際に使うパターンは主に下記の3つです。それぞれ問題点があります。AsyncTask実行中にActivity/Fragmentがdestroyされると、特別な考慮をしない限り参照がリークしたりクラッシュする特にcancel時の扱いが難しいエラーハンドリングが面倒AsyncTaskLoader正しく実装するためのboilerplate(コピペ)が非常に難しい(以前コード読んだときはリンク先の公式サンプルも間違っているように見えた)destroyや画面回転のとき、Activity/Fragmentが非アクティブなときなど、条件ごとの挙動があり複雑Activity/Fragmentでしか使用できないパラメータがbundleでしか渡せず(意図はわかるが)素直じゃないエラーハンドリングが面倒IntentService結果を返却する方法がイベント(EventBusやLocalBroadcast)に限られる作るごとにAndroidManifestへの追加が必要それ、RxJavaで解決できるよ解決策として、Promiseを導入する手段があります。RxJavaのObservableはPromiseのように使用することができます。clientライブラリのretrofitはRxJavaに対応していて、Observableを返却することができます。retrofitの場合、最初からバックグラウンドスレッドで処理されます}これを使った実装例は下記のようなものになります。※mainThread()はRxAndroidの実装を参照のこと。...}このようにAsyncTaskやAsyncTaskLoader、IntentServiceを新たに設置しなくても、リークしないしシンプルなコードで書けます・・!今回は画面回転時のキャッシュなどはしない割り切った実装ですが、できればその辺りにもtryしてみようと思います。※PromiseとObservableの大きな違いは、複数の値を返却できる点です。今回自分で作った実装ではAndroidSchedulers.mainThread()はRxAndroidのものを利用せず、参考に実装しなおしたものを使っています。RxAndroidを使わなかった理由は後述します。表示のたびにFragmentからModelを呼び出しているような場所に、変更通知イベントによる表示の更新を加えると、呼び出しの流れがやや複雑になってしまいます。EventBusなどを介すると、そのイベントがどこから飛んでくるのかが一目ではわからないという課題もあります。これをObserverパターンにすれば、変更内容が通知されるごとにrenderするだけで良くなり、またsubscribeするときにModelを参照しているので可読性が高まります。┘それ、RxJavaで(ry追記:元の実装だとタイミングによっては最新の状態を受け取れなくなってしまうのと、onBackpressureLatest()が効果がない状態だったので、replay(1)を使って書き直しました。ObservableはonComplete()やonError()を呼びださなければ、いつ何度でもデータを流すことができます。最新の1つだけをキューイングし残りを捨てる。値が飛んできたらリクエストを投げるObservableを返す。flatMapは返されたObservableをsubscribe()して後続につなぐ。maxConcurrentに1を指定すると、同時に1リクエストだけ走るようになります。複数回subscribeされても1つの結果を全員に配布する。replay(1)はsubscribe()された時に最新の1つのonNext()を渡す。refCount()は誰か1人でもsubscribe()してる時に上流の処理を実行する。回数分だけ(無駄に)並列にリクエストが走ってしまいます。Observableの概念で後述します。...}Fragment側に一切複雑な処理を書くことなく、リアクティブ感にあふれる、変更が勝手に反映される実装が書けました・・!※誰からでもpostできるようなイベントはEventBusを、特定の相手のイベントを見たい場合はObservableを、といった使い分けが(どこかで)提案されていました。結局使ってないですそれなりに調べないとわからなかったことまずは実装例を載せてみたのですが、リスト操作以降は理解しておくべき事柄が増加します。↑+Subject+できればBackpressureこれ以降は正直最初っから理解しようとするとしんどい領域かと思われるので、これらの扱いが簡単なリスト操作から始めたほうがよいと思います。非同期処理やってみるぜ・・!と思ったら、下記を順番にご覧ください。unsubscribe()をするのが面倒、かつそれを何とかするRxAndroidは今使えない追記:subscribe/unsbuscribeライフサイクルに関しては別のQiita記事(詳解RxJava)に今の見解を書きました追記:RxLifecycleが元のRxAndroidのunsubscribe周りに近いですが、上記の記事のようにonStart()だけで十分だなぁと思ったので使っていません。subscribe()で返却されるSubscriptionをunsubscribe()しないと、AsyncTaskのようにタスク完了までActivity/Fragmentがリークしてしまいます(一応onComplete()もしくはonError()の際に自動でunsubscribe()されます)。Jake氏となっている通り、今あまり使えない感じです。学習コストを抑えるためもかねて、onCreate()などのメソッドでsubscribeとunsubscribeを自分で叩くことにしました。自前でのSubscription管理を簡単にするには、CompositeSubscriptionにadd()して、あとで複数まとめてunsubscribe()します。...}更新処理を実行する場合など、subscribe()を何度も呼び出す場合があると思いますが、この場合タスクが完了するたびにいちいちCompositeSubscriptionからremove()しないとどんどん増えていってしまいます。そのため、subscribe()とunsubscribe()(or終了)の際に勝手にCompositeObservableのadd()とremove()を呼び出す仕組みを作りました。https://gist.github.com/ypresto/accec4409654a1830f54追記:これは必ずsubscribe()の一番直前で呼び出さないと、うまくunsubscribeされない場合があるので注意が必要です。render(result));※CompositeSubscriptionは一度unsubscribeすると再利用できないので注意。attachTo()はmCompositeObservableを作りなおすたびに呼び出す必要があります。追記:当時より理解が深まったので疑似コード付きの説明を別のQiita記事(詳解RxJava)に書きました最初に理解するのに時間がかかったものの1つが、実行スレッドを切り替えるためのSchedulerの使い方です。Schedulerを指定すると、ストリームの実行(onNext()、onComplete()、onError()などの呼び出し)のスレッドを変更することができます。呼び出した以降のストリームを別のスレッドで実行subscribeOn()は、遅いメソッドをmap()やカスタムのObservableで呼び出す場合は指定する必要があります。ソースに近いsubscribeOn()だけが有効です。何度も叩くとスレッド生成分無駄とのこと。retrofitは最初からスレッドが決まっているようで効果が無いみたいです。Threadでしか呼べないので、非同期のリクエストなどを叩いた際はsubscribe()の直前で必ず呼び出す必要があります。今のところRxAndroidのAndroidSchedulers.mainThread()を参考に再実装するのが良さそうだと思います。RxAndroidの1.0はAndroidSchedulersの実装だけを含むライブラリになったので、これを使うのが最適です。ドキュメントに記載のない限り、subscribeOn()の指定がなければsubscribe()したスレッドで実行されます。※Observableを返却できるflatMap()を使った際がややこしいですが、返したObservableでsubscribeOn()を叩いていると、onNext()を呼んだObservableに対応するスレッドで続きの処理も呼ばれるそうです。追記:下記の実装を(特に、遅い処理で)すると、subscriberがリークして、結果Activityとかがleakしてしまいます。詳しくは別のQiita記事に書く予定です最もシンプルなやり方をすると下記のようになります。Classやlambda、もしくはそれをラップしたオブジェクト)への参照が維持されるため、AsyncTaskと同様のリークが発生します。AbstractOnSubscribeを使うと最初からunsubscribe()によるキャンセルなどがサポートされた状態になります。}})).subscribeOn(Schedulers.io());※rxjava-async-utilを使うとAsyncObservable.start()を使用することもできますが、値がキャッシュされたりするので今回は見送りました。Observableがとっつきにくい問題もう1つとっつきにくかったものが、HotとColdという概念があるということです。render(users));最初はheavyMethod()の呼び出しは1回で済むと思ってしまいました。Observableは非同期なので、mObservableにはいかなる結果もキャッシュされておらず、heavyMethod()は3回呼ばれてしまいます。変更通知の実装例でshare()(publish().refCount()と同義)の呼び出しをしていたのは、分岐の根っこにして処理結果を共有するためです。Observableと呼びます。RxのHotとColdについてにて詳しく解説されています。Observableを、全員のsubscribe()が終わるまで遅らせるための仕組みで、refCount()を使うと誰かがsubscribe()している時だけ開始された状態になります。下記のようなものがあります。Observableは開始されるとその手前までのストリームを代表してsubscribe()している状態になります。Observableに自分で値を流すにはSubjectを使用するObservable.from(List)だと最初に決めた値しか流すことができません。後から値を流したい場合はSubjectを使用しonNext()を呼びます。SubjectはObservableであると同時にコールバック(Subscriber)でもあります。PublishSubjectPublishSubjectのonNext()を呼び出した時にsubscribe()したコールバックに同じ値を渡します通知(イベント)の実装に便利BehaviorSubjectsubscribe()の際に最後にonNext()された値を流し、以降onNext()されるたびに値を流す変更が起きる値を表現するのに便利Backpressureという概念(追記)Backpressureの仕組みについては別の記事を書きました。こちらも合わせてどうぞ。Backpressureについてはまだ勉強中です。ReactiveXのintroにも記載があるように、Iterableはpull(next()で取ってくる)、Observableはpush(onNext()で渡ってくる)です。pushの場合、自分でpullしている場合に比べ、連打されて処理が追いつかない時の制御が難しくなります。これを制御するために、あとどのくらいならonNext()が叩かれても大丈夫かをソース側の方に伝える仕組みです。https://github.com/ReactiveX/RxJava/wiki/BackpressureUIで1文字入力ごとに処理呼び出す場合などに役立ちそうですが、そのような場合一定時間内のイベント数を制御するOperatorを使ってみてくださいと書かれています。StudioとXcodeを同時起動しながらAtomでRubyとかES6とか書きつつgitでmergeしてpushする日々。