Node.js异步编程详解：事件发布订阅、Promise模式及流程控制库实践

本文旨在探讨Node.js环境下的异步编程实现机制，为开发者提供实践参考。

Node.js异步编程主要解决方案

当前Node.js中处理异步操作的主流方案主要包含以下三类：

事件发布与订阅模式
Promise或Deferred模式
流程控制库

事件发布与订阅模式

关于事件发布与订阅模式，

Node.js内置了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个侦听器，系统将输出一条警告信息。如需移除该限制，可调用emitter.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模式详解

Promise或Deferred模式

使用事件方式时，执行流程需预先设定，这是发布/订阅模式的固有运行机制所致。


$.get('/api',{
  success:onSuccess,
  err:onError,
  complete:onComplete
})
//需要严谨设置目标

是否存在一种先执行异步调用、延迟传递处理结果的方式？接下来介绍针对此类场景的解决方案：Promise/Deferred模式。

Promise/A规范

Promise/A提议对单个异步操作进行了如下抽象定义：

Promise对象仅处于三种状态之一：未完成态（Pending）、完成态（Fulfilled）和失败态（Rejected）。
Promise的状态转换方向固定，仅能从未完成态转向完成态或失败态，不可逆向转换，且完成态与失败态之间不可相互转化。
一旦Promise状态发生转换，该状态即不可更改。

一个具备then()方法的对象即可被视为Promise。

该方法接受处理完成态或错误态的回调函数。
可选地支持作为第三个参数的progress事件回调。
then()方法仅接受function类型的对象，其他类型将被忽略。
then()方法需返回一个新的Promise对象，以支持链式调用。

以下代码通过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
}

转换为Promise结构。


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.jb51.net/books/481114.html

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。

Node.js异步编程核心机制：事件、Promise与流程控制

Node.js异步编程主要解决方案

事件发布与订阅模式

多异步协作与雪崩问题

Promise/Deferred模式详解

Promise在多异步协作中的应用

实用工具推荐