10の知識が必要なRXJSは、例を使用して機能します

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

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

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

キーポイント

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

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

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

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

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

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

配列との関係

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

配列は簡単です。この意味で、配列には本質的な時間性はありません。

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

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

rxjs

に精通しています

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

const task_stream =
  // 创建所有数据库中任务的流
  getTasks().
    // 只获取此用户的任务
    filter((task) => task.user_id == user_id).
    // 获取未完成的任务
    filter((task) => !task.completed).
    // 只获取任务名称
    map((task) => task.name)

/* 任务如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

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

宣言的な性質は、より複雑な「現実世界」関数を追加することで明らかになります。必要だと仮定します：

• 完成または未完成のタスクを表示するというユーザーの選択に応じてリクエストを開始します
ユーザーが選択をすばやく変更したときに、帯域幅を無駄にすることを避けるために、2番目のリクエストを最後の選択に送信します
• 最大3つのリクエストを再試行します
サーバーが前回とは異なる応答を送信した場合にのみ、ビューを塗り直します。
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（）は、最後のボタンクリックにのみ焦点を合わせてください

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

シンプルで直接的で強力なエラー処理。

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

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

reduce（）

take（） / takhlie（）

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

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

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

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

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

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

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

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

const task_stream =
  // 创建所有数据库中任务的流
  getTasks().
    // 只获取此用户的任务
    filter((task) => task.user_id == user_id).
    // 获取未完成的任务
    filter((task) => !task.completed).
    // 只获取任务名称
    map((task) => task.name)

/* 任务如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

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

また、

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

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を使用して結果を収集します

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

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

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

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

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

見たサイトの数 すべての名前の総長さ

。

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

take（）およびtakhlie（）は、単純なストリームの基本的な関数を補完します。

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

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

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

高次フロー操作

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

これらの関数は、配列ではなく観測可能なオブジェクトで動作することを除いて、馴染みのあるリスト操作とほぼ同じです。

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]アレイ＃を使用して配列に対してプログラムする方法を知っている場合、rxjsの使用方法を既に知っています！」

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

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の観察可能なオブジェクト

約束は、非同期に取得する単一の将来の価値を表しています。たとえば、サーバーへの呼び出しの結果です。

約束の決定的な特徴は、将来の価値のみを表すことです。複数の非同期データを返すことはできません。

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

約束する値

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

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

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

複数のストリームを組み合わせた

const task_stream =
  // 创建所有数据库中任务的流
  getTasks().
    // 只获取此用户的任务
    filter((task) => task.user_id == user_id).
    // 获取未完成的任务
    filter((task) => !task.completed).
    // 只获取任务名称
    map((task) => task.name)

/* 任务如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

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

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（）を使用して、ストリームを組み合わせます
接続とマージは、ストリームを結合する最も一般的な2つの方法です。
接続は、終了するまで最初のストリームの値を放出し、2番目のストリームの値を放出することにより、新しいストリームを作成します。
MERGEは、アクティブストリームの値を放出することにより、複数のストリームから新しいストリームを作成します
Facebookメッセンジャーで同時に二人と話すことを考えてください。 concat（）は、両当事者からメッセージを受け取るが、別の人に返信する前にある人との会話を完了する状況です。 Merge（）は、グループチャットを作成し、同時に2つのメッセージストリームを受信するようなものです。

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

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

8を使用します

通常、観察可能なオブジェクトを放出するが、ソースからの最新の排出量にのみ焦点を当てる観察可能なオブジェクトを聞きたいと考えています。

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

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

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

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

ちなみに、

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

と呼ばれます

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

9

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

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

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

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

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

これは、チャートで視覚化するのが簡単です：

const task_stream =
  // 创建所有数据库中任务的流
  getTasks().
    // 只获取此用户的任务
    filter((task) => task.user_id == user_id).
    // 获取未完成的任务
    filter((task) => !task.completed).
    // 只获取任务名称
    map((task) => task.name)

/* 任务如下所示：
   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)

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

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

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

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

RXJS関数（FAQ）

に関する FAQ

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

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

RXJSでは、新しいObservable（）コンストラクターを使用して観測可能なオブジェクトを作成できます。このコンストラクターは、サブスクライバー関数と呼ばれる引数として関数を採用します。これは、最初に観測可能なオブジェクトにサブスクライブしたときに実行されます。基本的な例は次のとおりです
RXJSの主要な演算子とそれらの仕組みは何ですか？

const task_stream =
  // 创建所有数据库中任务的流
  getTasks().
    // 只获取此用户的任务
    filter((task) => task.user_id == user_id).
    // 获取未完成的任务
    filter((task) => !task.completed).
    // 只获取任务名称
    map((task) => task.name)

/* 任务如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

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

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

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

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

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

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

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では、観察可能なオブジェクトはホットまたはコールドになる場合があります。冷たい観測値は購読時に実行を開始しますが、ホットオブザーバブルは購読する前でさえ値を生成します。言い換えれば、冷たい観察可能なオブジェクトは不活性ですが、熱い観察可能なオブジェクトは不活性です。

RXJで複数の観測可能なオブジェクトを組み合わせる方法は？

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

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

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

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

AngularはRXJを組み込んでサポートし、さまざまな機能に内部的に使用します。また、独自のコードでRXJを使用して非同期操作を処理し、自動完成、除jitter、スロットリング、ポーリングなどの機能を実装することもできます。

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

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

以上が10の知識が必要なRXJSは、例を使用して機能しますの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。