异步编程解决方案之事件发布/订阅模式

Posted 2020-12-25 yingquliang

时间监听模式是广泛用于异步编程的模式，是回调函数的事件化，又称不发订阅模式。

nodejs的events模块就是发布订阅模式的一个简单实现，不存在preventDefault,stopPropagation,stopImmediatePropagation,等控制事件传递的方法。

它具有addListner/on(),once(),removeListner(),removeAllLisetner()和emit等基础监听事件方法。

事件发布/订阅十分简单，如下：

//订阅
emitter.on(‘event1‘, function (message) {
　　console.log(message)
})

//发布
emmiter.emit(‘event1‘, ‘this is a message‘);

可以看到，事件订阅就是一个高阶函数的应用。

 

事件侦听器就是一个钩子函数hook

const options = {
    host: ‘www.abc.com‘,
    port: ‘80‘,
    path: ‘/get‘,
    method: ‘GED‘
}

let req = http.request(options, function (res) {
    console.log(‘STATUS:‘ + res.statusCode);
    console.log(‘HEADERS:‘, JSON.stringify(res.headers));
    res.setEncoding(‘utf8‘);
    res.on(‘data‘, function (chunck) {
        console.log(chunck);
    });
    res.on(‘end‘, function () {
        console.log(‘ -- end -- ‘);
    })
    
    req.on(‘error‘, function (e) {
        console.log(‘problem with requrest: ‘ + e.message);
    })
    req.wirte(‘data
‘);
    req.wirte(‘data
‘);
    req.end();
});

在这段程序内，程序员只需要关注内部的error、data、end这些事件点上即可，至于内部的流程，无需关注。

备注：如果一个事件添加了超过十个侦听器就会被警告，防止CPU和内存暂用过多资源，emitter.setMaxListeners(0)将这个限制去掉；

　　　EventEmitter对error事件做了特殊对待，如果这个事件没有被捕获，就会引起线程退出。一个健壮的EventEmitter应该对error处理；

 

1、继承EventEmitter类是十分简单的，以下代码是node中Stream对象继承EventEmitter的例子

let events = require(‘events‘);
 
function Stream () {
  events.EventEmitter.call(this);
}

util.inherits(Stream, events.EventEmitter);

node核心模块中有近半数继承自EventEmitter

 

2、利用事件队列解决雪崩问题

假如站点刚好启动，这时候缓存中是不存在数据的，而如果访问量巨大，同一个sql发送到数据库中反复查询，会影响服务的整体性。

var select = function (callback) {
  db.select(‘SQL‘, function (results) {
    callback(results);
  })  
}

解决方案是添加一个状态锁

var status = ‘ready‘;
var select = function (callback) {
  if (status === ‘ready‘) {
    status === ‘pending‘;
    db.select(‘SQL‘, function(results) {
      status = ‘ready‘;
      callback(results);
    })
  }
}

但是这种情况下，多次调用select，只会有一次生效，后续select没有数据服务的，这个时候我们可以引入事件队列：

var proxy = new events.EventEmitter();
var status = ‘ready‘;
var select = function (callback) {
    proxy.once(‘selected‘, callback);
    if (status === ‘ready‘) {
        status === ‘pending‘;
        db.select(‘SQL‘, function (results) {
            proxy.emit(‘select‘, results);
            status = ‘ready‘;
        })
    }
}

这里面我应用了once方法，将请求回调都压在事件队列中，利用他执行一次就将监听器移除的特点，保证他每一个回调都执行一次。