はじめに
クリックイベント、入力イベント、タイマー、HTTPレスポンス。Webアプリには非同期に発生するイベントがいくつも登場しますが、それぞれをコールバックやPromiseで個別に処理していると、コードはあっという間に絡み合ってしまいます。「このイベントが起きたら、あのイベントと合わせて、こう処理したい」という要求が増えるほど、if文とフラグ変数だらけの状態管理に陥りがちです。
RxJSは、こうしたバラバラな非同期イベントを「Observable」という1本のストリームとして扱い、演算子を組み合わせるだけで合成・変換・制御できるようにするライブラリです。DOMイベントもタイマーもHTTPレスポンスも、すべて同じ作法で扱えるようになるので、非同期処理特有の複雑さがぐっと減ります。
とはいえ、説明を読むより実際に手を動かした方が理解は早いと思います。ボタンのクリックをストリームとして扱うサンプルを用意したので、先に挙動を見たい方はこちらからどうぞ。
RxJSとは
RxJS（Reactive Extensions for JavaScript）は、Observableを中心にした非同期・イベントベースのプログラムを構築するためのライブラリです。.NETやJavaなど各言語向けに存在する「ReactiveX」ファミリーのJavaScript実装にあたり、AngularをはじめとするフレームワークでもHTTP通信やフォーム制御の基盤として採用されています。
執筆時点の最新安定版はv7.8.2で、v6系からの全面的な書き直しにより、バンドルサイズの削減とパフォーマンス向上、TC39のObservable仕様提案への準拠が図られています。MITライセンスで公開されており、GitHub上でも長期にわたって活発に開発が続けられています。
主な特徴
- Observableによる統一的な表現 - DOMイベント、タイマー、HTTPレスポンスなど、性質の異なる非同期処理を同じインターフェースで扱えます
- 豊富な演算子（operators） -
map、filter、debounceTime、combineLatestなど、ストリームの変換・合成・制御を宣言的に書けます pipe()によるチェーン記法 - 複数の演算子を.pipe()で連結し、処理の流れを上から下に読める形で書けます- キャンセル可能なSubscription -
subscribe()が返すオブジェクトのunsubscribe()で、不要になった購読を明示的に止められます - スケジューラでの実行制御 - いつ・どのように値を発行するかを、スケジューラを通じて細かく制御できます
インストール
npmを使う場合は次のコマンドでインストールできます。
npm install rxjs
yarnやpnpmでも同様です。
yarn add rxjs
pnpm add rxjs
RxJSのサンプルを動かす
ボタンのクリックイベントをRxJSのfromEvent()でObservable化し、クリックのたびに値を累積していくサンプルです。DOMイベントをストリームとして扱う感覚をまず体験してみてください。
fromEvent()がDOMイベントをObservableに変換し、pipe()で連結したscan()が、直前の値を受け取りながら新しい値を返す演算子として働きます。配列のreduce()のストリーム版だとイメージすると分かりやすいです。
import { fromEvent } from 'rxjs'
import { scan } from 'rxjs'
fromEvent(button, 'click').pipe(
scan(count => count + 1, 0) // イベントが発火するたびに累積
).subscribe(count => {
countEl.textContent = String(count)
})
実際に動かせるものが下です。ボタンを何度か押して、カウントが増えていく様子を確認してください。
scan()の初期値を10にすれば10からカウントが始まりますし、間にmap(count => count * 2)を挟めば偶数だけがカウンターに表示されるようになります。RxJSでは「イベント発生」と「値の加工」を別々の演算子として分離して書けるのがポイントです。
基本的な使い方
RxJSの基本は、Observableを作り、pipe()で演算子を通し、subscribe()で購読する、という3ステップです。of()やrange()のようなファクトリ関数で値の列を作り、map()やfilter()で加工します。
import { range, filter, map } from 'rxjs'
const subscription = range(1, 10).pipe(
filter(x => x % 2 === 0), // 偶数だけを通す
map(x => x * x) // 2乗する
).subscribe({
next: value => console.log(value),
complete: () => console.log('完了しました'),
})
// 不要になったら購読を止める
subscription.unsubscribe()
subscribe()には関数を1つ渡すだけでも動きますが、上記のようにnext（値を受け取ったとき）やcomplete（ストリームが完了したとき）、error（エラー発生時）をオブジェクトで指定すると、それぞれの状態に応じた処理を書き分けられます。
実践的なユースケース
検索ワードでリストを絞り込む
入力イベントのたびにリスト全体を再計算していると、入力途中の状態でも毎回処理が走ってしまいます。RxJSのdebounceTime()とdistinctUntilChanged()を組み合わせると、「入力が一定時間止まったとき」「かつ前回と値が変わったとき」だけ処理を実行するように制御できます。
import { fromEvent } from 'rxjs'
import { map, debounceTime, distinctUntilChanged } from 'rxjs'
fromEvent(inputEl, 'input').pipe(
map(e => e.target.value),
debounceTime(200), // 200ms入力が止まるまで待つ
distinctUntilChanged() // 前回と同じ値なら発火しない
).subscribe(keyword => {
render(keyword)
})
debounceTime(200)の数値を大きくすると、入力が止まってから絞り込みが反映されるまでの間隔が長くなります。実際のアプリで検索APIを叩く場合は、この演算子を挟むだけでリクエスト回数を大幅に減らせます。
combineLatestで複数の入力を合成する
身長と体重のように、複数の入力値を組み合わせて1つの結果を出したい場面は多くあります。RxJSのcombineLatest()を使うと、複数のObservableのうち、いずれかに新しい値が来るたびに、最新の組み合わせをまとめて受け取れます。
import { combineLatest } from 'rxjs'
import { map, startWith } from 'rxjs'
combineLatest([height$, weight$]).pipe(
map(([h, w]) => (w / (h / 100) ** 2).toFixed(1))
).subscribe(bmi => {
bmiEl.textContent = bmi
})
身長・体重どちらの入力欄を変更しても、BMIの表示がその場で再計算されます。startWith()で初期値を与えているので、何も入力しなくてもページを開いた瞬間から結果が表示される点もポイントです。フォームの複数項目からリアルタイムにサマリーを出したいときに使える構成です。
retryとcatchErrorでエラーに強くする
ネットワーク越しの処理は失敗することがあります。RxJSのretry()演算子を使うと、Observableがエラーを出したときに指定回数だけ自動で再購読（再試行）できます。それでも失敗する場合はcatchError()で代替の値に差し替えられます。
import { retry, catchError, of } from 'rxjs'
flaky$.pipe(
retry(2), // 失敗したら最大2回まで再試行
catchError(err => of(`失敗: ${err.message}`)) // それでも失敗したら代替値を流す
).subscribe(result => console.log(result))
ボタンを何度か押すと、「実行開始」のログが複数回出てから「成功しました」または「最終的に失敗」に落ち着くのが分かります。retry(2)の数値を増やせば再試行の回数が増え、Math.random() < 0.6の閾値を下げれば成功しやすくなります。実際のHTTPリクエストでも、同じ演算子構成でリトライ処理を組み込めます。
まとめ
RxJSは、DOMイベントやタイマー、非同期処理をすべて「Observable」という共通の形で扱い、pipe()に演算子を並べるだけで変換・合成・制御を宣言的に書けるライブラリです。debounceTime()で入力の間引きをしたり、combineLatest()で複数の値を合成したり、retry()でエラーに強い処理を書いたりと、素のPromiseやコールバックでは煩雑になりがちな処理を、驚くほどシンプルに表現できます。
最初はObservable・演算子・Subscriptionといった独自の概念に戸惑うかもしれませんが、慣れると「非同期処理の設計図」を演算子の組み合わせとして書けるようになります。まずは今回のサンプルを書き換えながら、手元の非同期処理をRxJSで置き換えてみてはいかがでしょうか。