基于node+socket.io+redis的多房間多進程聊天室

Guakin_Huang 發布于2019-08-22 11:02 / 1055人閱讀

摘要：并且指定收到消息，以及端口的監聽方法。四代碼示例多房間實時聊天室配置版本須在里配置定義，并設置。使同一個的請求能夠落在同一個機器同一個進程中。通過主進程統一管理維護子進程，每個進程監聽一個端口。

一、相關技術介紹：

消息實時推送，指的是將消息實時地推送到瀏覽器，用戶不需要刷新瀏覽器就可以實時獲取最新的消息，實時聊天室的技術原理也是如此。傳統的Web站點為了實現推送技術，所用的技術都是輪詢，這種傳統的模式帶來很明顯的缺點，即瀏覽器需要不斷的向服務器發出請求。
短輪詢(Polling)

客戶端需要定時往瀏覽器輪詢發送請求，且只有當服務有數據更新后，客戶端的下一次輪詢請求才能拿到更新后的數據，在數據更新前的多次請求相當于無效。這對帶寬資源造成了極大的浪費，若提高輪詢定時器時間，又會有數據更新不及時的煩惱。
commet
為了解決短輪詢的弊端，一種基于http長連接的"服務器推"方式被hack出來。其與短輪詢的區別主要是，采用commet時，客戶端與服務端保持一個長連接，當數據發生改變時，服務端主動將數據推送到客戶端。Comet 又可以被細分為兩種實現方式，一種是長輪詢機制，一種是流技術。

長輪詢

長輪詢跟短輪詢不同的地方是，客戶端往服務端發送請求后，服務端判斷是否有數據更新，若沒有，則將請求hold住，等待數據更新時，才返回響應。這樣則避免了大量無效的http請求，但即使采用長輪詢方式，接受數據更新的最小時間間隔還是為2*RTT(往返時間)。

流技術

流技術(http stream)基于iframe實現。通過HTML標簽iframe src指向服務端，建立一個長連接。當有數據推送，則往客戶端返回，無須再請求。但流技術有個缺點就是，在瀏覽器頂部會一直出現頁面未加載完成的loading標示。

websocket

為了解決服務端如何更快地實時推送數據到客戶端以及以上推送方式技術的不足，HTML5中定義了Websocket協議，它是一種在單個TCP連接上進行全雙工通訊的協議。與http協議不同的請求/響應模式不同，Websocket在建立連接之前有一個Handshake（Opening Handshake）過程，建立連接之后，雙方即可雙向通信。當然，由于websocket是html5新特性，在部分瀏覽器(IE10以下)是不支持的。
我們來看下websocket的握手報文：

請求報文:

GET /chat HTTP/1.1
Host: server.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Protocol: chat
Sec-WebSocket-Version: 13
Origin: http://example.com

"Upgrade "、"Connection"：告訴服務器這個請求是一個websocket協議，需要區別處理

"Upgrade: websocket"：表明這是一個 WebSocket 類型請求，意在告訴 server 需要將通信協議切換到 WebSocket

"Sec-WebSocket-Key"：是 client 發送的一個 base64 編碼的密文，要求 server 必須返回一個對應加密的 "Sec-WebSocket-Accept" 應答，否則 client 會拋出 "Error during WebSocket handshake" 錯誤，并關閉連接

"Sec-WebSocket-Protocol"：一個用戶定義的字符串，用來區分同URL下，不同的服務所需要的協議

"Sec-WebSocket-Version"：Websocket Draft （協議版本）

響應報文：

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
Sec-WebSocket-Protocol: chat

"Sec-WebSocket-Accept"：這個則是經過服務器確認，并且加密過后的 Sec-WebSocket-Key。加密方式為將Sec-WebSocket-Key與一段固定的 GUID 字符串進行連接，然后進行SHA-1 hash，接著base64編碼得到。

socket.io(http://socket.io)

是一個完全由JavaScript實現，基于Node.js、支持WebSocket的協議用于實時通信、跨平臺的開源框架。Socket.IO除了支持WebSocket通訊協議外，還支持許多種輪詢機制以及其它實時通信方式，并封裝成了通用的接口，并能夠根據瀏覽器對通訊機制的支持情況自動地選擇最佳的方式來實現網絡實時應用。

首先，我們創建一個socket.io server對象，指定監聽80端口。并且指定收到message消息，以及socket端口的監聽方法。接著，當socket建立連接后，通過socket.emit方法，可以往客戶端發送消息。

 var io = require("socket.io")();
 io.on("connection", function(socket) {
    //接受消息
    socket.on("message", function (msg) {
        console.log("receive messge : " + msg );
    });
    
    //發送消息
    socket.emit("message", "hello");
    
    //斷開連接回調
    socket.on("disconnect", function () { 
        console.log("socket disconnect");
    });
});
io.listen(80);

客戶端的代碼也非常簡單，只要引入socket.io對應的客戶端庫(https://github.com/socketio/s...。
在socket建立連接的回調中，使用socket.emit以及socket.on就可以分別做消息的發送以及監聽了。

二、多節點集群架構設計

若只是單機部署應用，單純使用socket.io的消息事件監聽處理即可滿足我們的需求。但隨著業務的擴大，我們需要考慮多機集群部署，客戶端可以連接到任一節點，并發送消息。如何做到多節點的同時推送，我們需要建立一套多節點之間的消息分發/訂閱架構。這時我們引入redis的pub/sub功能。

redis
redis是一個key-value存儲系統，在該項目中主要起到一個消息分發中心(publish/subscribe)的作用。用戶通過socket.io namespace 訂閱房間號后，socket.io server則往redis訂閱(subscribe)該房間號channel。當在該房間中的某一用戶發送消息時，則通過redis的publish功能往redis該房間號channel publish消息。這樣所有訂閱該房間號channel的websocket連接則會收到消息回調，然后推送給客戶端。

nginx
由于采用了集群架構，則需要nginx來做反向代理。需要注意的是，websocket的支持需要nginx1.3以上版本。并且我們需要通過配置ip_hash做粘性會話(ip_hash)處理，避免在低版本瀏覽器socket.io使用兼容方案輪詢請求，請求到不同機器，造成session異常。

####三、架構設計圖

客戶端通過socket.io namespace 指定對應roomid，請求到nginx。nginx根據ip_hash反向代理到對應機器的某一端口的socket.io server 進程。建立websocket連接，并往redis訂閱對應到房間(roomid)channel。到這個時候，一個訂閱了某一房間的websocket通道建立完成。
當用戶發送消息時，socket.io server捕獲到該房間到消息后，即往redis對應房間id的channel publish消息。這時所有訂閱了該房間id channel的socket.io server就會收到訂閱響應，接著找到對應房間id的webscoket通道，并將消息推送到客戶端。

四、代碼示例(多房間實時聊天室)：

nginx配置(nginx版本須>1.3)：
在http{}里配置定義upstream，并設置ip_hash。使同一個ip的請求能夠落在同一個機器同一個進程中。如果改節點掛了，則自動重連到另外一個節點，該方案對于后期擴容也非常方便。

upstream io_nodes {
 ip_hash;
 server 127.0.0.1:6001;
 server 127.0.0.1:6002;
 server 127.0.0.1:6003;
 server 127.0.0.1:6004;
 server 127.0.0.1:6005;
 server 127.0.0.1:6006;
 server 127.0.0.1:6007;
 server 127.0.0.1:6008;
 server 10.x.x.x:6001;
 server 10.x.x.x:6002;
 server 10.x.x.x:6003;
 server 10.x.x.x:6004;
 server 10.x.x.x:6005;
 server 10.x.x.x:6006;
 server 10.x.x.x:6007;
 server 10.x.x.x:6008;
 }

在server中，配置location：

location / {
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header X-Forwarded-For  $proxy_add_x_forwarded_for;
    proxy_set_header Host $host;
    proxy_http_version 1.1;
    proxy_pass http://io_nodes;
    proxy_redirect off;
}

cluster.js
我們采用了多進程的設計，充分利用cpu多核優勢。通過主進程統一管理維護子進程，每個進程監聽一個端口。

var cupNum = require("os").cpus().length,
    workerArr = [],
    roomInfo = [];
var connectNum = 0;

for (var i = 0; i < cupNum; i++) {
    workerArr.push(fork("./fork_server.js", [6001 + i]));

    workerArr[i].on("message", function(msg) {
        if (msg.cmd && msg.cmd === "client connect") {
            connectNum++;
            console.log("socket server connectnum:" + connectNum);
        }
        if (msg.cmd && msg.cmd === "client disconnect") {
            connectNum--;
            console.log("socket server connectnum:" + connectNum);
        }
    });

fork_server.js

var process = require("process");

var io = require("socket.io")();

var num = 0;

var redis = require("redis");
var redisClient = redis.createClient;

//建立redis pub、sub連接
var pub = redisClient({port:13800, host: "127.0.0.1", password:"xxxx"});
var sub = redisClient({port: 13800, host:"127.0.0.1", password:"xxxx"});

var roomSet = {};

//獲取父進程傳遞端口
var port = parseInt(process.argv[2]);

//當websocket連接時
io.on("connection", function(socket) {

    //客戶端請求ws URL:  http://127.0.0.1:6001?roomid=k12_webcourse_room_1
    var roomid = socket.handshake.query.roomid;

    console.log("worker pid: " + process.pid  + " join roomid: "+ roomid);
    
    socket.on("join", function (data) {

        socket.join(roomid);    //加入房間
         
        // 往redis訂閱房間id
        if(!roomSet[roomid]){
            roomSet[roomid] = {};
            console.log("sub channel " + roomid);
            sub.subscribe(roomid);
        }

      roomSet[roomid][socket.id] = {};
      reportConnect();
      console.log(data.username + " join, IP: " + socket.client.conn.remoteAddress);
      roomSet[roomid][socket.id].username = data.username;
      // 往該房間id的reids channel publish用戶進入房間消息
      pub.publish(roomid, JSON.stringify({"event":"join","data": data}));
  });
  
  //用戶發言 推送消息到redis
  socket.on("say", function (data) {
    console.log("Received Message: " + data.text);
    pub.publish(roomid, JSON.stringify({"event":"broadcast_say","data": {
      username: roomSet[roomid][socket.id].username,
      text: data.text
    }}));
  });


    socket.on("disconnect", function() {
        num--;
        console.log("worker pid: " + process.pid + " clien disconnection num:" + num);
        process.send({
            cmd: "client disconnect"
        });

        if (roomSet[roomid] && roomSet[roomid][socket.id] && roomSet[roomid][socket.id].username) {
      console.log(roomSet[roomid][socket.id].username + " quit");
      pub.publish(roomid, JSON.stringify({"event":"broadcast_quit","data": {
        username: roomSet[roomid][socket.id].username
      }}));
    }
    roomSet[roomid] && roomSet[roomid][socket.id] && (delete roomSet[roomid][socket.id]);

    });
});

/**
 * 訂閱redis 回調
 * @param  {[type]} channel [頻道]
 * @param  {[type]} count   [數量]  
 * @return {[type]}         [description]
 */
sub.on("subscribe", function (channel, count) {
    console.log("worker pid: " + process.pid + " subscribe: " + channel);
});

/**
 * 收到redis publish 對應channel的消息
 * @param  {[type]} channel  [description]
 * @param  {[type]} message
 * @return {[type]}          [description]
 */
sub.on("message", function (channel, message) {
    console.log("message channel " + channel + ": " + message);
    //往對應房間廣播消息
    io.to(channel).emit("message", JSON.parse(message));
});

/**
 * 上報連接到master進程 
 * @return {[type]} [description]
 */
var reportConnect = function(){
    num++;
    console.log("worker pid: " + process.pid + " client connect connection num:" + num);
    process.send({
        cmd: "client connect"
    });
};


io.listen(port);

console.log("worker pid: " + process.pid + " listen port:" + port);

客戶端：

gihub源碼地址：https://github.com/493326889/...

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/91657.html

基于微信公眾號的多房間實時彈幕消息系統

摘要：同時借助實現在非接口中推送消息流。每分秒鐘最多的彈幕數目彈幕數量過多時優先加載最新的。項目起始原因源于數據庫課設和以前的一次突發奇想。其實還有其他微信公眾號的彈幕系統，但是我發現使用體驗不佳，因為那種彈幕系統都是私用，并且只支持同時進行一個房間的使用。所以便萌生了自己寫一個的想法。（第一次寫md，有點不會，希望諒解--）主要技術點 Redis（結合socket實現在非socke...

Anonymous1 2019-08-23 12:08 評論0 收藏0
vue+socket.io+express+mongodb 實現簡易多房間在線群聊

摘要：項目簡介主要是通過做一個多人在線多房間群聊的小項目來練手全棧技術的結合運用。編譯運行開啟服務，新建命令行窗口啟動服務端，新建命令行窗口啟動前端頁面然后在瀏覽器多個窗口打開，注冊不同賬號并登錄即可進行多用戶多房間在線聊天。項目簡介主要是通過做一個多人在線多房間群聊的小項目、來練手全棧技術的結合運用。項目源碼：chat-vue-node 主要技術： vue2全家桶 + socket....

android_c 2019-06-26 17:25 評論0 收藏0
前言項目開始是因為工作需要一個聊天室功能，但是因為某些原因最終選用的是基于xmpp協議的Strophe.js寫的。于是就想用node自己寫一套，本來只是想簡單的寫個聊天頁面，但是寫完了又不滿意，所以不斷的重構（似乎可以理解產品經理為什么老是改需求了?乛?乛?）。很多東西，比如mongodb，我也是第一次用，以前只接觸過mysql。所以都是一邊學一邊寫，利用工作之余的時間，斷斷續續的寫了幾個月（...

PAMPANG 2019-06-26 17:57 評論0 收藏0