溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
這篇文章給大家分享的是有關(guān)JS如何實現(xiàn)請求調(diào)度器的內(nèi)容。小編覺得挺實用的，因此分享給大家做個參考，一起跟隨小編過來看看吧。
為了獲取一批互不依賴的資源，通常從性能考慮可以用 Promise.all(arrayOfPromises)來并發(fā)執(zhí)行。比如我們已有 100 個應(yīng)用的 id，需求是聚合所有應(yīng)用的 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));上面的代碼在應(yīng)用個數(shù)不多的情況下，可以運行正常。當(dāng)應(yīng)用個數(shù)達(dá)到成千上萬時，對支持并發(fā)數(shù)不是很好的系統(tǒng)，你的「壓測」會把第三放服務(wù)器搞掛，暫時無法響應(yīng)請求：
<html> <head><title>502 Bad Gateway</title></head> <body bgcolor="white"> <center><h2>502 Bad Gateway</h2></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));
},
/**
* 獲取所有應(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ā)請求 `concurrency` 個應(yīng)用，稱為一個 chunk
* 前一個 `chunk` 并發(fā)完成后一個才繼續(xù)，直至所有應(yīng)用獲取完畢
*
* @private
*
* @param {[Number]} ids
* @param {Number} concurrency 一次并發(fā)的請求數(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, '\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;
});
},
/**
* 聚合所有應(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);
}
};
// 開始運行
aggregator.start().then(console.log);運行結(jié)果如圖 2：
目的達(dá)到了，因具備通用性，下面開始抽象成一個模式以便復(fù)用。
先模擬一個 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ā)送出去，達(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);
// 簡寫，等價于
// return acc.then(() => cur());
}, Promise.resolve());
tasks.then(() => {
console.log('done');
});注意觀察控制臺輸出，應(yīng)該串行輸出以下內(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
}
]感謝各位的閱讀！關(guān)于“JS如何實現(xiàn)請求調(diào)度器”這篇文章就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，讓大家可以學(xué)到更多知識，如果覺得文章不錯，可以把它分享出去讓更多的人看到吧！
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點不代表本網(wǎng)站立場，如果涉及侵權(quán)請聯(lián)系站長郵箱：is@yisu.com進(jìn)行舉報，并提供相關(guān)證據(jù)，一經(jīng)查實，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號
手機(jī)網(wǎng)站二維碼