浅谈Node.js之异步流控制
前言
在没有深度使用函数回调的经验的时候,去看这些内容还是有一点吃力的。由于Node.js独特的异步特性,才出现了“回调地狱”的问题,这篇文章中,我比较详细的记录了如何解决异步流问题。
文章会很长,而且这篇是对异步流模式的解释。文中会使用一个简单的网络蜘蛛的例子,它的作用是抓取指定URL的网页内容并保存在项目中,在文章的最后,可以找到整篇文章中的源码demo。
1.原生JavaScript模式
本篇不针对初学者,因此会省略掉大部分的基础内容的讲解:
(spider_v1.js)
constrequest=require("request"); constfs=require("fs"); constmkdirp=require("mkdirp"); constpath=require("path"); constutilities=require("./utilities"); functionspider(url,callback){ constfilename=utilities.urlToFilename(url); console.log(`filename:${filename}`); fs.exists(filename,exists=>{ if(!exists){ console.log(`Downloading${url}`); request(url,(err,response,body)=>{ if(err){ callback(err); }else{ mkdirp(path.dirname(filename),err=>{ if(err){ callback(err); }else{ fs.writeFile(filename,body,err=>{ if(err){ callback(err); }else{ callback(null,filename,true); } }); } }); } }); }else{ callback(null,filename,false); } }); } spider(process.argv[2],(err,filename,downloaded)=>{ if(err){ console.log(err); }elseif(downloaded){ console.log(`Completedthedownloadof${filename}`); }else{ console.log(`${filename}wasalreadydownloaded`); } });
上边的代码的流程大概是这样的:
- 把url转换成filename
- 判断该文件名是否存在,若存在直接返回,否则进入下一步
- 发请求,获取body
- 把body写入到文件中
这是一个非常简单版本的蜘蛛,他只能抓取一个url的内容,看到上边的回调多么令人头疼。那么我们开始进行优化。
首先,ifelse这种方式可以进行优化,这个很简单,不用多说,放一个对比效果:
///before if(err){ callback(err); }else{ callback(null,filename,true); } ///after if(err){ returncallback(err); } callback(null,filename,true);
代码这么写,嵌套就会少一层,但经验丰富的程序员会认为,这样写过重强调了error,我们编程的重点应该放在处理正确的数据上,在可读性上也存在这样的要求。
另一个优化是函数拆分,上边代码中的spider函数中,可以把下载文件和保存文件拆分出去。
(spider_v2.js)
constrequest=require("request"); constfs=require("fs"); constmkdirp=require("mkdirp"); constpath=require("path"); constutilities=require("./utilities"); functionsaveFile(filename,contents,callback){ mkdirp(path.dirname(filename),err=>{ if(err){ returncallback(err); } fs.writeFile(filename,contents,callback); }); } functiondownload(url,filename,callback){ console.log(`Downloading${url}`); request(url,(err,response,body)=>{ if(err){ returncallback(err); } saveFile(filename,body,err=>{ if(err){ returncallback(err); } console.log(`Downloadedandsaved:${url}`); callback(null,body); }); }) } functionspider(url,callback){ constfilename=utilities.urlToFilename(url); console.log(`filename:${filename}`); fs.exists(filename,exists=>{ if(exists){ returncallback(null,filename,false); } download(url,filename,err=>{ if(err){ returncallback(err); } callback(null,filename,true); }) }); } spider(process.argv[2],(err,filename,downloaded)=>{ if(err){ console.log(err); }elseif(downloaded){ console.log(`Completedthedownloadof${filename}`); }else{ console.log(`${filename}wasalreadydownloaded`); } });
上边的代码基本上是采用原生优化后的结果,但这个蜘蛛的功能太过简单,我们现在需要抓取某个网页中的所有url,这样才会引申出串行和并行的问题。
(spider_v3.js)
constrequest=require("request"); constfs=require("fs"); constmkdirp=require("mkdirp"); constpath=require("path"); constutilities=require("./utilities"); functionsaveFile(filename,contents,callback){ mkdirp(path.dirname(filename),err=>{ if(err){ returncallback(err); } fs.writeFile(filename,contents,callback); }); } functiondownload(url,filename,callback){ console.log(`Downloading${url}`); request(url,(err,response,body)=>{ if(err){ returncallback(err); } saveFile(filename,body,err=>{ if(err){ returncallback(err); } console.log(`Downloadedandsaved:${url}`); callback(null,body); }); }) } ///最大的启发是实现了如何异步循环遍历数组 functionspiderLinks(currentUrl,body,nesting,callback){ if(nesting===0){ returnprocess.nextTick(callback); } constlinks=utilities.getPageLinks(currentUrl,body); functioniterate(index){ if(index===links.length){ returncallback(); } spider(links[index],nesting-1,err=>{ if(err){ returncallback(err); } iterate((index+1)); }) } iterate(0); } functionspider(url,nesting,callback){ constfilename=utilities.urlToFilename(url); fs.readFile(filename,"utf8",(err,body)=>{ if(err){ if(err.code!=='ENOENT'){ returncallback(err); } returndownload(url,filename,(err,body)=>{ if(err){ returncallback(err); } spiderLinks(url,body,nesting,callback); }); } spiderLinks(url,body,nesting,callback); }); } spider(process.argv[2],2,(err,filename,downloaded)=>{ if(err){ console.log(err); }elseif(downloaded){ console.log(`Completedthedownloadof${filename}`); }else{ console.log(`${filename}wasalreadydownloaded`); } });
上边的代码相比之前的代码多了两个核心功能,首先是通过辅助类获取到了某个body中的links:
constlinks=utilities.getPageLinks(currentUrl,body);
内部实现就不解释了,另一个核心代码就是:
///最大的启发是实现了如何异步循环遍历数组 functionspiderLinks(currentUrl,body,nesting,callback){ if(nesting===0){ returnprocess.nextTick(callback); } constlinks=utilities.getPageLinks(currentUrl,body); functioniterate(index){ if(index===links.length){ returncallback(); } spider(links[index],nesting-1,err=>{ if(err){ returncallback(err); } iterate((index+1)); }) } iterate(0); }
可以说上边这一小段代码,就是采用原生实现异步串行的pattern了。除了这些之外,还引入了nesting的概念,通过这是这个属性,可以控制抓取层次。
到这里我们就完整的实现了串行的功能,考虑到性能,我们要开发并行抓取的功能。
(spider_v4.js)
constrequest=require("request"); constfs=require("fs"); constmkdirp=require("mkdirp"); constpath=require("path"); constutilities=require("./utilities"); functionsaveFile(filename,contents,callback){ mkdirp(path.dirname(filename),err=>{ if(err){ returncallback(err); } fs.writeFile(filename,contents,callback); }); } functiondownload(url,filename,callback){ console.log(`Downloading${url}`); request(url,(err,response,body)=>{ if(err){ returncallback(err); } saveFile(filename,body,err=>{ if(err){ returncallback(err); } console.log(`Downloadedandsaved:${url}`); callback(null,body); }); }) } ///最大的启发是实现了如何异步循环遍历数组 functionspiderLinks(currentUrl,body,nesting,callback){ if(nesting===0){ returnprocess.nextTick(callback); } constlinks=utilities.getPageLinks(currentUrl,body); if(links.length===0){ returnprocess.nextTick(callback); } letcompleted=0,hasErrors=false; functiondone(err){ if(err){ hasErrors=true; returncallback(err); } if(++completed===links.length&&!hasErrors){ returncallback(); } } links.forEach(link=>{ spider(link,nesting-1,done); }); } constspidering=newMap(); functionspider(url,nesting,callback){ if(spidering.has(url)){ returnprocess.nextTick(callback); } spidering.set(url,true); constfilename=utilities.urlToFilename(url); ///Inthispattern,therewillbesomeissues. ///Possibleproblemstodownloadthesameurlagainandagain。 fs.readFile(filename,"utf8",(err,body)=>{ if(err){ if(err.code!=='ENOENT'){ returncallback(err); } returndownload(url,filename,(err,body)=>{ if(err){ returncallback(err); } spiderLinks(url,body,nesting,callback); }); } spiderLinks(url,body,nesting,callback); }); } spider(process.argv[2],2,(err,filename,downloaded)=>{ if(err){ console.log(err); }elseif(downloaded){ console.log(`Completedthedownloadof${filename}`); }else{ console.log(`${filename}wasalreadydownloaded`); } });
这段代码同样很简单,也有两个核心内容。一个是如何实现并发:
///最大的启发是实现了如何异步循环遍历数组 functionspiderLinks(currentUrl,body,nesting,callback){ if(nesting===0){ returnprocess.nextTick(callback); } constlinks=utilities.getPageLinks(currentUrl,body); if(links.length===0){ returnprocess.nextTick(callback); } letcompleted=0,hasErrors=false; functiondone(err){ if(err){ hasErrors=true; returncallback(err); } if(++completed===links.length&&!hasErrors){ returncallback(); } } links.forEach(link=>{ spider(link,nesting-1,done); }); }
上边的代码可以说是实现并发的一个pattern。利用循环遍历来实现。另一个核心是,既然是并发的,那么利用fs.exists就会存在问题,可能会重复下载同一文件,这里的解决方案是:
- 使用Map缓存某一url,url应该作为key
现在我们又有了新的需求,要求限制同时并发的最大数,那么在这里就引进了一个我认为最重要的概念:队列。
(task-Queue.js)
classTaskQueue{ constructor(concurrency){ this.concurrency=concurrency; this.running=0; this.queue=[]; } pushTask(task){ this.queue.push(task); this.next(); } next(){ while(this.running{ this.running--; this.next(); }); this.running++; } } } module.exports=TaskQueue;
上边的代码就是队列的实现代码,核心是next()方法,可以看出,当task加入队列中后,会立刻执行,这不是说这个任务一定马上执行,而是指的是next会立刻调用。
(spider_v5.js)
constrequest=require("request"); constfs=require("fs"); constmkdirp=require("mkdirp"); constpath=require("path"); constutilities=require("./utilities"); constTaskQueue=require("./task-Queue"); constdownloadQueue=newTaskQueue(2); functionsaveFile(filename,contents,callback){ mkdirp(path.dirname(filename),err=>{ if(err){ returncallback(err); } fs.writeFile(filename,contents,callback); }); } functiondownload(url,filename,callback){ console.log(`Downloading${url}`); request(url,(err,response,body)=>{ if(err){ returncallback(err); } saveFile(filename,body,err=>{ if(err){ returncallback(err); } console.log(`Downloadedandsaved:${url}`); callback(null,body); }); }) } ///最大的启发是实现了如何异步循环遍历数组 functionspiderLinks(currentUrl,body,nesting,callback){ if(nesting===0){ returnprocess.nextTick(callback); } constlinks=utilities.getPageLinks(currentUrl,body); if(links.length===0){ returnprocess.nextTick(callback); } letcompleted=0,hasErrors=false; links.forEach(link=>{ ///给队列出传递一个任务,这个任务首先是一个函数,其次该函数接受一个参数 ///当调用任务时,触发该函数,然后给函数传递一个参数,告诉该函数在任务结束时干什么 downloadQueue.pushTask(done=>{ spider(link,nesting-1,err=>{ ///这里表示,只要发生错误,队列就会退出 if(err){ hasErrors=true; returncallback(err); } if(++completed===links.length&&!hasErrors){ callback(); } done(); }); }); }); } constspidering=newMap(); functionspider(url,nesting,callback){ if(spidering.has(url)){ returnprocess.nextTick(callback); } spidering.set(url,true); constfilename=utilities.urlToFilename(url); ///Inthispattern,therewillbesomeissues. ///Possibleproblemstodownloadthesameurlagainandagain。 fs.readFile(filename,"utf8",(err,body)=>{ if(err){ if(err.code!=='ENOENT'){ returncallback(err); } returndownload(url,filename,(err,body)=>{ if(err){ returncallback(err); } spiderLinks(url,body,nesting,callback); }); } spiderLinks(url,body,nesting,callback); }); } spider(process.argv[2],2,(err,filename,downloaded)=>{ if(err){ console.log(`error:${err}`); }elseif(downloaded){ console.log(`Completedthedownloadof${filename}`); }else{ console.log(`${filename}wasalreadydownloaded`); } });
因此,为了限制并发的个数,只需在spiderLinks方法中,把task遍历放入队列就可以了。这相对来说很简单。
到这里为止,我们使用原生JavaScript实现了一个有相对完整功能的网络蜘蛛,既能串行,也能并发,还可以控制并发个数。
2.使用async库
把不同的功能放到不同的函数中,会给我们带来巨大的好处,async库十分流行,它的性能也不错,它内部基于callback。
(spider_v6.js)
constrequest=require("request"); constfs=require("fs"); constmkdirp=require("mkdirp"); constpath=require("path"); constutilities=require("./utilities"); constseries=require("async/series"); consteachSeries=require("async/eachSeries"); functiondownload(url,filename,callback){ console.log(`Downloading${url}`); letbody; series([ callback=>{ request(url,(err,response,resBody)=>{ if(err){ returncallback(err); } body=resBody; callback(); }); }, mkdirp.bind(null,path.dirname(filename)), callback=>{ fs.writeFile(filename,body,callback); } ],err=>{ if(err){ returncallback(err); } console.log(`Downloadedandsaved:${url}`); callback(null,body); }); } ///最大的启发是实现了如何异步循环遍历数组 functionspiderLinks(currentUrl,body,nesting,callback){ if(nesting===0){ returnprocess.nextTick(callback); } constlinks=utilities.getPageLinks(currentUrl,body); if(links.length===0){ returnprocess.nextTick(callback); } eachSeries(links,(link,cb)=>{ "usestrict"; spider(link,nesting-1,cb); },callback); } constspidering=newMap(); functionspider(url,nesting,callback){ if(spidering.has(url)){ returnprocess.nextTick(callback); } spidering.set(url,true); constfilename=utilities.urlToFilename(url); fs.readFile(filename,"utf8",(err,body)=>{ if(err){ if(err.code!=='ENOENT'){ returncallback(err); } returndownload(url,filename,(err,body)=>{ if(err){ returncallback(err); } spiderLinks(url,body,nesting,callback); }); } spiderLinks(url,body,nesting,callback); }); } spider(process.argv[2],1,(err,filename,downloaded)=>{ if(err){ console.log(err); }elseif(downloaded){ console.log(`Completedthedownloadof${filename}`); }else{ console.log(`${filename}wasalreadydownloaded`); } });
在上边的代码中,我们只使用了async的三个功能:
constseries=require("async/series");//串行 consteachSeries=require("async/eachSeries");//并行 constqueue=require("async/queue");//队列
由于比较简单,就不做解释了。async中的队列的代码在(spider_v7.js)中,和上边我们自定义的队列很相似,也不做更多解释了。
3.Promise
Promise是一个协议,有很多库实现了这个协议,我们用的是ES6的实现。简单来说promise就是一个约定,如果完成了,就调用它的resolve方法,失败了就调用它的reject方法。它内有实现了then方法,then返回promise本身,这样就形成了调用链。
其实Promise的内容有很多,在实际应用中是如何把普通的函数promise化。这方面的内容在这里也不讲了,我自己也不够格
(spider_v8.js)
constutilities=require("./utilities"); constrequest=utilities.promisify(require("request")); constfs=require("fs"); constreadFile=utilities.promisify(fs.readFile); constwriteFile=utilities.promisify(fs.writeFile); constmkdirp=utilities.promisify(require("mkdirp")); constpath=require("path"); functionsaveFile(filename,contents,callback){ mkdirp(path.dirname(filename),err=>{ if(err){ returncallback(err); } fs.writeFile(filename,contents,callback); }); } functiondownload(url,filename){ console.log(`Downloading${url}`); letbody; returnrequest(url) .then(response=>{ "usestrict"; body=response.body; returnmkdirp(path.dirname(filename)); }) .then(()=>writeFile(filename,body)) .then(()=>{ "usestrict"; console.log(`Downloadedadnsaved:${url}`); returnbody; }); } ///promise编程的本质就是为了解决在函数中设置回调函数的问题 ///通过中间层promise来实现异步函数同步化 functionspiderLinks(currentUrl,body,nesting){ letpromise=Promise.resolve(); if(nesting===0){ returnpromise; } constlinks=utilities.getPageLinks(currentUrl,body); links.forEach(link=>{ "usestrict"; promise=promise.then(()=>spider(link,nesting-1)); }); returnpromise; } functionspider(url,nesting){ constfilename=utilities.urlToFilename(url); returnreadFile(filename,"utf8") .then( body=>spiderLinks(url,body,nesting), err=>{ "usestrict"; if(err.code!=='ENOENT'){ ///抛出错误,这个方便与在整个异步链的最后通过呢catch来捕获这个链中的错误 throwerr; } returndownload(url,filename) .then(body=>spiderLinks(url,body,nesting)); } ); } spider(process.argv[2],1) .then(()=>{ "usestrict"; console.log('Downloadcomplete'); }) .catch(err=>{ "usestrict"; console.log(err); });
可以看到上边的代码中的函数都是没有callback的,只需要在最后catch就可以了。
在设计api的时候,应该支持两种方式,及支持callback,又支持promise
functionasyncDivision(dividend,divisor,cb){ returnnewPromise((resolve,reject)=>{ "usestrict"; process.nextTick(()=>{ constresult=dividend/divisor; if(isNaN(result)||!Number.isFinite(result)){ consterror=newError("Invalidoperands"); if(cb){ cb(error); } returnreject(error); } if(cb){ cb(null,result); } resolve(result); }); }); } asyncDivision(10,2,(err,result)=>{ "usestrict"; if(err){ returnconsole.log(err); } console.log(result); }); asyncDivision(22,11) .then((result)=>console.log(result)) .catch((err)=>console.log(err));
4.Generator
Generator很有意思,他可以让暂停函数和恢复函数,利用thunkify和co这两个库,我们下边的代码实现起来非常酷。
(spider_v9.js)
constthunkify=require("thunkify"); constco=require("co"); constpath=require("path"); constutilities=require("./utilities"); constrequest=thunkify(require("request")); constfs=require("fs"); constmkdirp=thunkify(require("mkdirp")); constreadFile=thunkify(fs.readFile); constwriteFile=thunkify(fs.writeFile); constnextTick=thunkify(process.nextTick); function*download(url,filename){ console.log(`Downloading${url}`); constresponse=yieldrequest(url); console.log(response); constbody=response[1]; yieldmkdirp(path.dirname(filename)); yieldwriteFile(filename,body); console.log(`Downloadedandsaved${url}`); returnbody; } function*spider(url,nesting){ constfilename=utilities.urlToFilename(url); letbody; try{ body=yieldreadFile(filename,"utf8"); }catch(err){ if(err.code!=='ENOENT'){ throwerr; } body=yielddownload(url,filename); } yieldspiderLinks(url,body,nesting); } function*spiderLinks(currentUrl,body,nesting){ if(nesting===0){ returnnextTick(); } constlinks=utilities.getPageLinks(currentUrl,body); for(leti=0;i总结
我并没有写promise和generator并发的代码。以上这些内容来自于这本书nodejs-design-patterns。
demo下载
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持毛票票。