pomelo分布式聊天服务器详解

标签： pomelo 分布聊天 | 发表时间：2014-01-05 11:43 | 作者：snoopyxdy

出处：http://snoopyxdy.blog.163.com

说来也惭愧，知道pomelo框架已经一年有余了，最近因为有开发IM的需求，但却是第一次部署安装pomelo框架，对不起网易开发团队的朋友~

pomelo的wiki上有一个分布式chat聊天室的例子，开发团队写的很仔细，详细对比了传统单进程聊天服务器的弊端，并给出pomelo框架分布式聊天服务器的优势，相关wiki地址如下：

tutorial1 分布式聊天

部署这个聊天demo非常简单，去github上下载这个聊天室的源代码，然后根据wiki里的程序安装依赖，并且分别启动pomelo的game server和web server。

代码下载地址：

https://github.com/NetEase/chatofpomelo

我刚运行这个聊天室程序的时候确实有点迷糊，看了wiki上的架构图又是web server，又是gate server，还有多个connecter，还有chat server等等，在config .json里可以进行相关的一些配置：

"development":{

"connector":[

{"id":"connector-server-1", "host":"127.0.0.1", "port":4050, "clientPort": 3050, "frontend": true},

{"id":"connector-server-2", "host":"127.0.0.1", "port":4051, "clientPort": 3051, "frontend": true},

{"id":"connector-server-3", "host":"127.0.0.1", "port":4052, "clientPort": 3052, "frontend": true}

"chat":[

{"id":"chat-server-1", "host":"127.0.0.1", "port":6050},

{"id":"chat-server-2", "host":"127.0.0.1", "port":6051},

{"id":"chat-server-3", "host":"127.0.0.1", "port":6052}

"gate":[

{"id": "gate-server-1", "host": "127.0.0.1", "clientPort": 3014, "frontend": true}

]

其中frontend表示此服务器可以被用户请求到，clientPort表示此服务器对外的端口号，port表示此服务器对内部的rpc调用端口号。

启动好服务器之后，我们在浏览器地址栏中输入:http://127.0.0.1;3001/index.html就可以正常登录进行聊天了。从前端入手，我们先简单看一下前端页面的js代码，在web-server的public文件夹中存放了前端用到的html和js代码。

client.js就是整个聊天室用到的前端js代码，它的结构如下：

1、定义了很多用到的变量

2、定义了用来判断输入合法性的util对象

3、定义很多操作dom元素的function函数

4、定义queryEntry方法，这个方法比较重要，下面单独说明

5、定义很多事件，用来接收pomelo服务器响应的东西

6、对login按钮进行绑定click事件

7、对发送消息entry按钮绑定click事件

我们单独看下queryEntry方法，代码如下：

    // query connector     
function queryEntry(uid, callback) {     
	var route = 'gate.gateHandler.queryEntry';     
	pomelo.init({     
		host: window.location.hostname,     
		port: 3014,     
		log: true     
	}, function() {     
		pomelo.request(route, {     
			uid: uid     
		}, function(data) {     
			pomelo.disconnect();     
			if(data.code === 500) {     
				showError(LOGIN_ERROR);     
				return;     
			}     
			callback(data.host, data.port);     
		});     
	});     
};

其中我们看到pomelo.init方法，传入了host，port和log的参数，同时在回调函数里面使用pomelo.request方法将uid发送出去，在回调函数里断开连接，最后执行callback，将返回的data数据的host和port传入callback。

反正我第一次看这段代码是一头雾水，这个queryEntry函数是在用户点击登录之后执行的，我们打开public/js/lib/pomeloclient.js文件，找到init函数，代码如下：

    
    pomelo.init = function(params, cb){
        pomelo.params = params;
        params.debug = true;
        var host = params.host;
        var port = params.port;
         
        var url = 'ws://' + host;
        if(port) {
          url +=  ':' + port;
        }
         
        socket = io.connect(url, {'force new connection': true, reconnect: false});
         
        socket.on('connect', function(){
          console.log('[pomeloclient.init] websocket connected!');
          if (cb) {
            cb(socket);
          }
        });
         
        socket.on('reconnect', function() {
          console.log('reconnect');
        });
         
        socket.on('message', function(data){
          if(typeof data === 'string') {
            data = JSON.parse(data);
          }
          if(data instanceof Array) {
            processMessageBatch(pomelo, data);
          } else {
            processMessage(pomelo, data);
          }
        });
         
        socket.on('error', function(err) {
          console.log(err);
        });
         
        socket.on('disconnect', function(reason) {
          pomelo.emit('disconnect', reason);
        });
      };

其实上述代码就是利用socket.io于远程服务器建立连接，并且把socket对象传入回调函数。另外pomelo.request方法就是向这个socket发送数据，注意了整个pomelo对象是一个单例，所以我们在使用pomelo对象时同时只能连接一个服务器，所以代码中在连接gate服务器之后，获得connector服务器的主机名和端口就需要使用pomelo.disconnect();方法关闭这个连接，从而重新init连接被分配的connector服务器。

我们重点看下，这个route变量：

    var route = 'gate.gateHandler.queryEntry';

这个地址就代表着gate服务器的方法地址，其中gateHandler表示文件名，queryEntry表示exports对外的方法名，通过前端的如下代码：

   
   pomelo.request(route, {
			uid: uid   
		},function(){..

我们就把uid发送到了gate服务器中的handler文件夹中，gateHandler.js这个文件里的queryEntry方法中了。在queryEntry方法中，其实什么事情都没有去做，只不过将用户uid根据哈希算法分配到一台connector服务器，gate并不会去做路由转发，而是直接返回给客户端connector的host和port，所以我们就看到了上述代码中前端关闭与gate服务器的连接，将收到的信息host和port传给callback函数了。

通过上述这些代码，我们基本了解到web服务器主要就是用来展现静态资源的，把他换成nginx或者apache都可以。而gate服务器也是独立与系统的，它的作用也不过是根据用户名来哈希计算分配给这个客户端的connector地址。

1、用户登录登出过程：

接下来我们看下，用户第一次进入页面，点击登录按钮发生了什么？代码如下：

    
    queryEntry(username, function(host, port) {
         			pomelo.init({
     				host: host,
     				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);
     				});
     			});
     		});

queryEntry函数我们之前已经分析过了，回调函数接收的host和port值就是gate服务器分配的connector地址，我们使用同样的pomelo.init方法连接上connector服务器，然后调用远程地址"connector.entryHandler.enter"，将rid和username传给这个方法，当远程执行完毕之后，让此用户进入聊天室。这里我们打开connector文件夹下的entryHandler.js，查看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.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));

	//put user into channel
	self.app.rpc.chat.chatRemote.add(session, uid, self.app.get('serverId'), rid, true, function(users){
		next(null, {
			users:users
		});
	});
};

这里用到的pomelo的api比较多，我们逐一解释，

首先我们先获取session服务

    var sessionService = self.app.get('sessionService');

然后通过下面的代码，判断这个用户是否已经存在了，如果已经存在那么就要返回error错误

    
    if( !! sessionService.getByUid(uid)) {
    		next(null, {
    			code: 500,
    			error: true
    		});
    		return;
    	}

下面的代码是绑定用户uid到session中，并且将这个uid更新房间rid的session，然后利用push方法下发同步session，当session触发关闭事件后，就执行onUserLeave方法，并且绑定它的第一个参数是app

    
    session.bind(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));

这天通过app的rpc远程调用chatRemote.js的add方法，将一些参数传入，等待远程返回users对象，然后将users返回给客户端。

     
     self.app.rpc.chat.chatRemote.add(session, uid, self.app.get('serverId'), rid, true, function(users){
     		next(null, {
     			users:users
     		});
     	});

最后是用户离开的函数，远程调用kick方法，将用户剔除。

     
     var onUserLeave = function(app, session) {
     	if(!session || !session.uid) {
     		return;
     	}
     	app.rpc.chat.chatRemote.kick(session, session.uid, app.get('serverId'), session.get('rid'), null);
     };

connector服务器的代码分析完了，主要作用就是将session绑定用户id，同时同步和下发session到chatserver中去，让chatserver在处理聊天的时候可以获取到用户身份。

接下来打开chat/remote/chatRemote.js文件，看下add和kick方法是怎么定义的。

先定义一个ChatRemote类，通过app.get获取'channelService'服务，这个上面的sessionService一样，拿到channelService对象之后，我们调用this.channelService.getChannel(channel_name,flag)，获取一个指定频道，通过查看pomelo的api文档我们可知，第二个参数flag如果为true，如果没查找到这个channel，那么就会去创建这个channel。

然后通过channel.pushMessage(param);方法向这个频道的所用客户端广播，这将触发client.js的onAdd事件，同时将用户名作为参数传入。

channel.add(uid, sid);这里将新登录的用户uid和connector_server_id添加到此频道中去。然后通过将this.get方法的返回值作为参数，传给回调函数函数。

     
     var ChatRemote = function(app) {      
	this.app = app;      
	this.channelService = app.get('channelService');      
};
     ChatRemote.prototype.add = function(uid, sid, name, flag, cb) {
     	var channel = this.channelService.getChannel(name, flag);
     	var username = uid.split('*')[0];
     	var param = {
     		route: 'onAdd',
     		user: username
     	};
     	channel.pushMessage(param);
           
     	if( !! channel) {
     		channel.add(uid, sid);
     	}
           
     	cb(this.get(name, flag));
     };

我们看一下this.get函数做了什么事情，他的功能就是获取这个频道下面所有用户的uid数组

             ChatRemote.prototype.get = function(name, flag) {
       	var users = [];
       	var channel = this.channelService.getChannel(name, flag);
       	if( !! channel) {
       		users = channel.getMembers();
       	}
       	for(var i = 0; i < users.length; i++) {
       		users[i] = users[i].split('*')[0];
       	}
       	return users;
       };

我们通过connector的next函数，将这个用户数组传递给前端的回调函数执行，这样就将用户uid的列表正常返回给前端的client.js了。

另外一个kick的方法比较简单，主要就是将用户id从channel中剔除，然后触发用户的onLeave事件，告知这个channel中的用户此uid已经离开了。

     
     ChatRemote.prototype.kick = function(uid, sid, name, cb) {
     	var channel = this.channelService.getChannel(name, false);
     	// leave channel
     	if( !! channel) {
     		channel.leave(uid, sid);
     	}
     	var username = uid.split('*')[0];
     	var param = {
     		route: 'onLeave',
     		user: username
     	};
     	channel.pushMessage(param);
     	cb();
     };

至此，我们对用户进入聊天室和登出聊天室的功能已经有所了解了，下面我们要分析一下用户发送消息的广播和单播功能的实现。

2、消息广播和单播实现

我们还是打开public/client.js文件，找到用户发送消息的代码，如下：

代码中先定义了chat.chatHandler.send，这将直接使前端通过rpc调用chatHandler.js中的send方法。代码中还加入了一些合法性验证和去除空格的东西，核心代码是pomelo.request这段，前端将rid（频道名），content（消息内容），from（发送方用户id），target（接收方）作为参数传入，当服务器端处理完毕执行回调之后，我们通过addMessage函数将信息打印到网页上，其实后面那段$("#chatHistory").show();完全可以放在addMessage这个方法里面去，因为它本来就是addMessage的一个过程。

     
     //deal with chat mode.
     	$("#entry").keypress(function(e) {
     		var route = "chat.chatHandler.send";
     		var target = $("#usersList").val();
     		if(e.keyCode != 13 /* Return */ ) return;
     		var msg = $("#entry").attr("value").replace("\n", "");
     		if(!util.isBlank(msg)) {
     			pomelo.request(route, {
     				rid: rid,
     				content: msg,
     				from: username,
     				target: target
     			}, function(data) {
     				$("#entry").attr("value", ""); // clear the entry field.
     				if(target != '*' && target != username) {
     					addMessage(username, target, msg);
     					$("#chatHistory").show();
     				}
     			});
     		}
     	});

前端代码不处理任何逻辑，我们看下被远程rpc调用的chatHandler.js中的send方法是如何处理聊天消息的。

先通过session获得当前发送消息的用户的信息，然后调用channelService服务，获取频道对象，判断如果target是*，那就代表频道广播，直接channel.pushMessage将消息广播，触发前端client.js的onChat方法。

如果target是指定的uid，表示单播，我们先拼接目标用户id，然后根据我们之前保存的frontend的serverid拿到sid，最后我们通过pushMessageByUids将消息给指定的用户单播推送出去，注意这里不能使用channel对象而是使用channelService。

     handler.send = function(msg, session, next) {
	var rid = session.get('rid');
	var username = session.uid.split('*')[0];
	var channelService = this.app.get('channelService');
	var param = {
		route: 'onChat',
		msg: msg.content,
		from: username,
		target: msg.target
	};
	channel = channelService.getChannel(rid, false);

	//the target is all users
	if(msg.target == '*') {
		channel.pushMessage(param);
	}
	//the target is specific user
	else {
		var tuid = msg.target + '*' + rid;
		var tsid = channel.getMember(tuid)['sid'];
		channelService.pushMessageByUids(param, [{
			uid: tuid,
			sid: tsid
		}]);
	}
	next(null, {
		route: msg.route
	});
};

我们通过前端监听的onChat事件，将受到的消息放置在网页中，tip表示消息提醒功能。

     pomelo.on('onChat', function(data) {
		addMessage(data.from, data.target, data.msg);
		$("#chatHistory").show();
		if(data.from !== username)
			tip('message', data.from);
	});

这样我们整个的聊天室群聊和单聊功能都已经开发完毕了，不过在看这些源码过程中还是碰到一些疑问的，可能需要去翻pomelo源码才能解决，总体感觉pomelo框架的文档不够详细，上手教程也不够详细，很多api不知道怎么用法，估计真正投入生产还是要把pomelo框架的源代码翻个遍才能得心应手的使用。

看完聊天室的代码，给我几个有疑惑的地方，等接下来深入pomelo框架之后，应该会有所解答：

1、用户的session和channel信息的保存，默认应该是保存在内存中的，如何把它保存到数据库中

2、session和channel的同步效率如何，目前还没测试过

3、对于connector或者chatserver的容灾问题，demo中也没考虑

4、gateserver理论上是可以支持分布式扩展的吧

5、如果connector和chat还有gate不在一台服务器上的话怎么处理？如何分别启动这些服务器和同步下发config？

pomelo分布式聊天服务器详解

相关 [pomelo 分布聊天] 推荐：