想哪写哪,前面的坑如果不大碍事,咱就不填了。嘿嘿~
RPC:remote procedure call;是啥玩意就不说了。
下面直接看pomelo的实现:
对应module是pomelo-rpc,这个直接npm install可以就可以安装。
比如: npm install -d pomelo-rpc,下载到当前目录的node_modules目录里面。
安装完毕以后目录如下:
AUTHORS index.js lib LICENSE Makefile node_modules package.json README-Chinese.md README.md sample test
可以看到,实际上pomelo-rpc还是依赖不少其他module的,具体这些module干啥用,这个就不说了。自己看吧。
我们直接上:
首先sample目录如下:
sample/ ├── client.js ├── remote │ └── test │ └── service.js └── server.js直接node server.js可以启动一个server等待client的rpc调用。
我们再在其他终端里面:node client.js 就可以看到结果。
我们看看lib下面有什么:
lib/ ├── rpc-client │ ├── client.js │ ├── mailboxes │ │ ├── blackhole.js │ │ ├── tcp-mailbox.js │ │ └── ws-mailbox.js │ ├── mailbox.js │ ├── mailstation.js │ └── router.js ├── rpc-server #rpc 的server端实现 │ ├── acceptor.js #接收rpc的Factory,只有几行代码,这里写死了是直接调用ws-acceptor.js,也就是websocket的实现。 │ ├── acceptors #具体接收器的实现目录 │ │ ├── tcp-acceptor.js #tcp │ │ └── ws-acceptor.js #websocket │ ├── dispatcher.js #rpc的分发实现,具体就是根据下面会介绍到的树形结构来找到对应的rpc处理函数。 │ ├── gateway.js #这个实际上才是真正的server,负责服务的start,stop,并且初始化acceptor的callback函数 ———— dispatcher │ └── server.js #1、加载模块,并且把模块当成参数来初始化gateway。2:构建树形结构来注册这些被初始化的模块。 └── util ├── proxy.js ├── tracer.js └── utils.jsOK,从文件上看server的逻辑时序如下
server.js : #1、加载模块,并且把模块当成参数来初始化 gateway。2)构建树形结构,来注册这些被初始化的模块
gateway.js:# 初始化acceptor的callback函数 ———— dispatcher
acceptor.js:# acceptor的factory,这里写死了只生产ws类型的acceptor
ws-acceptor.js #websocket方式的接收器具体实现。
dispatcher.js:#简单根据client调用的树形结构来逐级查找注册的module,并且调用该module。
下面扒代码:
第一部分:注册,生成callback的树形结构
sample/server.js
1 var Server = require(‘..‘).server; 2 3 debugger; 4 // remote service path info list 5 var paths = [ 6 {namespace: ‘user‘, path: __dirname + ‘/remote/test‘} #这里就是所有要注册的module,本sample只有一个。 7 ]; 8 9 var port = 3333; 10 11 var server = Server.create({paths: paths, port: port}); 12 server.start(); 13 console.log(‘rpc server started.‘);
看出来pomelo里面的rpc使用封装的很漂亮。
其中11行 会调用到 lib/rpc-server/server.js文件的create函数,如下:
module.exports.create = function(opts) { if(!opts || !opts.port || opts.port < 0 || !opts.paths) { throw new Error(‘opts.port or opts.paths invalid.‘); } var services = loadRemoteServices(opts.paths, opts.context); opts.services = services; var gateway = Gateway.create(opts); return gateway; };
可以看出,实际上一个server就是一个gateway !!
这个gateway会接受 被loadRemoteServices生成的一系列services作为参数。
我们再看这个函数:
4 var loadRemoteServices = function(paths, context) { 5 var res = {}, item, m; 6 for(var i=0, l=paths.length; i<l; i++) { 7 item = paths[i]; 8 m = Loader.load(item.path, context); 9 10 if(m) { 11 createNamespace(item.namespace, res); 12 for(var s in m) { 13 res[item.namespace][s] = m[s]; 14 } 15 } 16 } 17 18 return res; 19 }; 20
21 var createNamespace = function(namespace, proxies) { 22 proxies[namespace] = proxies[namespace] || {}; 23 }; 24
其中load()前面一篇文章我们已经说过了,就是pomelo-loader。所以m就是一个module;
如前说述,所有的module生成工作和exports导出工作在load()函数已经完成。
第10 行后面的几行代码只是为了组织res而已,仔细看代码,或者通过debuger打印,可以看出。
>res Object user: Object service: Object echo: function (msg, cb) { ...pomelo会根据 初始化时候传入的namespace,js 模块的文件名,module中的exports名称来构建,
res[namespace][filename][exports.name] = m[exports.name];
这里就是一个树形的注册而已。等着被后面client来call。
而具体的call逻辑如下:
第二部分:调用,触发逻辑
上面第一部分已经注册了所有的module,当server.start()的时候,就是调用gateway.start(),进而调用accepter.listen();
这里的acceptor就是ws-accptor.js,对应部分代码如下:
pro.listen = function(port) { //check status if(!!this.inited) { utils.invokeCallback(this.cb, new Error(‘already inited.‘)); return; } this.inited = true; var self = this; this.server = sio.listen(port); # 1、第一步,调用socket.io的listen去监听网络。 this.server.set(‘log level‘, 0); this.server.server.on(‘error‘, function(err) { #错误处理,不说了。。 self.emit(‘error‘, err); }); this.server.sockets.on(‘connection‘, function(socket) { #2、注册client的链接事件; self.sockets[socket.id] = socket; self.emit(‘connection‘, {id: socket.id, ip: socket.handshake.address.address}); socket.on(‘message‘, function(pkg) { #3、在链接建立好以后,我们注册 “message”事件到ProcessMsgs()函数。 try { if(pkg instanceof Array) { processMsgs(socket, self, pkg); } else { processMsg(socket, self, pkg); } } catch(e) { // socke.io would broken if uncaugth the exception console.error(‘rpc server process message error: ‘ + e); } }); socket.on(‘disconnect‘, function(reason) { # 断开事件,不说了。。。 delete self.sockets[socket.id]; delete self.msgQueues[socket.id]; }); }); this.on(‘connection‘, ipFilter.bind(this)); #白名单,不说了。。。。 if(this.bufferMsg) { this._interval = setInterval(function() { flush(self); }, this.interval); } };
看上面代码中的注释,我们可以知道,一旦client有消息过来,我们就调用下面的函数:
var processMsg = function(socket, acceptor, pkg) { var tracer = new Tracer(acceptor.rpcLogger, acceptor.rpcDebugLog, pkg.remote, pkg.source, pkg.msg, pkg.traceId, pkg.seqId); tracer.info(‘server‘, __filename, ‘processMsg‘, ‘ws-acceptor receive message and try to process message‘); acceptor.cb.call(null, tracer, pkg.msg, function() { #####我们只看这个########################## var args = Array.prototype.slice.call(arguments, 0); for(var i=0, l=args.length; i<l; i++) { if(args[i] instanceof Error) { args[i] = cloneError(args[i]); } } 。。。。。。。。。。。。。。。。。。。。。。
接着上面说: 一旦client有消息来,我们就调用processMsg(),而processMsg()则调用acceptor的回调函数。也就是dispatcher.route()函数;
这个是在gateway初始化的时候挂在acceptor上面的。具体代码如下:
this.acceptor = this.acceptorFactory.create(opts, function(tracer, msg, cb) { dispatcher.route(tracer, msg, self.services, cb); });
我们再看,这个route()到底干啥,如下:
module.exports.route = function(tracer, msg, services, cb) { tracer.info(‘server‘, __filename, ‘route‘, ‘route messsage to appropriate service object‘); var namespace = services[msg.namespace]; var service = namespace[msg.service]; var method = service[msg.method]; var args = msg.args.slice(0); args.push(cb); method.apply(service, args); };上面代码去掉错误检测,看上去直观点。
就是:
methode = services[msg.namespace][msg.sevice][msg.method] 而已。
而这里的msg属性如下:
namespace :“user”
service:“service”
method:“echo”
看见了吧,就是上面第一部分注册的时候我们构建的树形结构嘛。
如果client调用的时候写的不匹配,则会在上面代码中错误检测部分被淘汰。
OK,完毕:
总结:
第一步:server构建一个树形的对应关系。
然后:根据client传递过来的并且被解析后的JSON来查找并且调用树形结构中对应的函数。
================下面看client,(坑)