常見(jiàn)的消息通知:

常見(jiàn)的站內(nèi)通知類(lèi)別（括號(hào)里是對(duì)自己目前項(xiàng)目出現(xiàn)情況的分析，讀者忽略）：

• 公告 Announcement （通道加入新的組織、某組織或用戶新上傳了某數(shù)據(jù)摘要、系統(tǒng)凌晨需要版本更新等事件）
• 提醒 Remind（用戶之間、系統(tǒng)與用戶之間）
  • 資源訂閱提醒（關(guān)注的數(shù)據(jù)摘要更新了內(nèi)容、評(píng)論等）
  • 資源發(fā)布提醒（我發(fā)布的數(shù)據(jù)摘要被評(píng)論了，被關(guān)注了，被申請(qǐng)交易了）
  • 系統(tǒng)提醒
• 私信 Mailbox（類(lèi)似聊天室吧，暫時(shí)沒(méi)有這需求）

實(shí)現(xiàn)思路與步驟等

除了用消息隊(duì)列MQ以外，筆者想到的是使用websocket協(xié)議實(shí)現(xiàn)，該協(xié)議為全雙工通信full-duplex，長(zhǎng)連接PersistentConnection，相比http來(lái)說(shuō)是種持久化協(xié)議。

其中主要的開(kāi)發(fā)步驟有：

• 綁定連接(用戶賬號(hào)和websocket之間的連接)
• 管理連接
• 收發(fā)消息(數(shù)據(jù)格式和讀取等具體實(shí)現(xiàn))。

其中，需要注意的點(diǎn)有：

• 長(zhǎng)連接的心跳激活處理；
• 服務(wù)端調(diào)優(yōu)實(shí)現(xiàn)高并發(fā)量client同時(shí)在線（單機(jī)服務(wù)器可以實(shí)現(xiàn)百萬(wàn)并發(fā)長(zhǎng)連接）；
• 群發(fā)消息；
• 服務(wù)端維持多用戶的狀態(tài)；
• 從WebSocket中獲取HttpSession進(jìn)行用戶相關(guān)操作等

具體實(shí)現(xiàn)思路：

• 前端使用WebSocket與服務(wù)端創(chuàng)建連接的時(shí)候，將用戶ID傳給服務(wù)端，服務(wù)端將用戶ID與channel關(guān)聯(lián)起來(lái)存儲(chǔ)，同時(shí)將channel放入到channel組中。（這里的channel就是服務(wù)器與客戶端之間的連接）
• 如果需要給所有用戶發(fā)送消息，直接執(zhí)行channel組的writeAndFlush()方法；
• 如果需要給指定用戶發(fā)送消息，根據(jù)用戶ID查詢(xún)到對(duì)應(yīng)的channel,然后執(zhí)行writeAndFlush()方法；
• 前端獲取到服務(wù)端推送的消息之后，將消息內(nèi)容展示到文本域中。

其他方法介紹

輪詢(xún)：客戶端定時(shí)向服務(wù)器發(fā)送Ajax請(qǐng)求，服務(wù)器接到請(qǐng)求后馬上返回響應(yīng)信息并關(guān)閉連接。 優(yōu)點(diǎn)：后端程序編寫(xiě)比較容易。 缺點(diǎn)：請(qǐng)求中有大半是無(wú)用，浪費(fèi)帶寬和服務(wù)器資源。 實(shí)例：適于小型應(yīng)用。

長(zhǎng)輪詢(xún)：客戶端向服務(wù)器發(fā)送Ajax請(qǐng)求，服務(wù)器接到請(qǐng)求后hold住連接，直到有新消息才返回響應(yīng)信息并關(guān)閉連接，客戶端處理完響應(yīng)信息后再向服務(wù)器發(fā)送新的請(qǐng)求。 優(yōu)點(diǎn)：在無(wú)消息的情況下不會(huì)頻繁的請(qǐng)求，耗費(fèi)資小。 缺點(diǎn)：服務(wù)器hold連接會(huì)消耗資源，返回?cái)?shù)據(jù)順序無(wú)保證，難于管理維護(hù)。 Comet異步的ashx， 實(shí)例：WebQQ、Hi網(wǎng)頁(yè)版、Facebook IM。

長(zhǎng)連接：在頁(yè)面里嵌入一個(gè)隱蔵iframe，將這個(gè)隱蔵iframe的src屬性設(shè)為對(duì)一個(gè)長(zhǎng)連接的請(qǐng)求或是采用xhr請(qǐng)求，服務(wù)器端就能源源不斷地往客戶端輸入數(shù)據(jù)。 優(yōu)點(diǎn)：消息即時(shí)到達(dá)，不發(fā)無(wú)用請(qǐng)求；管理起來(lái)也相對(duì)便。 缺點(diǎn)：服務(wù)器維護(hù)一個(gè)長(zhǎng)連接會(huì)增加開(kāi)銷(xiāo)。 實(shí)例：Gmail聊天

Flash Socket：在頁(yè)面中內(nèi)嵌入一個(gè)使用了Socket類(lèi)的 Flash 程序JavaScript通過(guò)調(diào)用此Flash程序提供的Socket接口與服務(wù)器端的Socket接口進(jìn)行通信，JavaScript在收到服務(wù)器端傳送的信息后控制頁(yè)面的顯示。 優(yōu)點(diǎn)：實(shí)現(xiàn)真正的即時(shí)通信，而不是偽即時(shí)。 缺點(diǎn)：客戶端必須安裝Flash插件；非HTTP協(xié)議，無(wú)法自動(dòng)穿越防火墻。 實(shí)例：網(wǎng)絡(luò)互動(dòng)游戲。

技術(shù)實(shí)現(xiàn)與相關(guān)包介紹

包介紹

nodejs不像其他的服務(wù)器，對(duì)于不同的連接，不支持進(jìn)程和線程操作，寫(xiě)這類(lèi)功能的時(shí)候就需要找更合適的包。

使用WebSocket協(xié)議的包有好多，這里我先講一種常用的包是nodejs-websocket包，網(wǎng)評(píng)說(shuō)使用較為繁瑣，這里就沒(méi)使用。它需要依賴(lài)于底層的C++,Python的環(huán)境，支持以node做客戶端的訪問(wèn)。當(dāng)然了，這里我一定要說(shuō)一下，nodejs-websocket是純粹的使用了WebSocket協(xié)議，因此使用時(shí)需要寫(xiě)心跳檢測(cè)，檢測(cè)用戶是否在線等情況。

我采用的是socket.io，它使用起來(lái)較為簡(jiǎn)單，功能強(qiáng)大，支持集成websocket服務(wù)器端和Express3框架與一身。它可以不需要心跳檢測(cè)，不過(guò)這也是個(gè)相對(duì)說(shuō)法，因?yàn)樗Y(jié)合封裝了輪詢(xún)機(jī)制和實(shí)時(shí)通信，當(dāng)websocket連接斷掉時(shí)，它會(huì)不停的嘗試連接，耗費(fèi)資源。當(dāng)然了，還有其他庫(kù)，比如node-websocket-server（不需要了解，直接放棄）。

技術(shù)實(shí)現(xiàn)

在實(shí)現(xiàn)前，考慮到發(fā)送消息時(shí)，向指定用戶發(fā)送WebSocket消息，但對(duì)方可能不在線，這種情況，我這么處理：

• 如果接收者在線，則存儲(chǔ)進(jìn)redis并實(shí)時(shí)發(fā)送消息；
• 否則將消息存儲(chǔ)到redis，等用戶登陸上線后主動(dòng)推送未讀消息。

socket.io的客戶端和服務(wù)端都有兩個(gè)函數(shù) on()、emit()，核心函數(shù)，可輕松實(shí)現(xiàn)客戶端與服務(wù)端的雙向通信。

• emit：觸發(fā)一個(gè)事件，第一個(gè)參數(shù)是事件名稱(chēng)，第二個(gè)參數(shù)是要發(fā)送到另一端的數(shù)據(jù)，第三個(gè)參數(shù)是一個(gè)回調(diào)函數(shù)用來(lái)確認(rèn)對(duì)方的接收信息（也可以說(shuō)時(shí)回執(zhí)），可忽略。
  • socket.emit 信息傳輸對(duì)象為當(dāng)前 socket 對(duì)應(yīng)的 client ，各個(gè)client socket 相互不影響。
  • socket.broadcast.emit 信息傳輸對(duì)象為所有 client ，排除當(dāng)前socket 對(duì)應(yīng)的 client。
  • io.sockets.emit信息傳輸對(duì)象為所有 client。
• on：注冊(cè)一個(gè)事件，用來(lái)監(jiān)聽(tīng) emit 觸發(fā)的事件。

服務(wù)端

直接上代碼：

    'use strict';

    // 維護(hù)socket連接的代碼
    const { addSocketId, getSocketId, deleteSocketId } = require('../../../utils/socket/socketId');
    // 保存消息
    const message = require('../saveMessage');
    // socket連接許可驗(yàn)證
    const { socketAuth } = require('../../../middleware/socket/index')

    // socket接口，傳入/bin/www.js
    function init(io) {
    
    /**
     * @description: 為每個(gè)傳入執(zhí)行的功能Socket，并且接收套接字和可選地將執(zhí)行延遲到下一個(gè)注冊(cè)的中間件的參數(shù)
     */    
    io.use((socket, next) => {
        if (socket.request.headers.cookie) return next();
        next(new Error('Authentication error'));
    });

    io.on('connection', function(socket) {

        /**
         * @description: 用戶登錄，則保存用戶連接的相關(guān)信息,并從redis拉取未讀消息，推送給該用戶
         */        
        socket.on('user_login', function(socketInfo) {       
            if(!socketInfo.userId) {
                // io.sockets.to(socketInfo['socketId']).emit('disconnect', '');
                return;
            }
            // 將用戶與socket插入數(shù)據(jù)庫(kù)中
            addSocketId(socketInfo);  
                      
            if (process.env.NODE_ENV === 'development') {
                displayUserInfo(socketInfo);
            };

            // 推送所有消息
            message.pushMessage(socketInfo['userId']).then(pushData => {
                io.sockets.to(socketInfo['socketId']).emit('push_message', pushData);
            });
        });
    
        /**
         * @description: 發(fā)給某用戶交易通知（在線實(shí)時(shí)通知，并存儲(chǔ)至redis）
         */        
        socket.on('todo', function(todoData) {    
            // 存入redis
            message.addMessage(todoData);
            // 檢測(cè)用戶是否在線
            message.isOnline(todoData['receiver_id']).then(isOnline => {
                // 用戶在線則通信
                if (isOnline == true) {
                    getSocketId(todoData['receiver_id']).then(socketId => {            
                        io.sockets.to(socketId).emit('todo_message', todoData);
                    }); 
                };  
            });
        });
    
        // TODO: 需要提醒前端在關(guān)閉窗口之前先斷開(kāi)連接（窗口刷新之前應(yīng)該不需要）
        /**
         * @description: 斷開(kāi)連接
         */        
        socket.on('disconnect', function() {
            // 從數(shù)據(jù)庫(kù)中刪除連接
            deleteSocketId(socket.id);
            // 判斷當(dāng)前是否是開(kāi)發(fā)環(huán)境
            if (process.env.NODE_ENV === 'development') {
                displayUserInfo();
            }
        });
    
    });
    
}

function displayUserInfo(user) {
    console.log(`當(dāng)前登錄用戶信息:${user}`);
    return;
}

module.exports = {
    init
};

上方代碼中，主要?jiǎng)?chuàng)建了connection事件，其下又有user_login、todo、disconnect事件，然后這些事件下又有其創(chuàng)建或監(jiān)聽(tīng)的事件。其中，user_login事件主要是監(jiān)聽(tīng)前端用戶的登錄成功，若用戶成功上線，則將redis內(nèi)的已讀未讀消息分類(lèi)后推送給客戶端。todo事件則是判斷用戶在線后，實(shí)時(shí)傳遞消息，需要注意使用io.sockets.to(socketId).emit(eventname, eventdata)實(shí)現(xiàn)P2P消息傳送，socketId即為接收消息用戶的WebSocket連接的ID?？蛻舳藙t需要監(jiān)聽(tīng)后面emit()參數(shù)中的eventname事件。disconnect事件則是在客戶端用戶登出或刷新頁(yè)面等認(rèn)為是斷開(kāi)WebSocket連接時(shí)，在維護(hù)的socket連接組中刪除該用戶的WebSocket連接信息。

當(dāng)然，在連接到connection事件前，有一個(gè)中間件io.use((socket, next) => {}，是判斷對(duì)方的連接是否有效（帶有cookie的主動(dòng)連接）。

然后，在/bin/www .js中引入io：

#!/usr/bin/env node

// 模塊依賴(lài)
var app = require('../app');
var http = require('http');
const socketIndex = require('../src/routes/socket/index/socket');

// 從環(huán)境中取端口，應(yīng)用到express
var port = normalizePort(process.env.PORT || '3000');
app.set('port', port);

// 創(chuàng)建http服務(wù)（將express注冊(cè)到http中）
server = http.createServer(app);

// 監(jiān)聽(tīng)
var io = require('socket.io')(server, {
  cors: {
      origin: '*'
  }
  // path: '/socket' // 重新定義socket連接路徑
});

// 全局聲明
global.io = io;

// socket的程序文件下引入io
socketIndex.init(io);

其中，引入函數(shù)init()即是上一段代碼中的init函數(shù),傳入?yún)?shù)即為在服務(wù)端入口中創(chuàng)建的io服務(wù)。io服務(wù)中需要傳入cors參數(shù)，解決跨域問(wèn)題，如果想更改websocket連接的地址，則使用path參數(shù)，其參數(shù)值即是在原先基礎(chǔ)的websocket連接地址后加上。

客戶端

首先創(chuàng)建一個(gè)socket對(duì)象，io() 的第一個(gè)參數(shù)是鏈接服務(wù)器的 URL，默認(rèn)情況下是 window.location（需要修改成服務(wù)端的URL，包括對(duì)應(yīng)的模塊或權(quán)限對(duì)應(yīng)的指定路徑，path參數(shù)）。

到此這篇關(guān)于nodejs+socket.io實(shí)現(xiàn)p2p消息實(shí)時(shí)發(fā)送的項(xiàng)目實(shí)踐的文章就介紹到這了,更多相關(guān)node socket.io實(shí)現(xiàn)p2p實(shí)時(shí)發(fā)送內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論