pomelo代码分析7———— pomelo的rpc实现分析

想哪写哪,前面的坑如果不大碍事,咱就不填了。嘿嘿~


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.js
OK,从文件上看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,(坑)



pomelo代码分析7———— pomelo的rpc实现分析

上一篇:ACM一些杂题2


下一篇:linux查看文件的几种方式