JavaScript、Angular、RxJSの達人になるための秘訣！flatMap、mergeMap、switchMap、concatMapを使いこなそう！

2024-05-15

JavaScript、Angular、RxJSにおけるflatMap、mergeMap、switchMap、concatMapの分かりやすい解説

各関数の詳細

flatMap（別名：mergeMap）

1つの入力Observableを複数のObservableに分割し、それらを平坦化して1つの出力Observableに統合します。
複数のObservableを同時に処理し、出力される順番は非同期処理の完了順になります。
複数のHTTPリクエストの同時実行や、データ構造の再構成などに適しています。

例：

const source = Observable.from([1, 2, 3]);
const result = source.pipe(
  flatMap(x => Observable.of(x * 2))
);
result.subscribe(x => console.log(x)); // 出力: 2, 4, 6

switchMap

1つの入力Observableを最新の状態のObservableに置き換え、そのObservableの値のみを出力します。
前回のObservableが完了する前に新しいObservableに切り替わるため、常に最新の値のみを処理します。
ユーザ入力へのリアルタイムな反応や、フォームデータの検証などに適しています。

const source = Observable.from(['A', 'B', 'C']);
const result = source.pipe(
  switchMap(x => Observable.of(x + '!!'))
);
result.subscribe(x => console.log(x)); // 出力: B!!, C!! (A!!は出力されない)

concatMap

処理順序は入力Observableの順序と一致します。
複数の処理を順番に実行したい場合や、処理結果を累積的に処理したい場合などに適しています。

const source = Observable.from([1, 2, 3]);
const result = source.pipe(
  concatMap(x => Observable.of(x * 2))
);
result.subscribe(x => console.log(x)); // 出力: 2, 4, 6

exhaustMap

switchMapと似ていますが、1つの入力Observableに対して同時に処理できるObservableを1つに制限します。
複数のHTTPリクエストを順番に実行したい場合や、競合状態を回避したい場合などに適しています。

const source = Observable.from(['A', 'B', 'C']);
const result = source.pipe(
  exhaustMap(x => Observable.of(x + '!!').delay(1000))
);
result.subscribe(x => console.log(x)); // 出力: A!!, B!!, C!! (順番に処理される)

flatMap、mergeMap、switchMap、concatMapは、それぞれ異なる動作を持つため、状況に応じて適切な関数を選択することが重要です。

複数のObservableを同時に処理したい場合はflatMap
常に最新の値のみを処理したい場合はswitchMap
処理順序を保ちたい場合はconcatMap
競合状態を回避したい場合はexhaustMap

これらの関数を理解することで、RxJSをより効果的に活用し、複雑な非同期処理を効率的に処理することができます。

補足

RxJSは、ReactiveXという非同期処理ライブラリを基盤としています。
Angularは、RxJSを組み込んでおり、非同期処理を扱う際に活用できます。

flatMap（別名：mergeMap）

const source = Observable.from([1, 2, 3]);
const result = source.pipe(
  flatMap(x => Observable.of(x * 2))
);
result.subscribe(x => console.log(x)); // 出力: 2, 4, 6

switchMap

const source = Observable.from(['A', 'B', 'C']);
const result = source.pipe(
  switchMap(x => Observable.of(x + '!!'))
);
result.subscribe(x => console.log(x)); // 出力: B!!, C!! (A!!は出力されない)

concatMap

const source = Observable.from([1, 2, 3]);
const result = source.pipe(
  concatMap(x => Observable.of(x * 2))
);
result.subscribe(x => console.log(x)); // 出力: 2, 4, 6

exhaustMap

const source = Observable.from(['A', 'B', 'C']);
const result = source.pipe(
  exhaustMap(x => Observable.of(x + '!!').delay(1000))
);
result.subscribe(x => console.log(x)); // 出力: A!!, B!!, C!! (順番に処理される)

解説

この例では、source Observableからemitされる各値 x を、Observable.of(x * 2) に変換して、出力Observableに統合しています。
Observable.of(x * 2) は、x の2倍の値をemitするObservableです。
そのため、出力Observableは 2, 4, 6 の値をemitします。

switchMap は、前のObservableが完了する前に新しいObservableに切り替わるため、A!! は出力されず、B!! と C!! のみが出力されます。

exhaustMap は、1つの入力Observableに対して同時に処理できるObservableを1つに制限するため、A!!、B!!、C!! の値が順番に1秒間隔でemitされます。

これらのサンプルコードは、それぞれの関数の基本的な動作を理解するためのものです。具体的な用途に合わせて、様々な処理を組み合わせることもできます。

RxJSは、奥深いライブラリであり、様々な使い方が可能です。ぜひ、公式ドキュメントやチュートリアルなどを参考に、RxJSを深く理解し、様々な問題を解決するのに役立ててください。

RxJSにおけるflatMap、mergeMap、switchMap、concatMap以外の選択肢

combineLatest

複数のObservableを同時に監視し、最新の値を組み合わせた結果をemitする関数です。
複数の入力ソースに基づいて動的に処理を変化させたい場合などに適しています。

const source1 = Observable.from([1, 2, 3]);
const source2 = Observable.from(['A', 'B', 'C']);
const result = source1.pipe(
  combineLatest(source2, (x, y) => x + ' ' + y)
);
result.subscribe(x => console.log(x)); // 出力: 1 A, 2 B, 3 C

withLatestFrom

主軸となるObservableの値が更新されるたびに、最新の値と別のObservableからの値を組み合わせた処理を実行したい場合などに適しています。

const source1 = Observable.from([1, 2, 3]).pipe(interval(1000));
const source2 = Observable.from(['A', 'B', 'C']).pipe(interval(2000));
const result = source1.pipe(
  withLatestFrom(source2, (x, y) => x + ' ' + y)
);
result.subscribe(x => console.log(x)); // 出力: 1 A, 2 B, 3 C (1秒後にA, 2秒後にB, 3秒後にCが出力される)

scan

入力Observableの値を累積的に処理し、その結果をemitする関数です。
処理の履歴を保持したい場合や、最終的な結果のみが必要な場合などに適しています。

const source = Observable.from([1, 2, 3]);
const result = source.pipe(
  scan((acc, x) => acc + x, 0)
);
result.subscribe(x => console.log(x)); // 出力: 1, 3, 6

groupBy

入力Observableの値をグループ分けし、それぞれのグループのObservableをemitする関数です。
異なる種類のデータ処理を並行に行いたい場合や、データ構造を再構成したい場合などに適しています。

const source = Observable.from(['A1', 'B2', 'C3', 'A4', 'B5']);
const result = source.pipe(
  groupBy(x => x[0]),
  mergeMap(group => group.pipe(scan((acc, x) => acc + x, '')))
);
result.subscribe(x => console.log(x)); // 出力: A1A4, B2B5, C3

window

処理を分割して独立させたい場合や、データ量を制限したい場合などに適しています。

const source = Observable.from([1, 2, 3, 4, 5]);
const result = source.pipe(
  window(2),
  mergeMap(group => group.pipe(reduce((acc, x) => acc + x, 0)))
);
result.subscribe(x => console.log(x)); // 出力: 3, 7