JS 实现请求调度器

2023-07-04 13:14:03 170

前言：JS天然支持并行请求，但与此同时会带来一些问题，比如会造成目标服务器压力过大，所以本文引入“请求调度器”来节制并发度。

TLDR;直接跳转『抽象和复用』章节。

为了获取一批互不依赖的资源，通常从性能考虑可以用Promise.all(arrayOfPromises)来并发执行。比如我们已有100个应用的id，需求是聚合所有应用的PV，我们通常会这么写：

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

//fetch函数发送HTTP请求，返回Promise
constappPromises=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));

上面的代码在应用个数不多的情况下，可以运行正常。当应用个数达到成千上万时，对支持并发数不是很好的系统，你的「压测」会把第三放服务器搞挂，暂时无法响应请求：


502BadGateway

502BadGateway
nginx/1.10.1

如何解决呢？

一个很自然的想法是，既然不支持这么多的并发请求，那就分割成几大块，每块为一个chunk，chunk内部的请求依然并发，但块的大小（chunkSize）限制在系统支持的最大并发数以内。前一个chunk结束后一个chunk才能继续执行，也就是说chunk内部的请求是并发的，但chunk之间是串行的。思路其实很简单，写起来却有一定难度。总结起来三个操作：分块、串行、聚合

难点在如何串行执行Promise，Promise仅提供了并行（Promise.all）功能，并没有提供串行功能。我们从简单的三个请求开始，看如何实现，启发式解决问题（heuristic）。

//task1,task2,task3是三个返回Promise的工厂函数，模拟我们的异步请求
consttask1=()=>newPromise((resolve)=>{
setTimeout(()=>{
resolve(1);
console.log('task1executed');
},1000);
});

consttask2=()=>newPromise((resolve)=>{
setTimeout(()=>{
resolve(2);
console.log('task2executed');
},1000);
});

consttask3=()=>newPromise((resolve)=>{
setTimeout(()=>{
resolve(3);
console.log('task3executed');
},1000);
});

//聚合结果
letresult=0;

constresultPromise=[task1,task2,task3].reduce((current,next)=>	
current.then((number)=>{
console.log('resolvedwithnumber',number);//task2,task3的Promise将在这里被resolve
result+=number;

returnnext();
}),

Promise.resolve(0))//聚合初始值

.then(function(last){
console.log('Thelastpromiseresolvedwithnumber',last);//task3的Promise在这里被resolve

result+=last;

console.log('allexecutedwithresult',result);

returnPromise.resolve(result);
});

运行结果如图1：

代码解析：我们想要的效果，直观展示其实是fn1().then(()=>fn2()).then(()=>fn3())。上面代码能让一组Promise按顺序执行的关键之处就在reduce这个“引擎”在一步步推动Promise工厂函数的执行。

难点解决了，我们看看最终代码：

/**
*模拟HTTP请求
*@param{String}url
*@return{Promise}
*/
functionfetch(url){
console.log(`Fetching${url}`);
returnnewPromise((resolve)=>{
setTimeout(()=>resolve({pv:Number(url.match(/\d+$/))}),2000);
});
}

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

constaggregator={
/**
*入口方法，开启定时任务
*
*@return{Promise}
*/
start(){
returnthis.fetchAppIds()
.then(ids=>this.fetchAppsSerially(ids,2))
.then(apps=>this.sumPv(apps))
.catch(error=>console.error(error));
},

/**
*获取所有应用的ID
*
*@private
*
*@return{Promise}
*/
fetchAppIds(){
returnPromise.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){
//分块
letchunkOfIds=ids.splice(0,concurrency);
consttasks=[];

while(chunkOfIds.length!==0){
tasks.push(this.promiseFactory(chunkOfIds));
chunkOfIds=ids.splice(0,concurrency);
}

//按块顺序执行
constresult=[];
returntasks.reduce((current,next)=>current.then((chunkOfApps)=>{
console.info('Chunkof',chunkOfApps.length,'concurrencyrequestshasfinishedwithresult:',chunkOfApps,'\n\n');
result.push(...chunkOfApps);//拍扁数组
returnnext();
}),Promise.resolve([]))
.then((lastchunkOfApps)=>{
console.info('Chunkof',lastchunkOfApps.length,'concurrencyrequestshasfinishedwithresult:',lastchunkOfApps,'\n\n');

result.push(...lastchunkOfApps);//再次拍扁它
console.info('Allchunkshasbeenexecutedwithresult',result);
returnresult;
});
},

/**
*聚合所有应用的PV
*
*@private
*
*@param{[]}apps
*@return{[type]}[description]
*/
sumPv(apps){
constinitial={pv:0};

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

//开始运行
aggregator.start().then(console.log);

运行结果如图2：

抽象和复用

目的达到了，因具备通用性，下面开始抽象成一个模式以便复用。

串行

先模拟一个httpget请求。

/**
*mockedhttpget.
*@param{string}url
*@returns{{url:string;delay:number;}}
*/
functionhttpGet(url){
constdelay=Math.random()*1000;

console.info('GET',url);

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

串行执行一批请求。

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

//批量请求函数，注意是delay执行的『函数』对了，否则会立即将请求发送出去，达不到串行的目的
consthttpGetters=ids.map(id=>
()=>httpGet(`https://jsonplaceholder.typicode.com/posts/${id}`)
);

//串行执行之
consttasks=awaithttpGetters.reduce((acc,cur)=>{
returnacc.then(cur);

//简写，等价于
//returnacc.then(()=>cur());
},Promise.resolve());

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

注意观察控制台输出，应该串行输出以下内容：

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

分段串行，段中并行

重点来了。本文的请求调度器实现

/**
*Schedulepromises.
*@param{Array<(...arg:any[])=>Promise>}factories
*@param{number}concurrency
*/
functionschedulePromises(factories,concurrency){
/**
*chunk
*@param{any[]}arr
*@param{number}size
*@returns{Array}
*/
constchunk=(arr,size=1)=>{
returnarr.reduce((acc,cur,idx)=>{
constmodulo=idx%size;

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

returnacc;
},[])
};

constchunks=chunk(factories,concurrency);

letresps=[];

returnchunks.reduce(
(acc,cur)=>{
returnacc
.then(()=>{
console.log('---');
returnPromise.all(cur.map(f=>f()));
})
.then((intermediateResponses)=>{
resps.push(...intermediateResponses);

returnresps;
})
},

Promise.resolve()
);
}

测试下，执行调度器：

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

控制台输出：

---
GEThttps://jsonplaceholder.typicode.com/posts/1
GEThttps://jsonplaceholder.typicode.com/posts/2
GEThttps://jsonplaceholder.typicode.com/posts/3
---
GEThttps://jsonplaceholder.typicode.com/posts/4
GEThttps://jsonplaceholder.typicode.com/posts/5
GEThttps://jsonplaceholder.typicode.com/posts/6
---
GEThttps://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
}
]

总结

如果并发请求的数量太大，可以考虑分块串行，块中请求并发。
问题看似复杂，不放先简化之，然后一步步推导出关键点，最后抽象，就能找到解决方案。
本文的精髓在于使用reduce作为串行推动的引擎，故掌握其对我们日常开发遇到的迷局破解可提供新思路，reduce精通见上篇你终于用Reduce了

JS 实现请求调度器

502BadGateway

抽象和复用

串行

分段串行，段中并行

总结

热门推荐

随机推荐