こんにちは、みんな。 オメリニツキー・セルゲイに連絡してください。 少し前まで、私はリアクティブ プログラミングに関するストリームをホストし、そこで JavaScript の非同期について話しました。 今日はこの資料についてまとめてみたいと思います。
しかし、本題を始める前に、導入をする必要があります。 それでは、定義から始めましょう。スタックとキューとは何ですか?
のスタック 要素が「後入れ先出し」LIFO ベースで取得されるコレクションです。
待ち行列 要素が原理 (「先入れ先出し」FIFO) に従って取得されるコレクションです。
さて、続けましょう。
JavaScript はシングルスレッドのプログラミング言語です。 これは、実行スレッドが XNUMX つと、関数が実行のためにキューに入れられるスタックが XNUMX つだけあることを意味します。 したがって、JavaScript は一度に XNUMX つの操作のみを実行できますが、他の操作は呼び出されるまでスタック上で順番を待ちます。
コールスタック は、簡単に言えば、プログラム内の私たちがいる場所に関する情報を記録するデータ構造です。 関数にジャンプすると、そのエントリがスタックの一番上にプッシュされます。 関数から戻ると、スタックから最上位の要素がポップされ、この関数を呼び出した場所に到達します。 スタックでできるのはこれだけです。 そして今、非常に興味深い質問があります。 では、JavaScript では非同期はどのように機能するのでしょうか?
実際、ブラウザには、スタックに加えて、いわゆる WebAPI を操作するための特別なキューがあります。 このキューの関数は、スタックが完全にクリアされた後にのみ順番に実行されます。 その後のみ、実行のためにキューからスタックに配置されます。 現時点でスタック上に少なくとも XNUMX つの要素がある場合、それらはスタックに追加できません。 このため、タイムアウトによる関数の呼び出しは、キューがいっぱいになっている間は関数がキューからスタックに到達できないため、時間的に不正確になることがよくあります。
次の例を見て、段階的に見てみましょう。 システム内で何が起こるか見てみましょう。
console.log('Hi');
setTimeout(function cb1() {
console.log('cb1');
}, 5000);
console.log('Bye');
1) 今のところ何も起こっていません。 ブラウザコンソールはクリーンで、コールスタックは空です。
2) 次に、コマンド console.log('Hi') がコール スタックに追加されます。
3) そしてそれは満たされる
4) その後、console.log('Hi') がコールスタックから削除されます。
5) 次に、setTimeout(function cb1() {… }) コマンドに進みます。 呼び出しスタックに追加されます。
6) setTimeout(function cb1() {… }) コマンドが実行されます。 ブラウザーは、Web API の一部であるタイマーを作成します。 カウントダウンを実行します。
7) setTimeout(function cb1() {… }) コマンドは作業を完了し、コール スタックから削除されます。
8) console.log('Bye') コマンドがコール スタックに追加されます。
9) console.log('Bye') コマンドが実行されます。
10) コマンド console.log('Bye') がコール スタックから削除されます。
11) 少なくとも 5000 ミリ秒が経過すると、タイマーが終了し、cb1 コールバックがコールバック キューに入れられます。
12) イベント ループは、コールバック キューから関数 cb1 を取得し、コール スタックにプッシュします。
13) cb1 関数が実行され、console.log('cb1') がコール スタックに追加されます。
14) console.log('cb1') コマンドが実行されます。
15) コマンド console.log('cb1') がコール スタックから削除されます。
16) 関数 cb1 が呼び出しスタックから削除されます。
ダイナミクスの例を見てみましょう。
さて、JavaScript で非同期がどのように実装されるかを調べました。 ここで、非同期コードの進化について簡単に説明しましょう。
非同期コードの進化。
a(function (resultsFromA) {
b(resultsFromA, function (resultsFromB) {
c(resultsFromB, function (resultsFromC) {
d(resultsFromC, function (resultsFromD) {
e(resultsFromD, function (resultsFromE) {
f(resultsFromE, function (resultsFromF) {
console.log(resultsFromF);
})
})
})
})
})
});JavaScript で知られている非同期プログラミングは、関数を使用してのみ実行できます。 これらは、他の変数と同様に他の関数に渡すことができます。 こうしてコールバックが誕生しました。 そして、それは悲しみ、憂鬱、悲しみに変わるまで、クールで楽しく熱狂的です。 なぜ？ はい、簡単です。
- コードの複雑さが増すと、プロジェクトはすぐにわかりにくい複数のネストされたブロック、つまり「コールバック地獄」に変わります。
- エラー処理は見落とされがちです。
- return で式を返すことはできません。
Promise の登場により、状況は少し改善されました。
new Promise(function(resolve, reject) {
setTimeout(() => resolve(1), 2000);
}).then((result) => {
alert(result);
return result + 2;
}).then((result) => {
throw new Error('FAILED HERE');
alert(result);
return result + 2;
}).then((result) => {
alert(result);
return result + 2;
}).catch((e) => {
console.log('error: ', e);
});- Promise チェーンが登場し、コードの可読性が向上しました
- エラーをインターセプトする別の方法がありました
- Promise.allを追加した並列実行
- ネストされた非同期は async/await で解決できます
しかし、その約束には限界があります。 たとえば、タンバリンを持って踊らなければ約束をキャンセルすることはできません。そして最も重要なことは、約束は XNUMX つの価値観で機能するということです。
さて、ここからは順調にリアクティブプログラミングに近づいていきます。 疲れた？ まあ、良いことは、カモメを醸造し、ブレインストーミングをし、戻って続きを読むことができるということです。 そして私は続けます。
リアクティブプログラミング - データ フローと変更の伝播に焦点を当てたプログラミング パラダイム。 データ ストリームとは何かを詳しく見てみましょう。
// Получаем ссылку на элемент
const input = ducument.querySelector('input');
const eventsArray = [];
// Пушим каждое событие в массив eventsArray
input.addEventListener('keyup',
event => eventsArray.push(event)
);入力フィールドがあると想像してみましょう。 配列を作成し、入力イベントのキーアップごとにイベントを配列に保存します。 同時に、配列が時間によってソートされていることにも注意してください。 後のイベントのインデックスは、以前のイベントのインデックスよりも大きくなります。 このような配列は単純化されたデータ フロー モデルですが、まだフローではありません。 この配列を安全にストリームと呼ぶためには、新しいデータが到着したことを何らかの方法でサブスクライバーに通知できなければなりません。 したがって、フローの定義に到達します。
データストリーム
const { interval } = Rx;
const { take } = RxOperators;
interval(1000).pipe(
take(4)
)
流れ 時間順に並べ替えられたデータの配列であり、データが変更されたことを示すことができます。 ここで、XNUMX つのアクションに対してコードのさまざまな部分で複数のイベントをトリガーする必要があるコードを記述することがどれほど便利になるかを想像してみてください。 ストリームを購読するだけで、変更が発生したときに通知されます。 RxJs ライブラリはこれを行うことができます。
RxJS は、監視可能なシーケンスを使用して非同期およびイベントベースのプログラムを操作するためのライブラリです。 ライブラリは主なタイプを提供します 観測できる、いくつかのヘルパー タイプ (オブザーバー、スケジューラー、サブジェクト) およびコレクションと同様にイベントを操作するための演算子 (マップ、フィルター、リデュース、すべて および JavaScript 配列の同様のもの)。
このライブラリの基本的な概念を理解しましょう。
オブザーバブル、オブザーバー、プロデューサー
Observable は、最初に説明する基本型です。 このクラスには、RxJs 実装の主要部分が含まれています。 これは監視可能なストリームに関連付けられており、subscribe メソッドを使用してサブスクライブできます。
Observable は、更新を作成するための補助メカニズム、いわゆる オブザーバー。 オブザーバーの値のソースはと呼ばれます プロデューサー。 配列、イテレータ、Web ソケット、ある種のイベントなどが考えられます。 したがって、observable はプロデューサーとオブザーバーの間の導体であると言えます。
Observable は XNUMX 種類の Observer イベントを処理します。
- 次 - 新しいデータ
- error - シーケンスが例外により終了した場合のエラー。 このイベントはシーケンスの終了も意味します。
- complete - シーケンスの終了に関する信号。 これは、新しいデータがもう存在しないことを意味します
デモを見てみましょう:
最初は1、2、3、1秒後の値を処理します。 4 を取得してスレッドを終了します。
考えが口に出ていた
そして、それについて書くよりも伝える方が面白いことに気づきました。 😀
サブスクリプション
ストリームをサブスクライブするときに、新しいクラスを作成します 購読、これにより、メソッドで購読を解除するオプションが提供されます 登録解除。 次のメソッドを使用してサブスクリプションをグループ化することもできます 加えます。 そうですね、次を使用してスレッドのグループを解除できるのは論理的です。 削除します。 add メソッドとremove メソッドは、別のサブスクリプションを入力として受け入れます。 サブスクリプションを解除するときは、あたかも unsubscribe メソッドを呼び出したかのように、すべての子サブスクリプションからサブスクリプションを解除することに注意してください。 どうぞ。
ストリームの種類
HOT
コールド
プロデューサーは観測可能な外部で作成されます
プロデューサーはobservable内で作成されます
データはオブザーバブルの作成時に渡されます
データは購読時に提供されます。
購読を解除するにはさらにロジックが必要です
スレッドが勝手に終了する
XNUMX対多の関係を使用します
XNUMX 対 XNUMX の関係を使用します
すべてのサブスクリプションには同じ価値があります
サブスクリプションは独立しています
契約していない場合、データが失われる可能性があります
新しいサブスクリプションのすべてのストリーム値を再発行します
例えるなら、映画館の映画のような熱い流れをイメージします。 どの時点で来たのか、その瞬間から見始めました。 私はそれらのコールドストリームとコールを比較します。 サポート。 発信者は誰でも留守番電話の録音を最初から最後まで聞くことができますが、登録を解除すれば電話を切ることができます。
いわゆる暖かい流れもあることに注意したいと思います（私はそのような定義に遭遇したのは非常にまれで、外国人コミュニティでのみです） - これは冷たい流れから熱い流れに変わる流れです。 疑問が生じます - どこで使用するかです））実践から例を示します。
私はAngularを使って作業しています。 彼は rxjs を積極的に使用しています。 サーバーにデータを取得するには、コールド ストリームが必要であり、asyncPipe を使用してテンプレートでこのストリームを使用します。 このパイプを数回使用すると、コールド ストリームの定義に戻り、各パイプがサーバーにデータを要求しますが、これは控えめに言っても奇妙です。 そして、コールド ストリームをウォーム ストリームに変換すると、リクエストは XNUMX 回発生します。
一般に、フローのタイプを理解することは初心者にとって非常に困難ですが、重要です。
オペレーター
return this.http.get(`${environment.apiUrl}/${this.apiUrl}/trade_companies`)
.pipe(
tap(({ data }: TradeCompanyList) => this.companies$$.next(cloneDeep(data))),
map(({ data }: TradeCompanyList) => data)
);オペレーターは、ストリームを操作する機会を提供します。 これらは、Observable 内を流れるイベントの制御に役立ちます。 最も人気のあるものをいくつか検討します。演算子に関する詳細情報は、役立つ情報のリンクで見つけることができます。
演算子
のヘルパー演算子から始めましょう。 単純な値に基づいて Observable を作成します。
演算子フィルター
フィルター オペレーターは、その名前が示すように、ストリーム信号をフィルター処理します。 演算子が true を返した場合、さらにスキップします。
演算子 - テイク
take - ストリームが終了するまでのエミット数の値を取得します。
オペレーター-debounceTime
debounceTime - 出力データ間の指定された時間間隔内にある発行された値を破棄します - 時間間隔が経過した後、最後の値を発行します。
const { Observable } = Rx;
const { debounceTime, take } = RxOperators;
Observable.create((observer) => {
let i = 1;
observer.next(i++);
// Испускаем значение раз в 1000мс
setInterval(() => {
observer.next(i++)
}, 1000);
// Испускаем значение раз в 1500мс
setInterval(() => {
observer.next(i++)
}, 1500);
}).pipe(
debounceTime(700), // Ожидаем 700мс значения прежде чем обработать
take(3)
);
演算子-takewhile
takewhile が false を返すまで値を発行し、その後スレッドからサブスクライブを解除します。
const { Observable } = Rx;
const { debounceTime, takeWhile } = RxOperators;
Observable.create((observer) => {
let i = 1;
observer.next(i++);
// Испускаем значение раз в 1000мс
setInterval(() => {
observer.next(i++)
}, 1000);
}).pipe(
takeWhile( producer => producer < 5 )
);
演算子結合最新
結合演算子 combinlatest は、promise.all に似ています。 複数のストリームを XNUMX つに結合します。 各スレッドが少なくとも XNUMX つの発行を行った後、それぞれから最新の値を配列として取得します。 さらに、結合されたストリームからの出力後、新しい値が与えられます。
const { combineLatest, Observable } = Rx;
const { take } = RxOperators;
const observer_1 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 1000мс
setInterval(() => {
observer.next('a: ' + i++);
}, 1000);
});
const observer_2 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 750мс
setInterval(() => {
observer.next('b: ' + i++);
}, 750);
});
combineLatest(observer_1, observer_2).pipe(take(5));
演算子-zip
Zip - 各ストリームからの値を待機し、これらの値に基づいて配列を形成します。 値がどのスレッドからもたらされない場合、グループは形成されません。
const { zip, Observable } = Rx;
const { take } = RxOperators;
const observer_1 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 1000мс
setInterval(() => {
observer.next('a: ' + i++);
}, 1000);
});
const observer_2 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 750
setInterval(() => {
observer.next('b: ' + i++);
}, 750);
});
const observer_3 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 500
setInterval(() => {
observer.next('c: ' + i++);
}, 500);
});
zip(observer_1, observer_2, observer_3).pipe(take(5));
演算子 - forkJoin
forkJoin もスレッドに参加しますが、すべてのスレッドが完了した場合にのみ値を出力します。
const { forkJoin, Observable } = Rx;
const { take } = RxOperators;
const observer_1 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 1000мс
setInterval(() => {
observer.next('a: ' + i++);
}, 1000);
}).pipe(take(3));
const observer_2 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 750
setInterval(() => {
observer.next('b: ' + i++);
}, 750);
}).pipe(take(5));
const observer_3 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 500
setInterval(() => {
observer.next('c: ' + i++);
}, 500);
}).pipe(take(4));
forkJoin(observer_1, observer_2, observer_3);
オペレーターマップ
マップ変換演算子は、放出値を新しい値に変換します。
const { Observable } = Rx;
const { take, map } = RxOperators;
Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 1000мс
setInterval(() => {
observer.next(i++);
}, 1000);
}).pipe(
map(x => x * 10),
take(3)
);
オペレーター - 共有、タップ
タップ演算子を使用すると、副作用、つまりシーケンスに影響を与えないアクションを実行できます。
共有ユーティリティのオペレータは、コールド ストリームをホット ストリームに変えることができます。
オペレーターは完了です。 主題に移りましょう。
考えが口に出ていた
そして、お茶を飲みに行きました。 こういう例はもう飽きた 😀
被験者の家族
サブジェクト ファミリは、ホット スレッドの代表的な例です。 これらのクラスは、オブザーバブルとオブザーバーとして同時に機能する一種のハイブリッドです。 話題のストリームなので、購読を解除する必要があります。 主な方法について言えば、次のとおりです。
- 次 - 新しいデータをストリームに渡す
- error - エラーとスレッドの終了
- 完了 - スレッドの終わり
- 購読 - ストリームを購読する
- unsubscribe - スレッドからの購読を解除します
- asObservable - オブザーバーに変換します
- toPromise - 約束に変換します
4 5種類の科目を割り当てます。
考えが口に出ていた
配信では4つと言ったのですが、あとXNUMXつ追加されていたようです。 ことわざにあるように、生きて学びましょう。
単純な主題 new Subject()- 最も単純な種類の主題。 パラメータなしで作成されました。 サブスクリプション後にのみ取得された値を渡します。
行動主体 new BehaviorSubject( defaultData<T> ) - 私の意見では、最も一般的なタイプの主題です。 入力にはデフォルト値が使用されます。 定期購読時に送信される最新号のデータを常に保存します。 このクラスには、ストリームの現在の値を返す便利な value メソッドもあります。
リプレイ件名 new ReplaySubject(bufferSize?: number, windowTime?: number) - オプションで、最初の引数として、それ自体に保存する値のバッファーのサイズを取り、XNUMX 番目の引数には変更が必要になります。
非同期件名 new AsyncSubject() - サブスクライブ時には何も起こらず、完了した場合にのみ値が返されます。 ストリームの最後の値のみが返されます。
Webソケットの件名 new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig<T> | Observable<T>, destination?: Observer<T>) - ドキュメントにはそれについて記載されておらず、私自身も初めて知りました。 彼が何をしているのか誰にも分からないので、書き加えておきます。
ふー。 さて、私が今日言いたかったことはすべて検討しました。 この情報がお役に立てば幸いです。 [役立つ情報] タブで文献リストを自分で読むことができます。
役立つ情報
出所： habr.com