客服电话
APP下载
迪恩网络APP
随时随地掌握行业动态
官方微信
扫描二维码
关注迪恩网络微信公众号
为了获取一批互不依赖的资源，通常从性能考虑可以用
const ids = [1001, 1002, 1003, 1004, 1005];
const urlPrefix = 'http://opensearch.example.com/api/apps';
// fetch 函数发送 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));
上面的代码在应用个数不多的情况下，可以运行正常。当应用个数达到成千上万时，对支持并发数不是很好的系统，你的「压测」会把第三放服务器搞挂，暂时无法响应请求： <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> 如何解决呢？ 一个很自然的想法是，既然不支持这么多的并发请求，那就分割成几大块，每块为一个 难点在如何串行执行 Promise，Promise 仅提供了并行（
// task1, task2, task3 是三个返回 Promise 的工厂函数，模拟我们的异步请求
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);
});
// 聚合结果
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);
});
运行结果如图 1： 代码解析：我们想要的效果，直观展示其实是 难点解决了，我们看看最终代码：
/**
* 模拟 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 = {
/**
* 入口方法，开启定时任务
*
* @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));
},
/**
* 获取所有应用的详情
*
* 一次并发请求 `concurrency` 个应用，称为一个 chunk
* 前一个 `chunk` 并发完成后一个才继续，直至所有应用获取完毕
*
* @private
*
* @param {[Number]} ids
* @param {Number} concurrency 一次并发的请求数量
* @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);
}
// 按块顺序执行
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); // 拍扁数组
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);
运行结果如图 2： 抽象和复用
|
请发表评论