对于感兴趣的人，
这是我在最新的 socket.io 上运行的原型聊天服务器，以 express、多核和 redis 作为中间体。
广播消息将发送给所有房间用户，无论他们是否连接到不同的节点和端口实例。
跑吧
在其他机器上
- 节点客户端.js
- 或用于测试节点client.js 7001、7002、7003 ......
server.js
var options = {
//workers: 2, // total workers (default: cpu cores count).
first_port: 7000, // 8000, 8001 are worker's ports (default: 8000).
proxy_port: 5000, // default (5000).
session_hash: function (req, res) { return req.connection.remoteAddress; },
no_sockets: false // allow socket.io proxy (default: false).
};
require('sticky-socket-cluster')(options, start);
function start(port) {
// requirements
var express = require('express');
var http = require('http');
var socketio = require('socket.io');
var path = require('path');
var sticky = require('sticky-session');
var app = express();
var server = http.createServer(app);
var io = socketio.listen(server);
var redis = require('socket.io-redis');
io.adapter(redis({ host: 'localhost', port: 6379 }));
server.listen(port, function() {
console.log(' - listening on ' + port+ ' ' + __dirname);
});
// require our chatserver
var ChatServer = require('./chatserver');
// initialize a new chat server.
new ChatServer({io: io, port: port}).init();
}
聊天服务器.js
RoomUtil = (function(){
roomMessages = {};
return {
getMessages : function(room_id,limit,cb){
//TODO
cb(roomMessages[room_id] || []);
},
postMessage : function(message,room_id,cb){
if (!roomMessages[room_id]) roomMessages[room_id] = [];
roomMessages[room_id].push(message);
cb();
}
}
})();
var Server = function(options) {
var self = this;
self.io = options.io;
// users array
self.users = [];
// initialize function
self.init = function() {
console.log("init");
// Fired upon a connection
self.io.on('connection', function(socket) {
console.log("incoming connection");
// var ru = new RoomUser();
self.handleConnection(socket,options.port);
});
}
// socket handler for an incoming socket
self.handleConnection = function(socket,port) {
// wait for a login message
socket.emit("incoming connection",{});
socket.on("joinroom",function(data,joinroom_callback){
console.log("attempt to join room ",data.room_id," on port ",port);
if (!data.room_id){
console.log("cannon join room -> no room id given");
return socket.disconnect();
}
else{
var room_id = data.room_id;
socket.join(room_id,function(){
console.log(socket.rooms);
RoomUtil.getMessages(data.room_id,50,function(messages){
console.log("client succesfully joined room ",data.room_id);
joinroom_callback(null,{'messages':messages});
});
socket.on("login",function(data,login_callback){
if (!data.username){
login_callback("invalid userdata",null);
}
else{
login_callback(null,1);
socket.on("post_message",function(data,message_callback){
if (!data.message || data.message == ""){
console.log("empty message posted. ignore");
message_callback("invalid_message",null);
}
else{
console.log("received message on port ",port,data.message);
message_callback(null,1);
RoomUtil.postMessage(data.message,room_id,function(){
RoomUtil.getMessages(room_id,50,function(messages){
console.log("emit messages to room id ",room_id);
//socket.to(room_id).emit('update_messages', messages);
//socket.broadcast.to(room_id).emit('update_messages', messages);
//socket.broadcast.to(room_id).emit('update_messages', messages);
//self.io.to(room_id).emit('update_messages', messages);
self.io.in(room_id).emit('update_messages', messages);
});
})
}
});
}
});
});
}
});
}
}
module.exports = Server;
client.js
var servercon = 'http://localhost:'+(process.argv[2] || 5000);
console.log("going to connect to "+servercon)
var socket = require('socket.io-client')(servercon);
var readline = require('readline');
var rl = readline.createInterface({
input: process.stdin,
output: process.stdout
});
socket.on('connect', function(){
console.log("connected, going to login");
socket.emit("joinroom",{"room_id":123123}, function(error,data){
if (error){
console.log("cannot join room ",error);
}
else{
console.log("succesfully joined room -> going to login now");
console.log("received messages count",data.messages.length);
socket.emit("login",{username:"John Mckain"}, function(error, message){
if (error){
console.log("error logging in ",error);
}
else{
console.log("logged in succesfully -> post message now");
var readline = function(){
rl.question("type in a message -> ", function(message) {
socket.emit("post_message",{'message':message}, function(error, message){
if (error){
console.log("error posting message");
readline();
}
else{
console.log("succesfully posted message");
readline();
}
});
});
}
readline();
}
});
socket.on("update_messages",function(data){
console.log("received new messages count ",data.length,data);
});
}
});
});
socket.on('event', function(data){
console.log("event send",data);
});
socket.on('disconnect', function(e){
console.log("disconnected",e);
});
socket.on("welcome",function(data){
console.log("on welcome ",data)
})
socket.on("pong",function(e){
console.log("pong")
})