pomelo广播的实现（chat例子分析）

转载自： http://blog.csdn.net/fjslovejhl/article/details/11703651

      其实最开始要读pomelo框架无非是因为自己没有读过什么node.js框架的源码，不过后来就逐渐变成了想要知道pomelo框架是如何实现广播的，貌似这也是游戏服务器比较重要的功能吧。。。。

一开始会觉得这种广播在分布式的环境下实现会比较的复杂。。但是当搞明白了pomelo的实现之后，发现它是采用了一种折中的方法实现广播。。虽然没有刚开始自己想的那么牛逼，不过觉得也算是一种比较好的解决方案吧。。

那么接下来就用pomelo给的chat这个例子来分析吧，来看登录吧，首先会向gate服务器发起连接：

function queryEntry(uid, callback) {
    var route = 'gate.gateHandler.queryEntry';
    pomelo.init({
        host: window.location.hostname,
        port: 3014,
        log: true
    }, function() {
        pomelo.request(route, {  //发起请求，用于获取用于连接的connector服务器的地址
            uid: uid
        }, function(data) {
            pomelo.disconnect();
            if(data.code === 500) {
                showError(LOGIN_ERROR);
                return;
            }
            callback(data.host, data.port);
        });
    });
};

这部分代码主要要完成的目的就是与gate进行通信，gate会返回该客户用于连接的connector服务器的地址，我们来看看gate服务器是怎么生成这个地址的吧：

 //next是一个函数，用于执行一些操作，将返回的数据发送回去
handler.queryEntry = function(msg, session, next) {
    var uid = msg.uid;
    if(!uid) {
        next(null, {
            code: 500
        });
        return;
    }
    // get all connectors
    var connectors = this.app.getServersByType('connector');  //获取素有connector服务器的配置信息
    if(!connectors || connectors.length === 0) {
        next(null, {  //第一个参error，第二个参数wie返回给客户端的信息
            code: 500
        });
        return;
    }
    // select connector
    var res = dispatcher.dispatch(uid, connectors);   //选取一个connector服务器
    next(null, {
        code: 200,
        host: res.host,
        port: res.clientPort
    });
};
var crc = require('crc');

module.exports.dispatch = function(uid, connectors) {
    var index = Math.abs(crc.crc32(uid)) % connectors.length;
    return connectors[index];
};

到这里就应该知道gate服务器是怎么挑选connector服务器的了吧。。。那么在获取了用于连接的connector之后，就应该建立与connector服务器的连接，进行登录了。。。代码如下：

//query entry of connection
queryEntry(username, function(host, port) {
    pomelo.init({
        host: host,  //这里是返回的用于连接的connector服务器的host与port
        port: port,
        log: true
    }, function() {
        var route = "connector.entryHandler.enter";  //这里可以当做是进行登录吧
        pomelo.request(route, {
            username: username,
            rid: rid
        }, function(data) {
            if(data.error) {
                showError(DUPLICATE_ERROR);
                return;
            }
            setName();
            setRoom();
            showChat();
            initUserList(data);
        });
    });
});

可以看到这里调用的是connector服务器的handler的enter方法，然后传过去的参数是username和rid（房间的id），那么我们来看看这个connector服务器的enter方法干了些什么事情吧：

handler.enter = function(msg, session, next) {
    var self = this;
    var rid = msg.rid;
    var uid = msg.username + '*' + rid  //用户名字还要加上组名字
    var sessionService = self.app.get('sessionService');

    //duplicate log in
    if( !! sessionService.getByUid(uid)) {  //表示有相同的用户了
        next(null, {
            code: 500,
            error: true
        });
        return;
    }

    session.bind(uid);  //将这个session与uid绑定起来
    session.set('rid', rid);
    session.push('rid', function(err) {
        if(err) {
            console.error('set rid for session service failed! error is : %j', err.stack);
        }
    });
    session.on('closed', onUserLeave.bind(null, self.app));  //设置closed事件的处理函数

    //put user into channel
    //这里session适用于挑选后台的chat服务器的，这里还要讲当前frontend服务器的serverID传送过去，因为后台要知道当前channel的用户都在哪些frontend服务器上面连接着
    //这里挑选后台的chat服务器的时候，用的是rid，所以可以保证同一个房间的人分到同一个chatserver
    self.app.rpc.chat.chatRemote.add(session, uid, self.app.get('serverId'), rid, true, function(users){
        next(null, {
            users:users  //远程服务器返回的当前channel里面的所有的用户
        });
    });
};

其实这部分的处理韩式很简单的，无非是处理一下从connector组件中分配的 session，设置一下rid，uid等基本的信息，最后有一个比较重要的操作，那就是进行chat的远程调用，在chat服务器中添加一个user， 这里上面的注释也已经很清楚了，其实到这里就已经知道pomelo是怎么实现广播的了，但是还是来看看究竟是怎么搞的吧。。。

那么我们来看看这个远程调用是怎么进行的，如果看过之前对pomelo框架proxy模块的分析，上面的实际上执行的是下面的方法：

      /*
{ namespace: 'sys',
    serverType: 'chat',
    path: '/home/fjs/Desktop/pomelo/game-server/node_modules/pomelo/lib/common/remote/backend/' },
*/
      proxyCB.call(null, serviceName, methodName, args, attach, invoke);  //调用proxyCB方法来处理数据

这里serviceName就是chatRemote，methodName是add，args就就是上面传进来的参数，attach就是这个远程调用的基本西溪，例如上面注释的那种形式，invoke可以忽略，那么我们在来看看proxyCB函数究竟干了些设么事情吧：

var proxyCB = function(client, serviceName, methodName, args, attach, invoke) {
  if(client.state !== STATE_STARTED) {
    throw new Error('[pomelo-rpc] fail to invoke rpc proxy for client is not running');
  }

  if(args.length < 2) {
    logger.error('[pomelo-rpc] invalid rpc invoke, arguments length less than 2, namespace: %j, serverType, %j, serviceName: %j, methodName: %j',
      attach.namespace, attach.serverType, serviceName, methodName);
    return;
  }

  var routeParam = args.shift(); //用于route的参数，一般情况下是session
  var cb = args.pop();  //用于处理返回消息的回调函数
  //其实msg也就是pomelo定义的远程方法调用的消息格式，远程服务器会根据这个消息来解析需要调用的方法名字等信息
  //namespace可以是sys和user，servicename是当前调用的js源码或者说模块的名字，method就是方法的名字，args为传给方法的参数
  var msg = {namespace: attach.namespace, serverType: attach.serverType,
    service: serviceName, method: methodName, args: args};
  // do rpc message route caculate
  var route, target;
  if(typeof client.router === 'function') {
    route = client.router;
    target = null;
  } else if(typeof client.router.route === 'function') {
    route = client.router.route;  //router函数，是用于在服务器中挑选一个，甚至可以理解为负载均衡吧
    target = client.router;
  } else {
    logger.error('[pomelo-rpc] invalid route function.');
    return;
  }

//这里调用route函数获取serverID，想这个server发送消息
  route.call(target, routeParam, msg, client._routeContext, function(err, serverId) {
    if(err) {
      utils.invokeCallback(cb, err, serverId);
      return;
    }

    client.rpcInvoke(serverId, msg, cb);
  });
};

其实这里之所以想要再将整个rpc的过程又弄出来，主要就是想要证明一个东西，那就是 同一个房间的用于将会被分配到同一个后台chat服务器，在这里我们可以看到一个route函数，不知道大家是否记得在application的时候的一 段代码：app.route('chat', routeUtil.chat);   //chat是server类型，第二个是route函数

这里就会为挑选后台的chat服务器提供一个route函数，那么将在这里使用，那么我们在这里来看看这个函数是怎么定义的吧：

var exp = module.exports;
var dispatcher = require('./dispatcher');

exp.chat = function(session, msg, app, cb) {
    var chatServers = app.getServersByType('chat');

    if(!chatServers || chatServers.length === 0) {
        cb(new Error('can not find chat servers.'));
        return;
    }

    var res = dispatcher.dispatch(session.get('rid'), chatServers);  //这里可以保证相同的rid最后访问的是同一个chat服务器

    cb(null, res.id);
};

这里将会会dispatch函数传入rid，也就是房间的id，这也就能够知道为什么同一个房间的用于将会被分配到同一个chat服务器了吧。。。好了那么这里对rpc的说明就到此了吧，那么接下来来看看调用的chat服务器的add方法究竟干了些什么事情吧：

//当有用户进来的时候会调用这个方法
//这里uid是用户的id，sid是前端的connector服务器id，那么是房间的id，
//由于这里是远程的rpc调用访问的方法，cb是用于将执行结果返回过rpc客户端
hatRemote.prototype.add = function(uid, sid, name, flag, cb) {
var channel = this.channelService.getChannel(name, flag);  //这里的flag表示如果没有这个channel的时候要创建这个channel
var username = uid.split('*')[0];
var param = {
    route: 'onAdd',
    user: username
};
channel.pushMessage(param);  //想这个channel广播消息，表示当前有用户加入了channel

if( !! channel) {
    channel.add(uid, sid);  // 在这个channel中添加一个人，这里还将前段的connector服务器的serverid也传进去了
}

cb(this.get(name, flag)); //获取当前channel所有的用户，返回回去
;

这部分其实代码一看就基本上就能明白的差不多吧，无非是根据房间的名字来获取这个房间 的channel，然后再想这个channel广播有用户加进来的消息，接着还要在这个channel中设置新的用户，并且还要讲当前channel中所 有的用户返回，，，。。那么这里涉及到广播消息的就是channel.pushMessage(param);

但是在分析这个广播消息的方法之前，我们先来看看channel中是如何添加用户的吧，也就是channel.add(uid, sid);  // 在这个channel中添加一个人，这里还将前段的connector服务器的serverid也传进去了  

 //在当前channel中添加一个新的user，uid是username*rid  ，sid就是这个user所属的前端connector服务器的id
Channel.prototype.add = function(uid, sid) {
  if(this.state > ST_INITED) {
    return false;
  } else {
    //这里add说白了就是为了记录当前的前端connector服务器总所属于当前channel的user
    var res = add(uid, sid, this.groups);  //用于添加用户，这里groups是用于记录当前前端connector服务器属于当前channel的所有的user
    if(res) {
      this.records[uid] = {sid: sid, uid: uid};  //相当于是记录当前user的前端服务器
    }
    return res;
  }
};

/**

其实这个函数要执行的工作就两个：

（1）记录当前user的前端connector服务器

（2）记录这个前端connector服务器的user

到这里应该就更能够明白pomelo是怎么进行广播的了吧

那么接下来我们还是来看看这个广播方法究竟是怎么进行的广播的吧

//将数据发送给这个channel的所有用户
hannel.prototype.pushMessage = function(route, msg, cb) {
 if(this.state !== ST_INITED) {
   utils.invokeCallback(new Error('channel is not running now'));
   return;
 }

 if(typeof route !== 'string') {
   cb = msg;
   msg = route;
   route = msg.route;
 }
/这里group是保存了所有有当前用户的前端connector服务器
 sendMessageByGroup(this.__channelService__, route, msg, this.groups, cb);
;

好像没什么意思吧，那么继续来看这个sendMessageByGroup方法吧：

 //这个函数用于向group的所有的用户发送消息，这里group保存的数据格式是
 //key：前端的connector服务器的serverID
 //value：[]，一个数组保存这个服务器中要接受数据的user
var sendMessageByGroup = function(channelService, route, msg, groups, cb) {
  var app = channelService.app;
  var namespace = 'sys';  //这里是进行rpc的参数
  var service = 'channelRemote';  //服务
  var method = 'pushMessage';  //方法
  var count = utils.size(groups);
  var successFlag = false;
  var failIds = [];

  if(count === 0) {
    // group is empty
    utils.invokeCallback(cb);
    return;
  }

  var latch = countDownLatch.createCountDownLatch(count, function(){
    if(!successFlag) {
      utils.invokeCallback(cb, new Error('all uids push message fail'));
      return;
    }
    utils.invokeCallback(cb, null, failIds);
  });

  var rpcCB = function(err, fails) {
    if(err) {
      logger.error('[pushMessage] fail to dispatch msg, err:' + err.stack);
      latch.done();
      return;
    }
    if(fails) {
      failIds = failIds.concat(fails);
    }
    successFlag = true;
    latch.done();
  };

  var group;
  for(var sid in groups) {
    group = groups[sid]; //当前server要接受数据的用户
    if(group && group.length > 0) {
      //向相应的服务器发送rpc消息，将数据发送给对应的用户
      //挨个向所有的服务器发送消息
      app.rpcInvoke(sid, {namespace: namespace, service: service,
        method: method, args: [route, msg, groups[sid]]}, rpcCB);
    } else {
      // empty group
      process.nextTick(rpcCB);
    }
  }
};

其实上面的方法无非就是调用前端connector服务器的方法，让他们将数据发送给最终的用户，那么到这里后端的chat服务器要做的事情就差不多了，工作又回到了前端的connector服务器，那么又来看看吧：

//uid是应该接受数据的user，msg就是要发送的数据，route就是route
emote.prototype.pushMessage = function(route, msg, uids, cb) {
 if(!msg){
   logger.error('Can not send empty message! route : %j, compressed msg : %j',
       route, msg);
   return;
 }

 var connector = this.app.components.__connector__;

 var sessionService = this.app.get('sessionService');
 var fails = [], sids = [], sessions, j, k;
 for(var i=0, l=uids.length; i<l; i++) {
   sessions = sessionService.getByUid(uids[i]);  //获取这个用户的session
   if(!sessions) {  //如果么有session，那么发送失败
     fails.push(uids[i]);
   } else {
     for(j=0, k=sessions.length; j<k; j++) {
       sids.push(sessions[j].id);  //这里session的id其实就是connector组件中socket的id,其实这里session就已经有send方法了，为什么不调用？
     }
   }
 }

 connector.send(null, route, msg, sids, {isPush: true}, function(err) {  //调用connector的send方法将数据发送给刚刚弄出来的socket
   cb(err, fails);
 });
;

哈，到这里整个广播的过程我想基本上也就弄的比较的清楚了。。。最后再用一张图来总结一下吧：

好了，到这里就基本上搞清楚了我最开始想要搞清楚的问题。。那么整体上pomelo框架的内容就差不太多了，可能还有一些模块要做的事情我没有细看，不过也无所谓吧。。以后如果真有机会用pomelo框架的时候再去看也不迟，只要搞清楚了pomelo框架总体的脉络。。。

其实这里还有一种高并发网络系统的设计思想，叫做连接的离散化。。。这里将用户的连接 分不到不同的connector服务器上，但是如果他们属于同一个房间，则他们访问的依然是同一个后台chat服务器，那么也就做到了职责的分离，前台的 connector服务器主要负责维护用户的连接，用于尽可能多的连接最终的用户，而后台的chat服务器就专职处理一些业务逻辑，只需要维护较少的与前 端connector服务器的rpc连接就可以了。。。