JS 實(shí)現(xiàn)請(qǐng)求調(diào)度器
前言：JS 天然支持并行請(qǐng)求，但與此同時(shí)會(huì)帶來(lái)一些問(wèn)題，比如會(huì)造成目標(biāo)服務(wù)器壓力過(guò)大，所以本文引入“請(qǐng)求調(diào)度器”來(lái)節(jié)制并發(fā)度。
TLDR; 直接跳轉(zhuǎn)『抽象和復(fù)用』章節(jié)。
為了獲取一批互不依賴(lài)的資源，通常從性能考慮可以用 Promise.all(arrayOfPromises)來(lái)并發(fā)執(zhí)行。比如我們已有 100 個(gè)應(yīng)用的 id，需求是聚合所有應(yīng)用的 PV，我們通常會(huì)這么寫(xiě)：
const ids = [1001, 1002, 1003, 1004, 1005];const urlPrefix = ’http://opensearch.example.com/api/apps’;// fetch 函數(shù)發(fā)送 HTTP 請(qǐng)求，返回 Promiseconst appPromises = ids.map(id => `${urlPrefix}/${id}`).map(fetch);Promise.all(appPromises) // 通過(guò) reduce 做累加 .then(apps => apps.reduce((initial, current) => initial + current.pv, 0)) .catch((error) => console.log(error));
上面的代碼在應(yīng)用個(gè)數(shù)不多的情況下，可以運(yùn)行正常。當(dāng)應(yīng)用個(gè)數(shù)達(dá)到成千上萬(wàn)時(shí)，對(duì)支持并發(fā)數(shù)不是很好的系統(tǒng)，你的「壓測(cè)」會(huì)把第三放服務(wù)器搞掛，暫時(shí)無(wú)法響應(yīng)請(qǐng)求：
<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>
如何解決呢？
一個(gè)很自然的想法是，既然不支持這么多的并發(fā)請(qǐng)求，那就分割成幾大塊，每塊為一個(gè) chunk，chunk 內(nèi)部的請(qǐng)求依然并發(fā)，但塊的大?。╟hunkSize）限制在系統(tǒng)支持的最大并發(fā)數(shù)以?xún)?nèi)。前一個(gè) chunk 結(jié)束后一個(gè) chunk 才能繼續(xù)執(zhí)行，也就是說(shuō) chunk 內(nèi)部的請(qǐng)求是并發(fā)的，但 chunk 之間是串行的。思路其實(shí)很簡(jiǎn)單，寫(xiě)起來(lái)卻有一定難度?？偨Y(jié)起來(lái)三個(gè)操作：分塊、串行、聚合
難點(diǎn)在如何串行執(zhí)行 Promise，Promise 僅提供了并行（Promise.all）功能，并沒(méi)有提供串行功能。我們從簡(jiǎn)單的三個(gè)請(qǐng)求開(kāi)始，看如何實(shí)現(xiàn)，啟發(fā)式解決問(wèn)題（heuristic）。
// task1, task2, task3 是三個(gè)返回 Promise 的工廠(chǎng)函數(shù)，模擬我們的異步請(qǐng)求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); });
運(yùn)行結(jié)果如圖 1：
代碼解析：我們想要的效果，直觀(guān)展示其實(shí)是 fn1().then(() => fn2()).then(() => fn3())。上面代碼能讓一組 Promise 按順序執(zhí)行的關(guān)鍵之處就在 reduce 這個(gè)“引擎”在一步步推動(dòng) Promise 工廠(chǎng)函數(shù)的執(zhí)行。
難點(diǎn)解決了，我們看看最終代碼：
/** * 模擬 HTTP 請(qǐng)求 * @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 = { /** * 入口方法，開(kāi)啟定時(shí)任務(wù) * * @return {Promise} */ start() { return this.fetchAppIds() .then(ids => this.fetchAppsSerially(ids, 2)) .then(apps => this.sumPv(apps)) .catch(error => console.error(error)); }, /** * 獲取所有應(yīng)用的 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)); }, /** * 獲取所有應(yīng)用的詳情 * * 一次并發(fā)請(qǐng)求 `concurrency` 個(gè)應(yīng)用，稱(chēng)為一個(gè) chunk * 前一個(gè) `chunk` 并發(fā)完成后一個(gè)才繼續(xù)，直至所有應(yīng)用獲取完畢 * * @private * * @param {[Number]} ids * @param {Number} concurrency 一次并發(fā)的請(qǐng)求數(shù)量 * @return {[Object]} 所有應(yīng)用的信息 */ 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, ’nn’); result.push(...chunkOfApps); // 拍扁數(shù)組 return next(); }), Promise.resolve([])) .then((lastchunkOfApps) => { console.info(’Chunk of’, lastchunkOfApps.length, ’concurrency requests has finished with result:’, lastchunkOfApps, ’nn’); result.push(...lastchunkOfApps); // 再次拍扁它 console.info(’All chunks has been executed with result’, result); return result; }); }, /** * 聚合所有應(yīng)用的 PV * * @private * * @param {[]} apps * @return {[type]} [description] */ sumPv(apps) { const initial = { pv: 0 }; return apps.reduce((accumulator, app) => ({ pv: accumulator.pv + app.pv }), initial); }};// 開(kāi)始運(yùn)行aggregator.start().then(console.log);
運(yùn)行結(jié)果如圖 2：
目的達(dá)到了，因具備通用性，下面開(kāi)始抽象成一個(gè)模式以便復(fù)用。
串行先模擬一個(gè) http get 請(qǐng)求。
/** * 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í)行一批請(qǐng)求。
const ids = [1, 2, 3, 4, 5, 6, 7];// 批量請(qǐng)求函數(shù)，注意是 delay 執(zhí)行的『函數(shù)』對(duì)了，否則會(huì)立即將請(qǐng)求發(fā)送出去，達(dá)不到串行的目的const httpGetters = ids.map(id => () => httpGet(`https://jsonplaceholder.typicode.com/posts/${id}`));// 串行執(zhí)行之const tasks = await httpGetters.reduce((acc, cur) => { return acc.then(cur); // 簡(jiǎn)寫(xiě)，等價(jià)于 // return acc.then(() => cur());}, Promise.resolve());tasks.then(() => { console.log(’done’);});
注意觀(guān)察控制臺(tái)輸出，應(yīng)該串行輸出以下內(nèi)容：
GET https://jsonplaceholder.typicode.com/posts/1GET https://jsonplaceholder.typicode.com/posts/2GET https://jsonplaceholder.typicode.com/posts/3GET https://jsonplaceholder.typicode.com/posts/4GET https://jsonplaceholder.typicode.com/posts/5GET https://jsonplaceholder.typicode.com/posts/6GET https://jsonplaceholder.typicode.com/posts/7分段串行，段中并行
重點(diǎn)來(lái)了。本文的請(qǐng)求調(diào)度器實(shí)現(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() );}
測(cè)試下，執(zhí)行調(diào)度器：
// 分段串行，段中并行schedulePromises(httpGetters, 3).then((resps) => { console.log(’resps:’, resps);});
控制臺(tái)輸出：
---GET https://jsonplaceholder.typicode.com/posts/1GET https://jsonplaceholder.typicode.com/posts/2GET https://jsonplaceholder.typicode.com/posts/3---GET https://jsonplaceholder.typicode.com/posts/4GET https://jsonplaceholder.typicode.com/posts/5GET https://jsonplaceholder.typicode.com/posts/6---GET https://jsonplaceholder.typicode.com/posts/7resps: [ { '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é) 如果并發(fā)請(qǐng)求的數(shù)量太大，可以考慮分塊串行，塊中請(qǐng)求并發(fā)。 問(wèn)題看似復(fù)雜，不放先簡(jiǎn)化之，然后一步步推導(dǎo)出關(guān)鍵點(diǎn)，最后抽象，就能找到解決方案。 本文的精髓在于使用 reduce 作為串行推動(dòng)的引擎，故掌握其對(duì)我們?nèi)粘ｉ_(kāi)發(fā)遇到的迷局破解可提供新思路，reduce 精通見(jiàn)上篇 你終于用 Reduce 了 🎉。
以上就是JS 實(shí)現(xiàn)請(qǐng)求調(diào)度器的詳細(xì)內(nèi)容，更多關(guān)于JS 請(qǐng)求調(diào)度器的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章！
相關(guān)文章：
1. IntelliJ IDEA安裝插件的方法步驟2. 通過(guò)Django Admin+HttpRunner1.5.6實(shí)現(xiàn)簡(jiǎn)易接口測(cè)試平臺(tái)3. Docker 部署 Prometheus的安裝詳細(xì)教程4. IntelliJ IDEA設(shè)置自動(dòng)提示功能快捷鍵的方法5. idea重置默認(rèn)配置的方法步驟6. idea打開(kāi)多個(gè)窗口的操作方法7. idea設(shè)置代碼格式化的方法步驟8. idea給項(xiàng)目打war包的方法步驟9. IntelliJ IDEA調(diào)整字體大小的方法10. IntelliJ IDEA刪除類(lèi)的方法步驟
網(wǎng)公網(wǎng)安備