JSはリクエストディスパッチャーを実装する

はじめに: JS は当然並列リクエストをサポートしていますが、同時にターゲットサーバーに過度の負荷をかけるなどの問題も発生します。そこで、この記事では並列処理を制御する「リクエストスケジューラ」を紹介します。
TLDR; 「抽象化と再利用」セクションに直接ジャンプします。

独立したリソースのバッチを取得するには、パフォーマンス上の理由から、通常、 Promise.all(arrayOfPromises)使用して同時実行できます。たとえば、すでに 100 個のアプリケーション ID があり、すべてのアプリケーションの PV を集計する必要がある場合、通常は次のように記述します。

定数ID = [1001, 1002, 1003, 1004, 1005];
urlPrefix を 'http://opensearch.example.com/api/apps' に設定します。

// fetch関数はHTTPリクエストを送信し、Promiseを返します
const appPromises = ids.map(id => `${urlPrefix}/${id}`).map(fetch);

Promise.all(アプリPromises)
 // 累積して、reduce.then(apps => apps.reduce((initial, current) => initial + current.pv, 0)) を実行します。
 .catch((エラー) => console.log(エラー));

上記のコードは、アプリケーションがあまり多くない場合は正常に実行できます。アプリケーションの数が数万に達すると、同時リクエストを十分にサポートしていないシステムでは、「ストレステスト」によってサードパーティサーバーがクラッシュし、一時的にリクエストに応答できなくなります。

<html>
<head><title>502 不正なゲートウェイ</title></head>
<body bgcolor="white">
<center><h1>502 不正なゲートウェイ</h1></center>
<hr><center>nginx/1.10.1</center>
</本文>
</html>

どうすれば解決できるでしょうか?

自然な考え方としては、それほど多くの同時リクエストはサポートされていないため、それをいくつかの大きなブロックに分割し、各ブロックをchunkにすることができます。 chunk内のリクエストは引き続き同時ですが、チャンクサイズ ( chunkSize ) は、システムでサポートされる同時リクエストの最大数に制限されます。次のchunk 、前のchunkが終了した後にのみ実行を続行できます。つまり、 chunk内の要求は同時ですが、 chunk間の要求はシリアルです。アイデアは実はとてもシンプルですが、書くのは難しいです。まとめると、ブロック、シリアル、集約の3つの操作があります。

難しいのは、Promise をシリアルに実行する方法です。Promise は並列 ( Promise.all ) 機能のみを提供し、シリアル機能は提供していません。まず 3 つの簡単なリクエストから始めて、それらを実装し、問題を経験的に解決する方法を見ていきます。

// task1、task2、task3 は Promise を返す 3 つのファクトリ関数で、非同期リクエストをシミュレートします。const task1 = () => new Promise((resolve) => {
 タイムアウトを設定する(() => {
 解決する(1);
 console.log('タスク1が実行されました');
 }, 1000);
});

const task2 = () => 新しい Promise((resolve) => {
 タイムアウトを設定する(() => {
 解決する(2);
 console.log('タスク2が実行されました');
 }, 1000);
});

const task3 = () => 新しい Promise((resolve) => {
 タイムアウトを設定する(() => {
 解決する(3);
 console.log('タスク3が実行されました');
 }, 1000);
});

// 集計結果 let result = 0;

const resultPromise = [タスク1、タスク2、タスク3].reduce((現在、次) => 	 
 current.then((数値) => {
 console.log('numberで解決されました', number); // task2、task3のPromiseはここで解決されます
 結果 += 数値;

 次の() を返します。
 }）、
 
 Promise.resolve(0)) // 集約初期値.then(function(last) {
 console.log('number で解決された最後の Promise', last); // task3 の Promise はここで解決されます

 結果 += 最後;

 console.log('すべて実行され、結果が出ました', result);

 Promise.resolve(result) を返します。
 });

実行結果は図 1 に示されています。

コード分析: 実際に望む効果はfn1().then(() => fn2()).then(() => fn3()) 。上記のコードでPromiseのグループを順番に実行できるようにする鍵は、 reduce 「エンジン」がPromiseファクトリ関数の実行を段階的に実行していることです。

問題は解決されました。最終的なコードを見てみましょう。

/**
 * HTTPリクエストをシミュレートする * @param {String} url 
 * @return {プロミス}
 */
関数 fetch(url) {
 console.log(`${url} を取得しています`);
 新しいPromise((resolve) => {を返す
 setTimeout(() => 解決({ pv: Number(url.match(/\d+$/)) }), 2000);
 });
}

urlPrefix を 'http://opensearch.example.com/api/apps' に設定します。

定数アグリゲータ = {
 /**
 * 入力方法、スケジュールされたタスクの開始* 
 * @return {プロミス}
 */
 始める（） {
 this.fetchAppIds() を返す
 .then(ids => this.fetchAppsSerially(ids, 2)) を実行します。
 .then(アプリ => this.sumPv(アプリ))
 .catch(エラー => console.error(エラー));
 },
 
 /**
 * すべてのアプリケーションIDを取得する
 *
 * @プライベート
 * 
 * @return {プロミス}
 */
 アプリケーションIDを取得する() {
 Promise.resolve([1001, 1002, 1003, 1004, 1005]) を返します。
 },

 プロミスファクトリー(ID) {
 return () => Promise.all(ids.map(id => `${urlPrefix}/${id}`).map(fetch));
 },
 
 /**
 * すべてのアプリの詳細を取得 * 
 * `concurrency`アプリケーションの同時リクエストはチャンクと呼ばれます
 * 前のチャンクが同時に完了した後、すべてのアプリケーションがそれを取得するまで次のチャンクが続行されます。*
 * @プライベート
 *
 * @param {[Number]} ID
 * @param {Number} concurrency 一度に同時実行されるリクエストの数* @return {[Object]} すべてのアプリケーションに関する情報*/
 fetchAppsSerially(ids、同時実行性 = 100) {
 // チャンク化 let chunkOfIds = ids.splice(0, concurrency);
 定数タスク = [];
 
 (chunkOfIds.length !== 0) の場合 {
 タスクをプッシュします(this.promiseFactory(chunkOfIds));
 chunkOfIds = ids.splice(0, 同時実行性);
 }
 
 // ブロック順に実行 const result = [];
 タスクを返します。reduce((current, next) => current.then((chunkOfApps) => {
 console.info('チャンク', chunkOfApps.length, '同時実行要求が結果で終了しました:', chunkOfApps, '\n\n');
 result.push(...chunkOfApps); // 配列をフラット化する return next();
 })、Promise.resolve([]))
 .then((最後のアプリのチャンク) => {
 console.info('チャンク', lastchunkOfApps.length, '同時実行要求が結果で終了しました:', lastchunkOfApps, '\n\n');

 result.push(...lastchunkOfApps); // 再度フラット化します console.info('すべてのチャンクが result で実行されました', result);
 結果を返します。
 });
 },
 
 /**
 * すべてのアプリケーションの合計PV
 * 
 * @プライベート
 * 
 * @param {[]} アプリ 
 * @return {[type]} [説明]
 */
 合計Pv(アプリ) {
 定数初期値 = { pv: 0 };

 戻り値: apps.reduce((accumulator, app) => ({ pv: accumulator.pv + app.pv }), initial);
 }
};

// 実行を開始します。aggregator.start().then(console.log);

実行結果は図 2 に示されています。

抽象化と再利用

目的は達成されました。これは普遍的なので、再利用のためにパターンに抽象化していきます。

シリアル

まず、http get リクエストをシミュレートします。

/**
 * モックされた http get。
 * @param {文字列} URL
 * @returns {{ url: string; delay: number; }}
 */
関数 httpGet(url) {
 定数遅延 = Math.random() * 1000;

 console.info('GET', url);

 新しいPromise((resolve) => {を返す
 タイムアウトを設定する(() => {
 解決する（{
 URL、
 遅れ、
 日付: Date.now()
 })
 }、 遅れ）;
 })
}

一連のリクエストを順番に実行します。

定数ID = [1, 2, 3, 4, 5, 6, 7];

// バッチリクエスト関数。遅延によって実行される「関数」が正しいことに注意してください。そうでない場合、リクエストはすぐに送信され、シリアルの目的は達成されません。const httpGetters = ids.map(id => 
 () => httpGet(`https://jsonplaceholder.typicode.com/posts/${id}`)
);

// シリアル実行 const tasks = await httpGetters.reduce((acc, cur) => {
 acc.then(cur) を返します。
 
 // 省略形、 // return acc.then(() => cur()); と同等
}, Promise.resolve());

タスク.then(() => {
 console.log('完了');
});

コンソール出力に注意してください。次の内容がシリアルで出力されるはずです。

https://jsonplaceholder.typicode.com/posts/1 を取得します。
https://jsonplaceholder.typicode.com/posts/2 を取得します。
https://jsonplaceholder.typicode.com/posts/3 を取得します。
https://jsonplaceholder.typicode.com/posts/4 を取得します。
https://jsonplaceholder.typicode.com/posts/5 を取得します。
https://jsonplaceholder.typicode.com/posts/6 を取得します。
https://jsonplaceholder.typicode.com/posts/7 を取得します。

セグメントシリアル、セグメントパラレル

ここからが本題です。この記事のリクエストスケジューラの実装

/**
 * 約束をスケジュールします。
 * @param {Array<(...arg: any[]) => Promise<any>>} ファクトリ 
 * @param {number} 同時実行数 
 */
関数schedulePromises(factories, concurrency) {
 /**
 * かたまり
 * @param {any[]} 引数 
 * @param {数値} サイズ 
 * @returns {Array<any[]>}
 */
 const チャンク = (arr, サイズ = 1) => {
 arr.reduce((acc, cur, idx) => { を返します
 const modulo = idx % サイズ;

 （剰余 === 0）の場合{
 acc[acc.length] = [cur];
 } それ以外 {
 acc[acc.length - 1].push(cur);
 }

 acc を返します。
 }, [])
 };

 const chunks = chunk(factories, 並行性);

 resps = [] とします。

 チャンクを返す.reduce(
 (acc, cur) => {
 返品
 .then(() => {
  コンソールログ('---');
  Promise.all(cur.map(f => f())) を返します。
 })
 .then((中間応答) => {
  resps.push(...中間レスポンス);

  応答を返す。
 })
 },

 Promise.resolve()
 );
}

テスト中に、スケジューラを実行します。

// セグメント化されたシリアル、セグメント化された並列 schedulePromises(httpGetters, 3).then((resps) => {
 console.log('resps:', resps);
});

コンソール出力:

---
https://jsonplaceholder.typicode.com/posts/1 を取得します。
https://jsonplaceholder.typicode.com/posts/2 を取得します。
https://jsonplaceholder.typicode.com/posts/3 を取得します。
---
https://jsonplaceholder.typicode.com/posts/4 を取得します。
https://jsonplaceholder.typicode.com/posts/5 を取得します。
https://jsonplaceholder.typicode.com/posts/6 を取得します。
---
https://jsonplaceholder.typicode.com/posts/7 を取得します。

回答: [
 {
 「URL」: 「https://jsonplaceholder.typicode.com/posts/1」、
 「遅延」: 733.010980640727,
 「」：1615131322163
 },
 {
 "url": "https://jsonplaceholder.typicode.com/posts/2",
 「遅延」: 594.5056229848931,
 「」：1615131322024
 },
 {
 "url": "https://jsonplaceholder.typicode.com/posts/3",
 「遅延」: 738.8230109146299,
 「」：1615131322168
 },
 {
 「URL」: 「https://jsonplaceholder.typicode.com/posts/4」、
 「遅延」: 525.4604386109747,
 「」：1615131322698
 },
 {
 "url": "https://jsonplaceholder.typicode.com/posts/5",
 「遅延」: 29.086379722201183,
 「」：1615131322201
 },
 {
 "url": "https://jsonplaceholder.typicode.com/posts/6",
 「遅延」: 592.2345027398272,
 「」：1615131322765
 },
 {
 「url」: 「https://jsonplaceholder.typicode.com/posts/7」、
 「遅延」: 513.0684467560949,
 「」：1615131323284
 }
]

要約する

同時リクエストの数が多すぎる場合は、リクエストをブロックに分割して順番に処理し、ブロック内でリクエストを同時に処理することを検討できます。
問題は複雑に思えるかもしれませんが、まずは単純化し、次に重要なポイントを段階的に推論し、最後に抽象化して解決策を見つけることができます。
この記事の本質は、 reduceシリアル駆動エンジンとして使用することです。reduce をマスターすると、日々の開発で遭遇するパズルを解決するための新しいアイデアが得られます。reduce reduceマスターするには、前の記事「ついに Reduce を使用」を参照してください。

上記はリクエストスケジューラのJS実装の詳細です。JSリクエストスケジューラの詳細については、123WORDPRESS.COMの他の関連記事に注目してください。

以下もご興味があるかもしれません: