本文介绍了Node异步编程,分享给大家,具体如下:
目前的异步编程主要解决方案有:
事件发布/订阅模式
Node自身提供了events模块,可以轻松实现事件的发布/订阅
//订阅 emmiter.on("event1",function(message){ console.log(message); }) //发布 emmiter.emit("event1","I am mesaage!");
侦听器可以很灵活地添加和删除,使得事件和具体处理逻辑之间可以很轻松的关联和解耦
事件发布/订阅模式常常用来解耦业务逻辑,事件发布者无需关注订阅的侦听器如何实现业务逻辑,甚至不用关注有多少个侦听器存在,数据通过消息的方式可以很灵活的进行传递。
下面的HTTP就是典型的应用场景
var req = http.request(options,function(res){ res.on('data',function(chunk){ console.log('Body:'+ chunk); }) res.on('end',function(){ //TODO }) })
如果一个事件添加了超过10个侦听器,将会得到一条警告,可以通过调用emmite.setMaxListeners(0)将这个限制去掉
继承events模块
var events = require('events'); function Stream(){ events.EventEmiiter.call(this); } util.inherits(Stream,events.EventEmitter);
利用事件队列解决雪崩问题
所谓雪崩问题,就是在高访问量,大并发量的情况下缓存失效的情况,此时大量的请求同时融入数据库中,数据库无法同时承受如此大的查询请求,进而往前影响到网站整体的响应速度
解决方案:
var proxy = new events.EventEmitter(); var status = "ready"; var seletc = function(callback){ proxy.once("selected",callback);//为每次请求订阅这个查询时间,推入事件回调函数队列 if(status === 'ready'){ status = 'pending';//设置状态为进行中以防止引起多次查询操作 db.select("SQL",function(results){ proxy.emit("selected",results); //查询操作完成后发布时间 status = 'ready';//重新定义为已准备状态 }) } }
多异步之间的协作方案
以上情况事件与侦听器的关系都是一对多的,但在异步编程中,也会出现事件与侦听器多对一的情况。
这里以渲染页面所需要的模板读取、数据读取和本地化资源读取为例简要介绍一下
var count = 0 ; var results = {}; var done = function(key,value){ result[key] = value; count++; if(count === 3){ render(results); } } fs.readFile(template_path,"utf8",function(err,template){ done('template',template) }) db.query(sql,function(err,data){ done('data',data); }) l10n.get(function(err,resources){ done('resources',resources) })
偏函数方案
var after = function(times,callback){ var count = 0, result = {}; return function(key,value){ results[key] = value; count++; if(count === times){ callback(results); } } } var done = after(times,render); var emitter = new events.Emitter(); emitter.on('done',done); //一个侦听器 emitter.on('done',other); //如果业务增长,可以完成多对多的方案 fs.readFile(template_path,"utf8",function(err,template){ emitter.emit('done','template',template); }) db.query(sql,function(err,data){ emitter.emit('done','data',data); }) l10n.get(function(err,resources){ emitter.emit('done','resources',resources) })
引入EventProxy模块方案
var proxy = new EventProxy(); proxy.all('template','data','resources',function(template,data,resources){ //TODO }) fs.readFile(template_path,'utf8',function(err,template){ proxy.emit('template',template); }) db.query(sql,function(err,data){ proxy.emit('data',data); }) l10n.get(function(err,resources){ proxy.emit('resources',resources); })
Promise/Deferred模式
以上使用事件的方式时,执行流程都需要被预先设定,这是发布/订阅模式的运行机制所决定的。
$.get('/api',{ success:onSuccess, err:onError, complete:onComplete }) //需要严谨设置目标
那么是否有一种先执行异步调用,延迟传递处理的方式的?接下来要说的就是针对这种情况的方式:Promise/Deferred模式
Promise/A
Promise/A提议对单个异步操作做出了这样的抽象定义:
一个Promise对象只要具备then()即可
通过Node的events模块来模拟一个Promise的实现
var Promise = function(){ EventEmitter.call(this) } util.inherits(Promise,EventEmitter); Promise.prototype.then = function(fulfilledHandler,errHandler,progeressHandler){ if(typeof fulfilledHandler === 'function'){ this.once('success',fulfilledHandler); //实现监听对应事件 } if(typeof errorHandler === 'function'){ this.once('error',errorHandler) } if(typeof progressHandler === 'function'){ this.on('progress',progressHandler); } return this; }
以上通过then()将回调函数存放起来,接下来就是等待success、error、progress事件被触发,实现这个功能的对象称为Deferred对象,即延迟对象。
var Deferred = function(){ this.state = 'unfulfilled'; this.promise = new Promise(); } Deferred.prototype.resolve = function(obj){ //当异步完成后可将resolve作为回调函数,触发相关事件 this.state = 'fulfilled'; this.promise.emit('success',obj); } Deferred.prototype.reject = function(err){ this.state = 'failed'; this.promise.emit('error',err); } Deferred.prototype.progress = function(data){ this.promise.emit('progress',data) }
因此,可以对一个典型的响应对象进行封装
res.setEncoding('utf8'); res.on('data',function(chunk){ console.log("Body:" + chunk); }) res.on('end',function(){ //done }) res.on('error',function(err){ //error }
转换成
res.then(function(){ //done },function(err){ //error },function(chunk){ console.log('Body:' + chunk); })
要完成上面的转换,首先需要对res对象进行封装,对data,end,error等事件进行promisify
var promisify = function(res){ var deferred = new Deferred(); //创建一个延迟对象来在res的异步完成回调中发布相关事件 var result = ''; //用来在progress中持续接收数据 res.on('data',function(chunk){ //res的异步操作,回调中发布事件 result += chunk; deferred.progress(chunk); }) res.on('end',function(){ deferred.resolve(result); }) res.on('error',function(err){ deferred.reject(err); }); return deferred.promise //返回deferred.promise,让外界不能改变deferred的状态,只能让promise的then()方法去接收外界来侦听相关事件。 } promisify(res).then(function(){ //done },function(err){ //error },function(chunk){ console.log('Body:' + chunk); })
以上,它将业务中不可变的部分封装在了Deferred中,将可变的部分交给了Promise
Promise中的多异步协作
Deferred.prototype.all = function(promises){ var count = promises.length; //记录传进的promise的个数 var that = this; //保存调用all的对象 var results = [];//存放所有promise完成的结果 promises.forEach(function(promise,i){//对promises逐个进行调用 promise.then(function(data){//每个promise成功之后,存放结果到result中,count--,直到所有promise被处理完了,才出发deferred的resolve方法,发布事件,传递结果出去 count--; result[i] = data; if(count === 0){ that.resolve(results); } },function(err){ that.reject(err); }); }); return this.promise; //返回promise来让外界侦听这个deferred发布的事件。 } var promise1 = readFile('foo.txt','utf-8');//这里的文件读取已经经过promise化 var promise2 = readFile('bar.txt','utf-8'); var deferred = new Deferred(); deferred.all([promise1,promise2]).thne(function(results){//promise1和promise2的then方法在deferred内部的all方法所调用,用于同步所有的promise //TODO },function(err){ //TODO })
支持序列执行的Promise
尝试改造一下代码以实现链式调用
var Deferred = function(){ this.promise = new Promise() } //完成态 Deferred.prototype.resolve = function(obj){ var promise = this.promise; var handler; while((handler = promise.queue.shift())){ if(handler && handler.fulfilled){ var ret = handler.fulfilled(obj); if(ret && ret.isPromise){ ret.queue = promise.queue; this.promise = ret; return; } } } } //失败态 Deferred.prototype.reject = function(err){ var promise = this.promise; var handler; while((handler = promise.queue.shift())){ if(handler && handler.error){ var ret = handler.error(err); if(ret && ret.isPromise){ ret.queue = promise.queue; this.promise = ret; return } } } } //生成回调函数 Deferred.prototype.callback = function(){ var that = this; return function(err,file){ if(err){ return that.reject(err); } that.resolve(file) } } var Promise = function(){ this.queue = []; //队列用于存储待执行的回到函数 this.isPromise = true; }; Promise.prototype.then = function(fulfilledHandler,errorHandler,progressHandler){ var handler = {}; if(typeof fulfilledHandler === 'function'){ handler.fulfilled = fulfilledHandler; } if(typeof errorHandler === 'function'){ handler.error = errorHandler; } this.queue.push(handler); return this; } var readFile1 = function(file,encoding){ var deferred = new Deferred(); fs.readFile(file,encoding,deferred.callback()); return deferred.promise; } var readFile2 = function(file,encoding){ var deferred = new Deferred(); fs.readFile(file,encoding,deferred.callback()); return deferred.promise; } readFile1('file1.txt','utf8').then(function(file1){ return readFile2(file1.trim(),'utf8') }).then(function(file2){ console.log(file2) })
流程控制库另外进行总结
参考《深入浅出node.js》一书,想学学习可以下载电子书,下载地址:https://www.nhooo.com/books/481114.html
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持呐喊教程。
声明:本文内容来源于网络,版权归原作者所有,内容由互联网用户自发贡献自行上传,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任。如果您发现有涉嫌版权的内容,欢迎发送邮件至:notice#nhooo.com(发邮件时,请将#更换为@)进行举报,并提供相关证据,一经查实,本站将立刻删除涉嫌侵权内容。