この記事は、Florian RapplとMoritzKr?gerによってレビューされました。 SetePointのコンテンツを完璧にしてくれたSetePointのすべてのピアレビューアに感謝します！

機(jī)能的反応性プログラミング（FRP）への関心が高まるにつれて、RXJSはこのパラダイムで最も人気のあるJavaScriptライブラリの1つになりました。この記事では、RXJSのトップ10の必見関數(shù)を調(diào)べます。

注：この記事では、RXJSの基本に精通していると想定しています。これは、「RXJSとの機(jī)能的な反応性プログラミングから始まる」という記事で説明されています。

キーポイント

• rxjsは、機(jī)能的なリアクティブプログラミング（FRP）を促進(jìn)するために、時(shí)間の経過とともに充填された配列と同様の観測(cè)可能なオブジェクトを活用し、アプリケーションでより宣言的で強(qiáng)力なエラー処理を可能にします。
• map()、filter()、reduce()、take()などのRXJの単純なストリームのコア操作は、ミラーアレイ操作ですが、時(shí)間の経過とともに値を発する數(shù)値のストリームに適用されます。
• flatMap()やswitch()などの特別な関數(shù)は、複雑なデータ構(gòu)造を処理し、複數(shù)のストリームを個(gè)別に管理するために重要であり、高度なリアクティブプログラミングタスクに重要です。
• 、concat()、merge()などの演算子を使用して、複數(shù)のストリームを効果的に組み合わせることができ、それぞれがストリーム管理とデータの同期に異なる役割を果たします。 combineLatest()
• 関數(shù)は、フロー制御とリソース管理におけるRXJの柔軟性を示す外部條件に基づくメカニズムを提供します。 takeUntil()

リアクティブプログラミング

Reactiveプログラミングは、観察可能なオブジェクトと呼ばれるデータストリームを基本的なプログラミングユニットとして取得するプログラミングパラダイムです。

ストリーム - またはRXJS Jargonの観察可能なオブジェクト - イベントリスナーのように：どちらも何かが起こるのを待ち、それが起こったときにあなたに通知します。 OnClickリスナーから取得した一連の非同期通知は、データフローの完璧な例です。

言い換えれば、観察可能なオブジェクトは、時(shí)間の経過とともに充填された配列にすぎません。

この配列の要素は、ファイルシステム、DOMイベント、API呼び出し、さらには配列などの変換された同期データなど、ほぼどこからでも屆きます。基本的に、リアクティブプログラミングは、観察可能なオブジェクトをプログラムの構(gòu)成要素として使用することに過ぎません。

配列との関係

明示的に変更されていない限り、コンテンツが最終的なため、

配列は簡(jiǎn)単です。この意味で、配列には本質(zhì)的な時(shí)間性はありません。

一方、観察可能なオブジェクトは時(shí)間によって定義されます。せいぜい、ストリームがこれまでに受信していることを知ることができます[1、2、3]。それを決定するのは、あなたがあなたのプログラムではなくデータソースであるかどうか、あなたがそれを決定するのはデータソースであるかどうかはわかりません。

ストリームと配列の関係は非常に深いため、ほとんどの反応性拡張は、リスト操作がコアである機(jī)能プログラミングの世界に由來します。

rxjs

に精通しています

一般的なTo Doアプリを検討してください。 rxjsを使用してユーザーの未完成のタスクの名前を表示する方法の質(zhì)問を見てみましょう：

const task_stream =
  // 創(chuàng)建所有數(shù)據(jù)庫(kù)中任務(wù)的流
  getTasks().
    // 只獲取此用戶的任務(wù)
    filter((task) => task.user_id == user_id).
    // 獲取未完成的任務(wù)
    filter((task) => !task.completed).
    // 只獲取任務(wù)名稱
    map((task) => task.name)

/* 任務(wù)如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

これまでのところ、これは単なる配列拡張機(jī)能ですが、リアクティブプログラミングの機(jī)能スタイルを示しています。

宣言的な性質(zhì)は、より複雑な「現(xiàn)実世界」関數(shù)を追加することで明らかになります。必要だと仮定します：

• 完成または未完成のタスクを表示するというユーザーの選択に応じてリクエストを開始します
ユーザーが選択をすばやく変更したときに、帯域幅を無駄にすることを避けるために、2番目のリクエストを最後の選択に送信します
• 最大3つのリクエストを再試行します
サーバーが前回とは異なる応答を送信した場(chǎng)合にのみ、ビューを塗り直します。
steply分解：

const task_stream =
  parameter_stream.
    debounce(1000).
    map((parameter) => {
      getTasks().
        retry(3).
        filter((task) => task.user_id === user_id).
        filter((task) => task.completed === parameter).
        map((task)    => task.name)
    }).
    flatMap(Rx.Observable.from).
    distinctUntilChanged().
    update()

parameter_streamは、ユーザーが完了したか、未完成のタスクをパラメーターに保存したいかを示します

debounce（）は、最後のボタンクリックにのみ焦點(diǎn)を合わせてください

gettasks（）の周りの部分は以前と同じです
• distictInuntilChanged（）は、前回とは異なるサーバーの応答のみに従うことを保証します
update（）は、UIを更新して、サーバーから得られるものを反映する責(zé)任があります。
• デバウンス、再試行、および「異なるまで変更されるまで」のロジックを命令すると、コールバックベースのスタイルが効果的ですが、壊れやすく複雑です。
• 重要なのは、RXJを使用したプログラミングが許可することです
• 宣言プログラム
スケーラブルシステム

シンプルで直接的で強(qiáng)力なエラー処理。

RXJSのトップ10の必須関數(shù)を閲覧するプロセスでは、上記の例でそれぞれの関數(shù)に遭遇します。

単純なフロー操作
単純なストリーム（文字列などの単純な値を発するストリーム）の基本的な関數(shù)には次のものがあります。
map（）
filter（）

reduce（）

take（） / takhlie（）

take（）とtakhlime（）とは別に、これらはJavaScriptの高次配列関數(shù)に似ています。

これらの関數(shù)を適用します。問題を解決します。データベース內(nèi)のすべてのユーザーを.comまたは.org Webサイトで見つけて、Webサイト名の平均長(zhǎng)さを計(jì)算します。
jsonplaceholderは、ユーザーのソースとして機(jī)能します。これは、使用するユーザーデータのJSON表現(xiàn)です。
1を使用してデータを変換します
• ObservableオブジェクトにMap（）を使用することは、配列で使用するのと同じです。それ：
パラメーターとしてコールバックを受け入れます

呼び出す配列の各要素で実行します

元の配列の各要素が、コールバックによって生成された結(jié)果に置き換えられる新しい配列を返します。

観察可能なオブジェクトでmap（）を使用する場(chǎng)合の唯一の違いは次のとおりです。

新しい配列ではなく、新しい観測(cè)可能なオブジェクトを返します

Observableオブジェクトがすべてのプロジェクトをすぐにではなく、新しいプロジェクトを発行したときに実行されます。

Map（）を使用して、ユーザーデータストリームをWebサイト名のみを含むリストに変換できます。

const task_stream =
  // 創(chuàng)建所有數(shù)據(jù)庫(kù)中任務(wù)的流
  getTasks().
    // 只獲取此用戶的任務(wù)
    filter((task) => task.user_id == user_id).
    // 獲取未完成的任務(wù)
    filter((task) => !task.completed).
    // 只獲取任務(wù)名稱
    map((task) => task.name)

/* 任務(wù)如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

ここでは、MAPを使用して、各ユーザーのWebサイトに著信ストリーム內(nèi)の各ユーザーオブジェクトを置き換えます。

また、

rxjsでは、map（）をselect（）として呼び出すこともできます。どちらの名前も同じ関數(shù)を指します。

2

map（）のように、filter（）は、配列と同じオブジェクトでほぼ同じ役割を果たします。 .NETまたは.orgのWebサイトアドレスを持つすべてのユーザーを見つけるには、これを書くことができます：

const task_stream =
  parameter_stream.
    debounce(1000).
    map((parameter) => {
      getTasks().
        retry(3).
        filter((task) => task.user_id === user_id).
        filter((task) => task.completed === parameter).
        map((task)    => task.name)
    }).
    flatMap(Rx.Observable.from).
    distinctUntilChanged().
    update()

これにより、Webサイトが「ネット」または「組織」で終了するユーザーのみが選択されます。

filter（）には、where（）もあります。

3を使用して結(jié)果を収集します

reduce（）を使用すると、すべての単一値を使用して、単一の結(jié)果に変換できます。

resid（）は、多くの場(chǎng)合、最も混亂する基本リスト操作です。これは、filter（）またはmap（）とは異なり、その動(dòng)作は使用によって異なるためです。

通常、reduce（）は値のコレクションを取り、それらを単一のデータポイントに変換します。この例では、Webサイト名のStreamを提供し、Rediving（）を使用して、そのストリームを、見つけたWebサイトの數(shù)とその名前の長(zhǎng)さの合計(jì)を計(jì)算するオブジェクトに変換します。

ここでは、ストリームを単一のオブジェクトに簡(jiǎn)素化します。

source.
  map((user) => user.website)

見たサイトの數(shù) すべての名前の総長(zhǎng)さ

。

resid（）は、ソースの観察可能なオブジェクトが完了したときにのみ結(jié)果を返すことを覚えておいてください。ストリームが新しいアイテムを受信するたびにアキュムレータのステータスを知りたい場(chǎng)合は、代わりにscan（）を使用してください。
4を使用して結(jié)果を制限します

take（）およびtakhlie（）は、単純なストリームの基本的な関數(shù)を補(bǔ)完します。

（n）ストリームからn値を読み取り、登録解除します。

scan（）を使用して、Webサイトを受信するたびにオブジェクトを発し、最初の2つの値のみを使用できます。

rxjsは、特定のブールテストが確立される前に値を取得できるTakewhile（）も提供します。 Takewhile（）を使用して、上記のストリームを次のように記述できます。

高次フロー操作

source.
  map((user) => user.website).
  filter((website) => (website.endsWith('net') || website.endsWith('org'));
})

これらの関數(shù)は、配列ではなく観測(cè)可能なオブジェクトで動(dòng)作することを除いて、馴染みのあるリスト操作とほぼ同じです。

source.
  map((user) => user.website).
  filter((website) => (website.endsWith('net') || website.endsWith('org'))).
  reduce((data, website) => {
    return {
      count       : data.count += 1,
      name_length : data.name_length += website.length
    }
  }, { count : 0, name_length : 0 })

"[i]アレイ＃を使用して配列に対してプログラムする方法を知っている場(chǎng)合、rxjsの使用方法を既に知っています！」

配列には、単純な値（配列やオブジェクトなど）よりも複雑なデータを含めることができるように、観察可能なオブジェクトは、Promiseやその他の観察可能なオブジェクトなどの高次データを発することもできます。これは、よりプロフェッショナルなツールが作用する場(chǎng)所です。

5を使用して移行を押します

…実際、私たちはすでにそれを使用しています！

ソースストリームを定義するとき、frompromise（）とflatmap（）を呼び出します：

これは、3つの新しいメカニズムを使用します：

frompromise;

rx.observable.from;

フラットマップ。

source.
  map((user) => user.website).
  filter((website) => (website.endsWith('net') || website.endsWith('org'))).
  scan((data, website) => {
      return {
        count       : data.count += 1,
        name_length : data.name_length += website.length
      }
    }, { count : 0, name_length : 0 }).
  take(2);

Promiseの観察可能なオブジェクト

約束は、非同期に取得する?yún)g一の將來の価値を表しています。たとえば、サーバーへの呼び出しの結(jié)果です。

約束の決定的な特徴は、將來の価値のみを表すことです。複數(shù)の非同期データを返すことはできません。

これは、rx.observable.frompromise（）を使用すると、単一の値を発する観察可能なオブジェクトを取得することを意味します。

約束する値

約束の拒否価値。
Promiseが文字列または番號(hào)を返した場(chǎng)合、特別なことをする必要はありません。ただし、配列を返す場(chǎng)合（これは私たちの場(chǎng)合です）、配列自體を単一の値としてではなく、配列の內(nèi)容を発する観測(cè)可能なオブジェクトを作成することを好みます。
6を使用します
このプロセスはフラットニングと呼ばれ、flatmap（）プロセスです。多くの過負(fù)荷がありますが、最も単純で最も一般的に使用される過負(fù)荷のみを使用します。
flatmap（）を使用する場(chǎng)合、we：

単一の価値の解像度または約束の拒否を発行する観察可能なオブジェクトにflatmap（）を呼び出します

関數(shù)を渡して、新しい観測(cè)可能なオブジェクトを作成します。
この例では、rx.observable.from（）を渡します。これは、配列の値からシーケンスを作成します。
これは、短い序文のコードをカバーします：
rxjsは、flatmap（）：selectmany（）のエイリアスも提供します。

複數(shù)のストリームを組み合わせた

const task_stream =
  // 創(chuàng)建所有數(shù)據(jù)庫(kù)中任務(wù)的流
  getTasks().
    // 只獲取此用戶的任務(wù)
    filter((task) => task.user_id == user_id).
    // 獲取未完成的任務(wù)
    filter((task) => !task.completed).
    // 只獲取任務(wù)名稱
    map((task) => task.name)

/* 任務(wù)如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

通常、結(jié)合する必要がある複數(shù)のストリームがあります。ストリームを組み合わせる方法はたくさんありますが、他のストリームよりも頻繁に表示されるものもあります。

const task_stream =
  parameter_stream.
    debounce(1000).
    map((parameter) => {
      getTasks().
        retry(3).
        filter((task) => task.user_id === user_id).
        filter((task) => task.completed === parameter).
        map((task)    => task.name)
    }).
    flatMap(Rx.Observable.from).
    distinctUntilChanged().
    update()

7 concat（）を使用して、ストリームを組み合わせます
接続とマージは、ストリームを結(jié)合する最も一般的な2つの方法です。
接続は、終了するまで最初のストリームの値を放出し、2番目のストリームの値を放出することにより、新しいストリームを作成します。
MERGEは、アクティブストリームの値を放出することにより、複數(shù)のストリームから新しいストリームを作成します
Facebookメッセンジャーで同時(shí)に二人と話すことを考えてください。 concat（）は、両當(dāng)事者からメッセージを受け取るが、別の人に返信する前にある人との會(huì)話を完了する狀況です。 Merge（）は、グループチャットを作成し、同時(shí)に2つのメッセージストリームを受信するようなものです。

concat（）ストリームは、最初にSource1のすべての値を印刷し、Source1が終了した後にのみSource2の値の印刷を開始します。

merge（）ストリームは、受信した順序に従ってsource1とsource2の値を印刷します。2番目のストリームの値を発する前に、最初のストリームが完了するのを待ちません。

8を使用します

通常、観察可能なオブジェクトを放出するが、ソースからの最新の排出量にのみ焦點(diǎn)を當(dāng)てる観察可能なオブジェクトを聞きたいと考えています。

source.
  map((user) => user.website)

Facebook Messengerの類推をさらに拡張するために、Switch（）はあなたが…まあ、誰(shuí)が現(xiàn)在メッセージを送信しているかに基づいて返信する人を切り替えます。

この目的のために、RXJSはスイッチを提供します。

ユーザーインターフェイスは、switch（）にいくつかの適切なユースケースを提供します。ユーザーが検索したいものを選択するたびにアプリケーションがリクエストを行うと、最新選択の結(jié)果を確認(rèn)したいだけであると仮定できます。したがって、Switch（）を使用して、最新の選択結(jié)果のみをリッスンします。

ちなみに、

帯域幅を無駄にしないようにし、ユーザーが最後に毎秒行うときのみサーバーへのアクセスを選択するようにする必要があります。これに使用する関數(shù)は、debounce（）

と呼ばれます

反対方向に進(jìn)み、最初の選択肢に従う場(chǎng)合のみを使用する場(chǎng)合は、Throttle（）を使用できます。同じAPIを持っていますが、反対に動(dòng)作します。

9

特定のIDを持つユーザーが投稿またはユーザーを検索できるようにしたい場(chǎng)合はどうなりますか？
デモンストレーションでは、別のドロップダウンメニューを作成し、ユーザーが取得するアイテムのIDを選択できるようにします。
2つの狀況があります。ユーザーの場(chǎng)合：
任意の選択肢を変更します
1. 2つのオプションを変更します。
combineLatest（）
を使用して、あらゆるレベルの変更に応答します

最初のケースでは、以下を使用してネットワークリクエストを開始するストリームを作成する必要があります。

ユーザーのエンドポイントの選択

最近ユーザーが選択したID。
…そして、ユーザーが選択を更新するときにこれを行います。
これは、combineLatest（）のためです

ストリームが値を放出するたびに、combineLatest（）はエミスト値を取得し、他のストリームから放出された最後のアイテムとペアをペアにペアにペアに合わせます。

これは、チャートで視覚化するのが簡(jiǎn)単です：

const task_stream =
  // 創(chuàng)建所有數(shù)據(jù)庫(kù)中任務(wù)的流
  getTasks().
    // 只獲取此用戶的任務(wù)
    filter((task) => task.user_id == user_id).
    // 獲取未完成的任務(wù)
    filter((task) => !task.completed).
    // 只獲取任務(wù)名稱
    map((task) => task.name)

/* 任務(wù)如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

zipを使用して、2つのストリームの変更のみに応答します

では、ユーザーがIDとエンドポイントフィールドの選択を更新するまで待って、combinElatest（）をzip（）に置き換えます。

const task_stream =
  parameter_stream.
    debounce(1000).
    map((parameter) => {
      getTasks().
        retry(3).
        filter((task) => task.user_id === user_id).
        filter((task) => task.completed === parameter).
        map((task)    => task.name)
    }).
    flatMap(Rx.Observable.from).
    distinctUntilChanged().
    update()

繰り返しますが、これはチャートで理解しやすいです：

combineLatest（）とは異なり、zip（）は、両方の観察可能なオブジェクトが更新された値の配列を送信する前に新しいコンテンツを発するまで待ちます。

10

最後に、Takeuntil（）を使用すると、2番目のストリームが値を発し始めるまで、最初のストリームを聞くことができます。

source.
  map((user) => user.website)

これは、ストリームを調(diào)整する必要があるが、それらを結(jié)合する必要がない場(chǎng)合に役立ちます。
要約
アレイに時(shí)間寸法を追加するだけで、プログラムについての新しい思考への扉が開きます。

source.
  map((user) => user.website).
  filter((website) => (website.endsWith('net') || website.endsWith('org'));
})

rxjsはここで見られるものよりもはるかに多くですが、それは長(zhǎng)い道のりを歩むのに十分です。

rxjs liteから始めて、ドキュメントを參照して時(shí)間をかけて行う準(zhǔn)備をしてください。あなたがそれを知る前に、すべてがストリームのように見えます…すべてがそうだからです。

RXJS関數(shù)（FAQ）

に関する FAQ

RXJSと従來のJavaScriptの主な違いは何ですか？
RXJSは、非同期またはコールバックベースのコードの組み合わせを簡(jiǎn)素化するために、観察可能なオブジェクトを使用したリアクティブプログラミングライブラリです。これは、より必須のプログラミングスタイルで従來のJavaScriptを使用することと比較されます。重要な違いは、データの処理方法です。RXJSはデータをストリームとして扱います。これは、さまざまな演算子を使用して動(dòng)作および変換できます。

RXJSで観測(cè)可能なオブジェクトを作成する方法は？

RXJSでは、新しいObservable（）コンストラクターを使用して観測(cè)可能なオブジェクトを作成できます。このコンストラクターは、サブスクライバー関數(shù)と呼ばれる引數(shù)として関數(shù)を採(cǎi)用します。これは、最初に観測(cè)可能なオブジェクトにサブスクライブしたときに実行されます?；镜膜世洗韦韦趣辘扦?
RXJSの主要な演算子とそれらの仕組みは何ですか？

const task_stream =
  // 創(chuàng)建所有數(shù)據(jù)庫(kù)中任務(wù)的流
  getTasks().
    // 只獲取此用戶的任務(wù)
    filter((task) => task.user_id == user_id).
    // 獲取未完成的任務(wù)
    filter((task) => !task.completed).
    // 只獲取任務(wù)名稱
    map((task) => task.name)

/* 任務(wù)如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

RXJSには、観察可能なオブジェクトとオブザーバー間でデータがどのように流れるかを制御するために使用できる幅広い演算子があります。主な演算子には、map（）、filter（）、resid（）、merge（）、concat（）が含まれます。これらの各オペレーターは、データの変換、特定の値のフィルタリング、複數(shù)のストリームの組み合わせなど、さまざまな方法でデータのストリームを操作します。

RXJのエラーを処理する方法は？

rxjsは、catcherror（）、retry（）、retry（）などのエラーを処理するいくつかの演算子を提供します。 CatcherRor（）演算子は、観察可能なストリームでエラーをキャッチし、新しい観測(cè)可能なオブジェクトを返したり、エラーを投げたりするために使用されます。 retry（）演算子を使用して、エラーが発生した場(chǎng)合に観測(cè)可能なオブジェクトに再登録できます。 retry（）演算子は似ていますが、いつ再試行するかをより制御します。

RXJの観測(cè)可能なオブジェクトのサブスクリプションをキャンセルする方法は？

Observableをサブスクライブすると、unsubscribe（）メソッドを備えたサブスクリプションを受け取ります。この方法を呼び出して、観察可能なオブジェクトの実行をキャンセルし、使用されているリソースをクリーンアップできます。例は次のとおりです。

中程度の加熱可能なオブジェクトと冷たい観測(cè)可能なオブジェクトの違いは何ですか？

const task_stream =
  parameter_stream.
    debounce(1000).
    map((parameter) => {
      getTasks().
        retry(3).
        filter((task) => task.user_id === user_id).
        filter((task) => task.completed === parameter).
        map((task)    => task.name)
    }).
    flatMap(Rx.Observable.from).
    distinctUntilChanged().
    update()

rxjsでは、観察可能なオブジェクトはホットまたはコールドになる場(chǎng)合があります。冷たい観測(cè)値は購(gòu)読時(shí)に実行を開始しますが、ホットオブザーバブルは購(gòu)読する前でさえ値を生成します。言い換えれば、冷たい観察可能なオブジェクトは不活性ですが、熱い観察可能なオブジェクトは不活性です。

RXJで複數(shù)の観測(cè)可能なオブジェクトを組み合わせる方法は？

rxjsは、merge（）、concat（）、combinelatest（）、zip（）などの複數(shù)の観測(cè)可能なオブジェクトを組み合わせた複數(shù)の演算子を提供します。これらの各演算子は、特定のニーズに応じて、さまざまな方法でデータストリームを組み合わせています。

RXJSのテーマの目的は何ですか？

RXJSのトピックは、複數(shù)のオブザーバーに値のマルチキャストを可能にする特別なタイプの観測(cè)可能なオブジェクトです。通常の観察可能なオブジェクトとは異なり、トピックは多くのリスナーのレジストリを維持します。

angularでrxjsを使用する方法は？

AngularはRXJを組み込んでサポートし、さまざまな機(jī)能に內(nèi)部的に使用します。また、獨(dú)自のコードでRXJを使用して非同期操作を処理し、自動(dòng)完成、除jitter、スロットリング、ポーリングなどの機(jī)能を?qū)g裝することもできます。

RXJの一般的なユースケースは何ですか？

RXJは、非同期データが必要なさまざまなシナリオで使用できます。いくつかの一般的なユースケースには、ユーザー入力の処理、HTTPリクエストの作成、WebSocketの使用、アニメーションの処理が含まれます。

以上が10の知識(shí)が必要なRXJSは、例を使用して機(jī)能しますの詳細(xì)內(nèi)容です。詳細(xì)については、PHP 中國(guó)語(yǔ) Web サイトの他の関連記事を參照してください。