欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

JS 實現(xiàn)請求調(diào)度器

 更新時間：2021年03月08日 14:45:10   作者：孟陬  
這篇文章主要介紹了JS 實現(xiàn)請求調(diào)度器的方法，幫助大家更好的理解和學習使用js，感興趣的朋友可以了解下

前言：JS 天然支持并行請求，但與此同時會帶來一些問題，比如會造成目標服務(wù)器壓力過大，所以本文引入“請求調(diào)度器”來節(jié)制并發(fā)度。

TLDR; 直接跳轉(zhuǎn)『抽象和復用』章節(jié)。

為了獲取一批互不依賴的資源，通常從性能考慮可以用 Promise.all(arrayOfPromises)來并發(fā)執(zhí)行。比如我們已有 100 個應用的 id，需求是聚合所有應用的 PV，我們通常會這么寫：

const ids = [1001, 1002, 1003, 1004, 1005];
const urlPrefix = 'http://opensearch.example.com/api/apps';

// fetch 函數(shù)發(fā)送 HTTP 請求，返回 Promise
const appPromises = ids.map(id => `${urlPrefix}/${id}`).map(fetch);

Promise.all(appPromises)
 // 通過 reduce 做累加
 .then(apps => apps.reduce((initial, current) => initial + current.pv, 0))
 .catch((error) => console.log(error));

上面的代碼在應用個數(shù)不多的情況下，可以運行正常。當應用個數(shù)達到成千上萬時，對支持并發(fā)數(shù)不是很好的系統(tǒng)，你的「壓測」會把第三放服務(wù)器搞掛，暫時無法響應請求：

<html>
<head><title>502 Bad Gateway</title></head>
<body bgcolor="white">
<center><h1>502 Bad Gateway</h1></center>
<hr><center>nginx/1.10.1</center>
</body>
</html>

如何解決呢？

一個很自然的想法是，既然不支持這么多的并發(fā)請求，那就分割成幾大塊，每塊為一個 chunk，chunk 內(nèi)部的請求依然并發(fā)，但塊的大?。?code>chunkSize）限制在系統(tǒng)支持的最大并發(fā)數(shù)以內(nèi)。前一個 chunk 結(jié)束后一個 chunk 才能繼續(xù)執(zhí)行，也就是說 chunk 內(nèi)部的請求是并發(fā)的，但 chunk 之間是串行的。思路其實很簡單，寫起來卻有一定難度?？偨Y(jié)起來三個操作：分塊、串行、聚合

難點在如何串行執(zhí)行 Promise，Promise 僅提供了并行（Promise.all）功能，并沒有提供串行功能。我們從簡單的三個請求開始，看如何實現(xiàn)，啟發(fā)式解決問題（heuristic）。

// task1, task2, task3 是三個返回 Promise 的工廠函數(shù)，模擬我們的異步請求
const task1 = () => new Promise((resolve) => {
 setTimeout(() => {
 resolve(1);
 console.log('task1 executed');
 }, 1000);
});

const task2 = () => new Promise((resolve) => {
 setTimeout(() => {
 resolve(2);
 console.log('task2 executed');
 }, 1000);
});

const task3 = () => new Promise((resolve) => {
 setTimeout(() => {
 resolve(3);
 console.log('task3 executed');
 }, 1000);
});

// 聚合結(jié)果
let result = 0;

const resultPromise = [task1, task2, task3].reduce((current, next) => 	 
 current.then((number) => {
 console.log('resolved with number', number); // task2, task3 的 Promise 將在這里被 resolve
 result += number;

 return next();
 }),
 
 Promise.resolve(0)) // 聚合初始值

 .then(function(last) {
 console.log('The last promise resolved with number', last); // task3 的 Promise 在這里被 resolve

 result += last;

 console.log('all executed with result', result);

 return Promise.resolve(result);
 });

運行結(jié)果如圖 1：

代碼解析：我們想要的效果，直觀展示其實是 fn1().then(() => fn2()).then(() => fn3())。上面代碼能讓一組 Promise 按順序執(zhí)行的關(guān)鍵之處就在 reduce 這個“引擎”在一步步推動 Promise 工廠函數(shù)的執(zhí)行。

難點解決了，我們看看最終代碼：

/**
 * 模擬 HTTP 請求
 * @param {String} url 
 * @return {Promise}
 */
function fetch(url) {
 console.log(`Fetching ${url}`);
 return new Promise((resolve) => {
 setTimeout(() => resolve({ pv: Number(url.match(/\d+$/)) }), 2000);
 });
}

const urlPrefix = 'http://opensearch.example.com/api/apps';

const aggregator = {
 /**
 * 入口方法，開啟定時任務(wù)
 * 
 * @return {Promise}
 */
 start() {
 return this.fetchAppIds()
 .then(ids => this.fetchAppsSerially(ids, 2))
 .then(apps => this.sumPv(apps))
 .catch(error => console.error(error));
 },
 
 /**
 * 獲取所有應用的 ID
 *
 * @private
 * 
 * @return {Promise}
 */
 fetchAppIds() {
 return Promise.resolve([1001, 1002, 1003, 1004, 1005]);
 },

 promiseFactory(ids) {
 return () => Promise.all(ids.map(id => `${urlPrefix}/${id}`).map(fetch));
 },
 
 /**
 * 獲取所有應用的詳情
 * 
 * 一次并發(fā)請求 `concurrency` 個應用，稱為一個 chunk
 * 前一個 `chunk` 并發(fā)完成后一個才繼續(xù)，直至所有應用獲取完畢
 *
 * @private
 *
 * @param {[Number]} ids
 * @param {Number} concurrency 一次并發(fā)的請求數(shù)量
 * @return {[Object]}  所有應用的信息
 */
 fetchAppsSerially(ids, concurrency = 100) {
 // 分塊
 let chunkOfIds = ids.splice(0, concurrency);
 const tasks = [];
 
 while (chunkOfIds.length !== 0) {
 tasks.push(this.promiseFactory(chunkOfIds));
 chunkOfIds = ids.splice(0, concurrency);
 }
 
 // 按塊順序執(zhí)行
 const result = [];
 return tasks.reduce((current, next) => current.then((chunkOfApps) => {
 console.info('Chunk of', chunkOfApps.length, 'concurrency requests has finished with result:', chunkOfApps, '\n\n');
 result.push(...chunkOfApps); // 拍扁數(shù)組
 return next();
 }), Promise.resolve([]))
 .then((lastchunkOfApps) => {
 console.info('Chunk of', lastchunkOfApps.length, 'concurrency requests has finished with result:', lastchunkOfApps, '\n\n');

 result.push(...lastchunkOfApps); // 再次拍扁它
 console.info('All chunks has been executed with result', result);
 return result;
 });
 },
 
 /**
 * 聚合所有應用的 PV
 * 
 * @private
 * 
 * @param {[]} apps 
 * @return {[type]} [description]
 */
 sumPv(apps) {
 const initial = { pv: 0 };

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

// 開始運行
aggregator.start().then(console.log);

運行結(jié)果如圖 2：

抽象和復用

目的達到了，因具備通用性，下面開始抽象成一個模式以便復用。

串行

先模擬一個 http get 請求。

/**
 * mocked http get.
 * @param {string} url
 * @returns {{ url: string; delay: number; }}
 */
function httpGet(url) {
 const delay = Math.random() * 1000;

 console.info('GET', url);

 return new Promise((resolve) => {
 setTimeout(() => {
 resolve({
 url,
 delay,
 at: Date.now()
 })
 }, delay);
 })
}

串行執(zhí)行一批請求。

const ids = [1, 2, 3, 4, 5, 6, 7];

// 批量請求函數(shù)，注意是 delay 執(zhí)行的『函數(shù)』對了，否則會立即將請求發(fā)送出去，達不到串行的目的
const httpGetters = ids.map(id => 
 () => httpGet(`https://jsonplaceholder.typicode.com/posts/${id}`)
);

// 串行執(zhí)行之
const tasks = await httpGetters.reduce((acc, cur) => {
 return acc.then(cur);
 
 // 簡寫，等價于
 // return acc.then(() => cur());
}, Promise.resolve());

tasks.then(() => {
 console.log('done');
});

注意觀察控制臺輸出，應該串行輸出以下內(nèi)容：

GET https://jsonplaceholder.typicode.com/posts/1
GET https://jsonplaceholder.typicode.com/posts/2
GET https://jsonplaceholder.typicode.com/posts/3
GET https://jsonplaceholder.typicode.com/posts/4
GET https://jsonplaceholder.typicode.com/posts/5
GET https://jsonplaceholder.typicode.com/posts/6
GET https://jsonplaceholder.typicode.com/posts/7

分段串行，段中并行

重點來了。本文的請求調(diào)度器實現(xiàn)

/**
 * Schedule promises.
 * @param {Array<(...arg: any[]) => Promise<any>>} factories 
 * @param {number} concurrency 
 */
function schedulePromises(factories, concurrency) {
 /**
 * chunk
 * @param {any[]} arr 
 * @param {number} size 
 * @returns {Array<any[]>}
 */
 const chunk = (arr, size = 1) => {
 return arr.reduce((acc, cur, idx) => {
 const modulo = idx % size;

 if (modulo === 0) {
 acc[acc.length] = [cur];
 } else {
 acc[acc.length - 1].push(cur);
 }

 return acc;
 }, [])
 };

 const chunks = chunk(factories, concurrency);

 let resps = [];

 return chunks.reduce(
 (acc, cur) => {
 return acc
 .then(() => {
  console.log('---');
  return Promise.all(cur.map(f => f()));
 })
 .then((intermediateResponses) => {
  resps.push(...intermediateResponses);

  return resps;
 })
 },

 Promise.resolve()
 );
}

測試下，執(zhí)行調(diào)度器：

// 分段串行，段中并行
schedulePromises(httpGetters, 3).then((resps) => {
 console.log('resps:', resps);
});

控制臺輸出：

---
GET https://jsonplaceholder.typicode.com/posts/1
GET https://jsonplaceholder.typicode.com/posts/2
GET https://jsonplaceholder.typicode.com/posts/3
---
GET https://jsonplaceholder.typicode.com/posts/4
GET https://jsonplaceholder.typicode.com/posts/5
GET https://jsonplaceholder.typicode.com/posts/6
---
GET https://jsonplaceholder.typicode.com/posts/7

resps: [
 {
 "url": "https://jsonplaceholder.typicode.com/posts/1",
 "delay": 733.010980640727,
 "at": 1615131322163
 },
 {
 "url": "https://jsonplaceholder.typicode.com/posts/2",
 "delay": 594.5056229848931,
 "at": 1615131322024
 },
 {
 "url": "https://jsonplaceholder.typicode.com/posts/3",
 "delay": 738.8230109146299,
 "at": 1615131322168
 },
 {
 "url": "https://jsonplaceholder.typicode.com/posts/4",
 "delay": 525.4604386109747,
 "at": 1615131322698
 },
 {
 "url": "https://jsonplaceholder.typicode.com/posts/5",
 "delay": 29.086379722201183,
 "at": 1615131322201
 },
 {
 "url": "https://jsonplaceholder.typicode.com/posts/6",
 "delay": 592.2345027398272,
 "at": 1615131322765
 },
 {
 "url": "https://jsonplaceholder.typicode.com/posts/7",
 "delay": 513.0684467560949,
 "at": 1615131323284
 }
]

總結(jié)

  1. 如果并發(fā)請求的數(shù)量太大，可以考慮分塊串行，塊中請求并發(fā)。
  2. 問題看似復雜，不放先簡化之，然后一步步推導出關(guān)鍵點，最后抽象，就能找到解決方案。
  3. 本文的精髓在于使用 reduce 作為串行推動的引擎，故掌握其對我們?nèi)粘ｉ_發(fā)遇到的迷局破解可提供新思路，reduce 精通見上篇 你終于用 Reduce 了 🎉。

以上就是JS 實現(xiàn)請求調(diào)度器的詳細內(nèi)容，更多關(guān)于JS 請求調(diào)度器的資料請關(guān)注腳本之家其它相關(guān)文章！

相關(guān)文章

最新評論