码迷，mamicode.com
首页 > Web开发 > 详细

使用Nodejs实现聊天服务器

时间：2015-12-29 15:56:16      阅读：307      评论：0      收藏：0      [点我收藏+]

标签：

两年前在项目中使用nodejs+socket.io+redis实现的聊天和推送服务器,
基本上几百行代码就实现了整个功能，在项目中单服务器单进程可以跑到支持
5000人左右同时在线。

主要思路

  1. 用户上线后，根据用户的userid和socket，保存到一个全局的map中
  2. 发送消息时，根据对方的userid找到对应的socket，通过socket.emit发送消息给对方
  3. 如果对方不在线，在redis中保留消息，在对方上线后推送给用户
  4. 用户下线后，从全局的map中删除对应的用户socket
  5. 由于需要保持长连接，客户端需要定时发心跳给服务端，所以定义了一个心跳消息，可以5分钟发一次

  • 用户上线

    io.sockets.on(‘connection‘, function (socket) {
        var address = socket.handshake.address;
        console.log(Date() + " New connection from " + address.address + ":" + address.port);
        socket.on(‘login‘, function (userinfo) {
            userid = userinfo.myAuraId
            var address = socket.handshake.address;
            var deviceid = userinfo.deviceId
            console.log(Date() + " Login from " + address.address + ":" + address.port + " " + userid + " " + deviceid);
            old_socket = sockets[userid]
            if (old_socket && old_socket.deviceid && deviceid && old_socket.deviceid != deviceid) {
                old_socket.relogin = 1
                old_socket.emit(‘logout‘)
                console.log("logout " + old_socket.userid + " " + old_socket.deviceid)
            }

            if (old_socket && old_socket != socket) {
                old_socket.disconnect()
            }

            socket.relogin = 0
            socket.userid = userid
            socket.deviceid = deviceid
    		//发送离线消息
            send_store_msg(socket, userid)

            sockets[userid] = socket
            
            //通知业务服务器，用户已登录
            pub.publish("login_message_channel", JSON.stringify(userinfo))

        })

  • 发送消息

    socket.on(‘chat‘, function (msg, ack) {
        //process_msg(msg)
        pub.publish("chat_filter_channel", JSON.stringify(msg))
        socket.userid = msg.from
        sockets[socket.userid] = socket
        if (ack) {
            ack(1)
        }

    })

  • 接收和回应心跳

    socket.on(‘hb‘, function (msg, ack) {
        if (ack) {
            ack(1)
        }
    })

  • 用户下线，删除对应的socket

    socket.on("disconnect", function () {
        var address = socket.handshake.address;
        console.log(Date() + " Disconnect from " + address.address + ":" + address.port);
        if (!socket.relogin) {
            delete sockets[socket.userid]
        }
    })

推送服务器

实现了聊天服务器后，对推送来说就很简单了

  1. 在redis里开个channel，业务服务器往这个channel里publish数据
  2. nodejs subscribe这个channel监听数据，找到对应用户发送消息即可。
  3. 用户不在线，可能也需要对这个离线推送消息做保留，具体看业务定义
var notification = redis.createClient()
notification.subscribe("notification")

// check redis notifcation channel
notification.on("message", function (pattern, msg) {
    var msgobj = JSON.parse(msg)
    var keys = msgobj.toWho
    var needStore = msgobj.needStore
    for (index in keys) {
        var key = keys[index]
        if (!needStore) {
            if (sockets[key]) {
                sockets[key].emit("notification", msg)
            }
        }
        else {
            var list = []
            store.hget("nodejs_notification", key, function (e, v) {
                if (v) {
                    list = JSON.parse(v);
                }
                list.push(msg)
                var msglist = JSON.stringify(list)
                store.hset("nodejs_notification", key, msglist, function (e, r) {
                })
            })

            if (sockets[key]) {
                send_notification(sockets[key], msg)
            }

        }
    }
})

function send_notification(socket, notif) {
    socket.emit("notification", notif, function ack() {
        store.hdel("nodejs_notification", socket.userid)
    })
}

客户端Lib库

服务端是使用socket.io实现，基本上socket.io的Lib都能兼容

以下是推荐的两个客户端Lib：

其他语言版本，可以在github搜索socket.io,找到对应的Lib库

Server端全部代码

//var io = require(‘socket.io‘).listen(80)
var app = require(‘http‘).createServer(handler)
    , io = require(‘socket.io‘).listen(app)

app.listen(80);

function handler(req, res) {
    if (req.url == "/monitor") {
        res.writeHead(200);
        res.end("OK");
    }
    else {
        res.writeHead(404);
        res.end();
    }
}


//io.disable(‘heartbeats‘)
//io.set(‘heartbeats‘, false);
io.set(‘transports‘, [‘websocket‘, ‘xhr-polling‘]);
io.set(‘heartbeat timeout‘, 5 * 60)
io.set(‘heartbeat interval‘, 4 * 60)
io.set(‘close timeout‘, 1 * 30);
io.set("log level", 1)
io.set("browser client", false)
io.set("browser client cache", false)
io.set("browser client cache", false)
var redis = require("redis")
var pub = redis.createClient()
var store = redis.createClient()
var snschat = redis.createClient()
var notification = redis.createClient()
var PUSH_TO_IOS_DELAY_TIME = 120000
snschat.subscribe("snschat");
notification.subscribe("notification")
var sockets = {}

pub.on("error", function (err) {
    console.log("Error " + err);
});

store.on("error", function (err) {
    console.log("Error " + err);
});

snschat.on("error", function (err) {
    console.log("Error " + err);
});


function send_msg_delay(socket) {
    store.hget("chat_history", socket.userid, function (e, v) {
        if (v) {
            list = JSON.parse(v);
            if (list.length > 0) {
                var msg = JSON.stringify(list)
                socket.isSendingChatMessage = false
                send_msg(socket, msg)
            }
        }
    })
}

function send_msg(socket, msg) {
    //delay for 5 sec
    if (socket.isSendingChatMessage) {
        setTimeout(function () {
            send_msg_delay(socket)
        }, 5000)
        return
    }
    socket.isSendingChatMessage = true

    //start send
    var callSendToIOS = sendToIOSDealy(socket.userid, PUSH_TO_IOS_DELAY_TIME)
    socket.emit("chat", msg, function ack(size) {
        clearTimeout(callSendToIOS)
        store.hget("chat_history", socket.userid, function (e, v) {
            if (v) {
                list = JSON.parse(v);
                //console.log("size="+size)
                if (list.length == size) {
                    store.hdel("chat_history", socket.userid, function (e, r) {
                    })
                }
                else if (size < list.length) {
                    list = list.splice(size)
                    var msglist = JSON.stringify(list)
                    store.hset("chat_history", socket.userid, msglist, function (e, r) {
                    })
                }
            }
            socket.isSendingChatMessage = false

        })
    })
}


function sendToIOSDealy(toWho, time) {
    return setTimeout(function () {
        sendToIOS(toWho)
    }, time)
}

function sendToIOS(toWho) {
    var obj = {"toWho": toWho}
    var msg = JSON.stringify(obj)
    console.log("delay send to ios channel:" + msg)
    pub.publish("chat_message_channel", msg)
}

function send_notification(socket, notif) {
    socket.emit("notification", notif, function ack() {
        store.hdel("nodejs_notification", socket.userid)
    })
}

function send_store_msg(socket, userid) {

    if (socket.isSendStoreMsg) {
        return;
    }

    socket.isSendingChatMessage = false

    store.hget("chat_history", userid, function (e, msg) {
        if (msg) {
            send_msg(socket, msg)
            store.hdel("chat_history", socket.userid, function (e, r) {
            })
        }
    })

    store.hget("nodejs_notification", userid, function (e, msg) {
        if (msg) {
            var msglist = JSON.parse(msg)
            for (var i = 0; i < msglist.length; i++) {
                send_notification(socket, msglist[i])
            }
            //socket.emit("notification", msg)
            //store.hdel("nodejs_notification", userid)
        }
    })
    socket.isSendStoreMsg = true
}

function saveToChatHistory(msg) {
    var list = []
    store.hget("chat_history", msg.to, function (e, v) {
        if (v) {
            list = JSON.parse(v);
        }
        list.push(msg)
        var msglist = JSON.stringify(list)
        store.hset("chat_history", msg.to, msglist, function (e, r) {
        })
    })
}


function pushToChatHistoryChannel(msg) {
    var msgStr = JSON.stringify(msg)
    pub.publish("chat_message_history_channel", msgStr)
}

function process_msg(msg) {
    var list = []
    store.hget("chat_history", msg.to, function (e, v) {
        if (v) {
            list = JSON.parse(v);
        }
        list.push(msg)
        var msglist = JSON.stringify(list)
        store.hset("chat_history", msg.to, msglist, function (e, r) {
        })
        if (sockets[msg.to]) {
            send_msg(sockets[msg.to], msglist)
        }
        else {
            sendToIOS(msg.to)
        }
        pushToChatHistoryChannel(msg)
    })
}

// check redis notifcation channel
notification.on("message", function (pattern, msg) {
    var msgobj = JSON.parse(msg)
    var keys = msgobj.toWho
    var needStore = msgobj.needStore
    for (index in keys) {
        var key = keys[index]
        if (!needStore) {
            if (sockets[key]) {
                sockets[key].emit("notification", msg)
            }
        }
        else {
            var list = []
            store.hget("nodejs_notification", key, function (e, v) {
                if (v) {
                    list = JSON.parse(v);
                }
                list.push(msg)
                var msglist = JSON.stringify(list)
                store.hset("nodejs_notification", key, msglist, function (e, r) {
                })
            })

            if (sockets[key]) {
                send_notification(sockets[key], msg)
            }

        }
    }
})

// check redis snschat channel
snschat.on("message", function (pattern, data) {
    msg = JSON.parse(data)
    process_msg(msg)
})


io.sockets.on(‘connection‘, function (socket) {
    var address = socket.handshake.address;
    console.log(Date() + " New connection from " + address.address + ":" + address.port);
    socket.on(‘login‘, function (userinfo) {
        userid = userinfo.myAuraId
var address = socket.handshake.address;
var deviceid = userinfo.deviceId
console.log(Date() + " Login from " + address.address + ":" + address.port + " " + userid + " " + deviceid);
        old_socket = sockets[userid]
if (old_socket && old_socket.deviceid && deviceid && old_socket.deviceid != deviceid) {
            old_socket.relogin = 1
            old_socket.emit(‘logout‘)
console.log("logout " + old_socket.userid + " " + old_socket.deviceid)
        }

if (old_socket && old_socket != socket) {
            old_socket.disconnect()
        }

        socket.relogin = 0
        socket.userid = userid
        socket.deviceid = deviceid

        send_store_msg(socket, userid)

        sockets[userid] = socket
        pub.publish("login_message_channel", JSON.stringify(userinfo))

    })

    socket.on(‘geo‘, function (geo, ack) {
if (geo.myAuraId) {
var now = new Date()
            pub.publish("geo", JSON.stringify({
                geo: geo, time: now.getTime()
            }))
            socket.userid = geo.myAuraId
            sockets[socket.userid] = socket

if (ack) {
                ack(1)
                send_store_msg(socket, userid)

            }
        }
    })

    socket.on(‘chat‘, function (msg, ack) {
//process_msg(msg)
        pub.publish("chat_filter_channel", JSON.stringify(msg))
        socket.userid = msg.from
        sockets[socket.userid] = socket
if (ack) {
            ack(1)
        }

    })

    socket.on(‘hb‘, function (msg, ack) {
if (ack) {
            ack(1)
        }
    })


    socket.on("disconnect", function () {
var address = socket.handshake.address;
console.log(Date() + " Disconnect from " + address.address + ":" + address.port);
if (!socket.relogin) {
delete sockets[socket.userid]
        }
    })

使用Nodejs实现聊天服务器

标签：

原文地址：http://www.cnblogs.com/jidan/p/5085792.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！