高效并发
使用Node时，我们经常需要在Node端调用其它远程服务，比如SOA，比如MongoDB等，假设我们是这样封装远程服务调用的：
function getDataFromRemoteServer(param) {
// 以下模拟远程服务执行
return new Promise(resolve => {
console.log('call getDataFromRemoteServer');
setTimeout(() => {
resolve(`Receive param: ${param}`);
}, 1000);
})
}
我们模拟下并发请求：
getDataFromRemoteServer('daniel').then(data => console.log(data));
getDataFromRemoteServer('daniel').then(data => console.log(data));
getDataFromRemoteServer('daniel').then(data => console.log(data));
结果打印：
call getDataFromRemoteServer
call getDataFromRemoteServer
call getDataFromRemoteServer
Receive param: daniel
Receive param: daniel
Receive param: daniel
3次调用返回相同的结果，所以有两次是浪费的（浪费网络资源，服务器的计算资源等），我们应该发挥节约美德，拒绝浪费。But How???
利用Javascript单线程特点，我们可以利用状态锁来限制后续方法的调用
我们引入状态锁来改造下原来的方法：
var status = 'ready';
function getDataFromRemoteServer(param) {
if (status === 'ready') {
status = 'pending';
// 以下模拟远程服务执行
return new Promise(resolve => {
console.log('call getDataFromRemoteServer');
setTimeout(() => {
resolve(`Receive param: ${param}`);
}, 1000);
}).then(data => {
status = 'ready';
return data;
})
}
}
getDataFromRemoteServer('daniel').then(data => console.log(data));
getDataFromRemoteServer('daniel').then(data => console.log(data));
getDataFromRemoteServer('daniel').then(data => console.log(data));
打印信息（你能看出我悄悄隐藏了错误信息吗?）
call getDataFromRemoteServer
Receive param: daniel
现在只打印了一个调用信息，但结果也只有一个，其它2个调用呢？
不好意思，其它2个报错了，因为除了第一次调用是正常返回后，其它请求都没有返回，所以也就没有.then的方法
为了避免报错，每一次返回都需要返回promise，我们引入事件机制，对代码再修改一下：
var EventEmitter = require('events');
var proxy = new EventEmitter();
var status = 'ready';
function getDataFromRemoteServer(param) {
if (status === 'ready') {
status = 'pending';
// 以下模拟远程服务执行
return new Promise(resolve => {
console.log('call getDataFromRemoteServer');
setTimeout(() => {
resolve(`Receive param: ${param}`);
}, 1000);
}).then(data => {
proxy.emit('finish', data);
status = 'ready';
return data;
})
} else {
return new Promise(resolve => {
proxy.on('finish', data => {
resolve(data);
})
});
}
}
getDataFromRemoteServer('daniel').then(data => console.log(data));
getDataFromRemoteServer('daniel').then(data => console.log(data));
getDataFromRemoteServer('daniel').then(data => console.log(data));
打印如下：
call getDataFromRemoteServer
Receive param: daniel
Receive param: daniel
Receive param: daniel
Yeah！结果显示只调用了一次远程服务，而3次调用都正常返回了正确的结果。
别太得意，有潜在的问题，你看得出吗？
我们试下这样调用：
getDataFromRemoteServer('daniel').then(data => console.log(data));
getDataFromRemoteServer('daniel').then(data => console.log(data));
getDataFromRemoteServer('daniel').then(data => console.log(data));
// 在三次并发并且结果返回后再调用一次
setTimeout(() => {
getDataFromRemoteServer('daniel').then(data => console.log(data));
}, 2000)
打印如下：
call getDataFromRemoteServer
==> on finish event
==> on finish event
Receive param: daniel
Receive param: daniel
Receive param: daniel
call getDataFromRemoteServer
==> on finish event
==> on finish event
Receive param: daniel
是不是又看出浪费问题了。因为我们前3次并发，有2次监听了事件，但执行完后并没有移除事件，所以有两次并不需要的事件监听被触发，随着并发请求越来越多，监听队列也越积越多，这是浪费，也是隐患
所以我们是不是每次都要手动移除事件监听就可以？
当然，可以，不过我们可以用更优雅更简单的方式，就是用once。
我们把上面的on换成once，再见证一下奇迹吧
==> on finish event
==> on finish event
Receive param: daniel
Receive param: daniel
Receive param: daniel
call getDataFromRemoteServer
Receive param: daniel
好了，开香槟庆祝一下吧。
喂喂，还没完呢，难道每一个远程方法的调用，我们都要这样麻烦地处理吗？而且status和proxy在这里好像是"全局"的吧，其它方法昨办？难道一个方法用一个名称："iStatus", "jStatus", .....
我们是聪明的程序猿，怎么可以干这种stupid的事呢。
前人种树，后人乘凉。我们现在就来“种树”，至于后人乘不乘凉，反正我们是要乘凉的
所以接下来，我们来把代码封装一下，让每一个远程调用都可以简单优雅地高效并发
const EventEmitter = require('events');
const crypto = require('crypto');
// 已封装 并发
function concurrent(promiseFn) {
return function (...params) {
var key = crypto.createHash('md5').update(JSON.stringify(params)).digest("hex"); // 用参数作为键, 因为每个参数的调用结果返回都不一样, 所以应当不同的请求
var keyObj = promiseFn[key] = promiseFn[key] || {status: 'ready', proxy: new EventEmitter()};
// 加状态锁和once事件机制, 避免并发访问重复调用
if (keyObj.status === 'ready') {
keyObj.status = 'pending';
promiseFn.apply(null, params).then(result => {
keyObj.proxy.emit('finish', result);
delete keyObj.status;
}).catch(err => {
keyObj.proxy.emit('error', err);
})
}
return new Promise((resolve, reject) => {
keyObj.proxy.once('finish', result => {
delete keyObj.proxy;
delete promiseFn[key];
resolve(result);
});
keyObj.proxy.once('error', err => {
delete keyObj.proxy;
delete promiseFn[key];
reject(err);
});
});
}
}
使用起来灰常方便，让我们还原远程服务最初的样子，演示如下：
function getDataFromRemoteServer(param) {
// 以下模拟远程服务执行
return new Promise(resolve => {
console.log('call getDataFromRemoteServer');
setTimeout(() => {
resolve(`Receive param: ${param}`);
}, 1000);
})
}
var $getDataFromRemoteServer = concurrent(getDataFromRemoteServer);
$getDataFromRemoteServer('daniel').then(data => console.log(data));
$getDataFromRemoteServer('daniel').then(data => console.log(data));
$getDataFromRemoteServer('daniel').then(data => console.log(data));
// 在三次并发并且结果返回后再调用一次
setTimeout(() => {
$getDataFromRemoteServer('daniel').then(data => console.log(data));
}, 1500)
至此该文告一段落了。
这里声明下应用场景：如果并发访问远程服务，相同的参数一定返回相同的结果，那么就可以放心使用（如通过provinceId去取所有的cities）
--EOF--
网友评论