站長(zhǎng)資訊網(wǎng)
我們知道，http 是一問(wèn)一答的模式，客戶端向服務(wù)器發(fā)送 http 請(qǐng)求，服務(wù)器返回 http 響應(yīng)。
這種模式對(duì)資源、數(shù)據(jù)的加載足夠用，但是需要數(shù)據(jù)推送的場(chǎng)景就不合適了。
有同學(xué)說(shuō)，http2 不是有 server push 么？
那只是推資源用的：
比如瀏覽器請(qǐng)求了 html，服務(wù)端可以連帶把 css 一起推送給瀏覽器。瀏覽器可以決定接不接收?！鞠嚓P(guān)教程推薦：nodejs視頻教程、編程教學(xué)】
對(duì)于即時(shí)通訊等實(shí)時(shí)性要求高的場(chǎng)景，就需要用 websocket 了。
websocket 嚴(yán)格來(lái)說(shuō)和 http 沒(méi)什么關(guān)系，是另外一種協(xié)議格式。但是需要一次從 http 到 websocekt 的切換過(guò)程。
切換過(guò)程詳細(xì)來(lái)說(shuō)是這樣的：
請(qǐng)求的時(shí)候帶上這幾個(gè) header：
Connection: Upgrade Upgrade: websocket Sec-WebSocket-Key: Ia3dQjfWrAug/6qm7mTZOg==
前兩個(gè)很容易理解，就是升級(jí)到 websocket 協(xié)議的意思。
第三個(gè) header 是保證安全用的一個(gè) key。
服務(wù)端返回這樣的 header：
HTTP/1.1 101 Switching Protocols Connection: Upgrade Upgrade: websocket Sec-WebSocket-Accept: JkE58n3uIigYDMvC+KsBbGZsp1A=
和請(qǐng)求 header 類似，Sec-WebSocket-Accept 是對(duì)請(qǐng)求帶過(guò)來(lái)的 Sec-WebSocket-Key 處理之后的結(jié)果。
加入這個(gè) header 的校驗(yàn)是為了確定對(duì)方一定是有 WebSocket 能力的，不然萬(wàn)一建立了連接對(duì)方卻一直沒(méi)消息，那不就白等了么。
那 Sec-WebSocket-Key 經(jīng)過(guò)什么處理能得到 Sec-WebSocket-Accept 呢？
我用 node 實(shí)現(xiàn)了一下，是這樣的：
const crypto = require('crypto'); function hashKey(key) { const sha1 = crypto.createHash('sha1'); sha1.update(key + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'); return sha1.digest('base64'); }
也就是用客戶端傳過(guò)來(lái)的 key，加上一個(gè)固定的字符串，經(jīng)過(guò) sha1 加密之后，轉(zhuǎn)成 base64 的結(jié)果。
這個(gè)字符串 258EAFA5-E914-47DA-95CA-C5AB0DC85B11 是固定的，不信你搜搜看：
隨便找個(gè)有 websocket 的網(wǎng)站，比如知乎就有：
過(guò)濾出 ws 類型的請(qǐng)求，看看這幾個(gè) header，是不是就是前面說(shuō)的那些。
這個(gè) Sec-WebSocket-Key 是 wk60yiym2FEwCAMVZE3FgQ==
而響應(yīng)的 Sec-WebSocket-Accept 是 XRfPnS+8xl11QWZherej/dkHPHM=
我們算算看：
是不是一毛一樣！
這就是 websocket 升級(jí)協(xié)議時(shí)候的 Sec-WebSocket-Key 對(duì)應(yīng)的 Sec-WebSocket-Accept 的計(jì)算過(guò)程。
這一步之后就換到 websocket 的協(xié)議了，那是一個(gè)全新的協(xié)議：
勾選 message 這一欄可以看到傳輸?shù)南ⅲ梢允俏谋?、可以是二進(jìn)制：
全新的協(xié)議？那具體是什么樣的協(xié)議呢？
這樣的：
大家習(xí)慣的 http 協(xié)議是 key:value 的 header 帶個(gè) body 的：
它是文本協(xié)議，每個(gè) header 都是容易理解的字符。
這樣好懂是好懂，但是傳輸占的空間太大了。
而 websocket 是二進(jìn)制協(xié)議，一個(gè)字節(jié)可以用來(lái)存儲(chǔ)很多信息：
比如協(xié)議的第一個(gè)字節(jié)，就存儲(chǔ)了 FIN（結(jié)束標(biāo)志）、opcode（內(nèi)容類型是 binary 還是 text） 等信息。
第二個(gè)字節(jié)存儲(chǔ)了 mask（是否有加密），payload（數(shù)據(jù)長(zhǎng)度）。
僅僅兩個(gè)字節(jié)，存儲(chǔ)了多少信息呀！
這就是二進(jìn)制協(xié)議比文本協(xié)議好的地方。
我們看到的 weboscket 的 message 的收發(fā)，其實(shí)底層都是拼成這樣的格式。
只是瀏覽器幫我們解析了這種格式的協(xié)議數(shù)據(jù)。
這就是 weboscket 的全部流程了。
其實(shí)還是挺清晰的，一個(gè)切換協(xié)議的過(guò)程，然后是二進(jìn)制的 weboscket 協(xié)議的收發(fā)。
那我們就用 Node.js 自己實(shí)現(xiàn)一個(gè) websocket 服務(wù)器吧！
定義個(gè) MyWebsocket 的 class：
const { EventEmitter } = require('events'); const http = require('http'); class MyWebsocket extends EventEmitter { constructor(options) { super(options); const server = http.createServer(); server.listen(options.port || 8080); server.on('upgrade', (req, socket) => { }); } }
繼承 EventEmitter 是為了可以用 emit 發(fā)送一些事件，外界可以通過(guò) on 監(jiān)聽(tīng)這個(gè)事件來(lái)處理。
我們?cè)跇?gòu)造函數(shù)里創(chuàng)建了一個(gè) http 服務(wù)，當(dāng) ungrade 事件發(fā)生，也就是收到了 Connection: upgrade 的 header 的時(shí)候，返回切換協(xié)議的 header。
返回的 header 前面已經(jīng)見(jiàn)過(guò)了，就是要對(duì) sec-websocket-key 做下處理。
server.on('upgrade', (req, socket) => { this.socket = socket; socket.setKeepAlive(true); const resHeaders = [ 'HTTP/1.1 101 Switching Protocols', 'Upgrade: websocket', 'Connection: Upgrade', 'Sec-WebSocket-Accept: ' + hashKey(req.headers['sec-websocket-key']), '', '' ].join('rn'); socket.write(resHeaders); socket.on('data', (data) => { console.log(data) }); socket.on('close', (error) => { this.emit('close'); }); });
我們拿到 socket，返回上面的 header，其中 key 做的處理就是前面聊過(guò)的算法：
function hashKey(key) { const sha1 = crypto.createHash('sha1'); sha1.update(key + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'); return sha1.digest('base64'); }
就這么簡(jiǎn)單，就已經(jīng)完成協(xié)議切換了。
不信我們?cè)囋嚳础?/p>
引入我們實(shí)現(xiàn)的 ws 服務(wù)器，跑起來(lái)：
const MyWebSocket = require('./ws'); const ws = new MyWebSocket({ port: 8080 }); ws.on('data', (data) => { console.log('receive data:' + data); }); ws.on('close', (code, reason) => { console.log('close:', code, reason); });
然后新建這樣一個(gè) html：
<!DOCTYPE HTML> <html> <body> <script> const ws = new WebSocket("ws://localhost:8080"); ws.onopen = function () { ws.send("發(fā)送數(shù)據(jù)"); setTimeout(() => { ws.send("發(fā)送數(shù)據(jù)2"); }, 3000) }; ws.onmessage = function (evt) { console.log(evt) }; ws.onclose = function () { }; </script> </body> </html>
用瀏覽器的 WebSocket api 建立連接，發(fā)送消息。
用 npx http-server . 起個(gè)靜態(tài)服務(wù)。
然后瀏覽器訪問(wèn)這個(gè) html：
這時(shí)打開 devtools 你就會(huì)發(fā)現(xiàn)協(xié)議切換成功了：
這 3 個(gè) header 還有 101 狀態(tài)碼都是我們返回的。
message 里也可以看到發(fā)送的消息：
再去服務(wù)端看看，也收到了這個(gè)消息：
只不過(guò)是 Buffer 的，也就是二進(jìn)制的。
接下來(lái)只要按照協(xié)議格式解析這個(gè) Buffer，并且生成響應(yīng)格式的協(xié)議數(shù)據(jù) Buffer 返回就可以收發(fā) websocket 數(shù)據(jù)了。
這一部分還是比較麻煩的，我們一點(diǎn)點(diǎn)來(lái)看。
我們需要第一個(gè)字節(jié)的后四位，也就是 opcode。
這樣寫：
const byte1 = bufferData.readUInt8(0); let opcode = byte1 & 0x0f;
讀取 8 位無(wú)符號(hào)整數(shù)的內(nèi)容，也就是一個(gè)字節(jié)的內(nèi)容。參數(shù)是偏移的字節(jié)，這里是 0。
通過(guò)位運(yùn)算取出后四位，這就是 opcode 了。
然后再處理第二個(gè)字節(jié)：
第一位是 mask 標(biāo)志位，后 7 位是 payload 長(zhǎng)度。
可以這樣取：
const byte2 = bufferData.readUInt8(1); const str2 = byte2.toString(2); const MASK = str2[0]; let payloadLength = parseInt(str2.substring(1), 2);
還是用 buffer.readUInt8 讀取一個(gè)字節(jié)的內(nèi)容。
先轉(zhuǎn)成二進(jìn)制字符串，這時(shí)第一位就是 mask，然后再截取后 7 位的子串，parseInt 成數(shù)字，這就是 payload 長(zhǎng)度了。
這樣前兩個(gè)字節(jié)的協(xié)議內(nèi)容就解析完了。
有同學(xué)可能問(wèn)了，后面咋還有倆 payload 長(zhǎng)度呢？
這是因?yàn)閿?shù)據(jù)不一定有多長(zhǎng)，可能需要 16 位存長(zhǎng)度，可能需要 32 位。
于是 websocket 協(xié)議就規(guī)定了如果那個(gè) 7 位的內(nèi)容不超過(guò) 125，那它就是 payload 長(zhǎng)度。
如果 7 位的內(nèi)容是 126，那就不用它了，用后面的 16 位的內(nèi)容作為 payload 長(zhǎng)度。
如果 7 位的內(nèi)容是 127，也不用它了，用后面那個(gè) 64 位的內(nèi)容作為 payload 長(zhǎng)度。
其實(shí)還是容易理解的，就是 3 個(gè) if else。
用代碼寫出來(lái)就是這樣的：
let payloadLength = parseInt(str2.substring(1), 2); let curByteIndex = 2; if (payloadLength === 126) { payloadLength = bufferData.readUInt16BE(2); curByteIndex += 2; } else if (payloadLength === 127) { payloadLength = bufferData.readBigUInt64BE(2); curByteIndex += 8; }
這里的 curByteIndex 是存儲(chǔ)當(dāng)前處理到第幾個(gè)字節(jié)的。
如果是 126，那就從第 3 個(gè)字節(jié)開始，讀取 2 個(gè)字節(jié)也就是 16 位的長(zhǎng)度，用 buffer.readUInt16BE 方法。
如果是 127，那就從第 3 個(gè)字節(jié)開始，讀取 8 個(gè)字節(jié)也就是 64 位的長(zhǎng)度，用 buffer.readBigUInt64BE 方法。
這樣就拿到了 payload 的長(zhǎng)度，然后再用這個(gè)長(zhǎng)度去截取內(nèi)容就好了。
但在讀取數(shù)據(jù)之前，還有個(gè) mask 要處理，這個(gè)是用來(lái)給內(nèi)容解密的：
讀 4 個(gè)字節(jié)，就是 mask key。
再后面的就可以根據(jù) payload 長(zhǎng)度讀出來(lái)。
let realData = null; if (MASK) { const maskKey = bufferData.slice(curByteIndex, curByteIndex + 4); curByteIndex += 4; const payloadData = bufferData.slice(curByteIndex, curByteIndex + payloadLength); realData = handleMask(maskKey, payloadData); } else { realData = bufferData.slice(curByteIndex, curByteIndex + payloadLength);; }
然后用 mask key 來(lái)解密數(shù)據(jù)。
這個(gè)算法也是固定的，用每個(gè)字節(jié)的 mask key 和數(shù)據(jù)的每一位做按位異或就好了：
function handleMask(maskBytes, data) { const payload = Buffer.alloc(data.length); for (let i = 0; i < data.length; i++) { payload[i] = maskBytes[i % 4] ^ data[i]; } return payload; }
這樣，我們就拿到了最終的數(shù)據(jù)！
但是傳給處理程序之前，還要根據(jù)類型來(lái)處理下，因?yàn)閮?nèi)容分幾種類型，也就是 opcode 有幾種值：
const OPCODES = { CONTINUE: 0, TEXT: 1, // 文本 BINARY: 2, // 二進(jìn)制 CLOSE: 8, PING: 9, PONG: 10, };
我們只處理文本和二進(jìn)制就好了：
handleRealData(opcode, realDataBuffer) { switch (opcode) { case OPCODES.TEXT: this.emit('data', realDataBuffer.toString('utf8')); break; case OPCODES.BINARY: this.emit('data', realDataBuffer); break; default: this.emit('close'); break; } }
文本就轉(zhuǎn)成 utf-8 的字符串，二進(jìn)制數(shù)據(jù)就直接用 buffer 的數(shù)據(jù)。
這樣，處理程序里就能拿到解析后的數(shù)據(jù)。
我們來(lái)試一下：
之前我們已經(jīng)能拿到 weboscket 協(xié)議內(nèi)容的 buffer 了：
而現(xiàn)在我們能正確解析出其中的數(shù)據(jù)：
至此，我們 websocket 協(xié)議的解析成功了！
這樣的協(xié)議格式的數(shù)據(jù)叫做 frame，也就是幀：
解析可以了，接下來(lái)我們?cè)賹?shí)現(xiàn)數(shù)據(jù)的發(fā)送。
發(fā)送也是構(gòu)造一樣的 frame 格式。
定義這樣一個(gè) send 方法：
send(data) { let opcode; let buffer; if (Buffer.isBuffer(data)) { opcode = OPCODES.BINARY; buffer = data; } else if (typeof data === 'string') { opcode = OPCODES.TEXT; buffer = Buffer.from(data, 'utf8'); } else { console.error('暫不支持發(fā)送的數(shù)據(jù)類型') } this.doSend(opcode, buffer); } doSend(opcode, bufferDatafer) { this.socket.write(encodeMessage(opcode, bufferDatafer)); }
根據(jù)發(fā)送的是文本還是二進(jìn)制數(shù)據(jù)來(lái)對(duì)內(nèi)容作處理。
然后構(gòu)造 websocket 的 frame：
function encodeMessage(opcode, payload) { //payload.length < 126 let bufferData = Buffer.alloc(payload.length + 2 + 0);; let byte1 = parseInt('10000000', 2) | opcode; // 設(shè)置 FIN 為 1 let byte2 = payload.length; bufferData.writeUInt8(byte1, 0); bufferData.writeUInt8(byte2, 1); payload.copy(bufferData, 2); return bufferData; }
我們只處理數(shù)據(jù)長(zhǎng)度小于 125 的情況。
第一個(gè)字節(jié)是 opcode，我們把第一位置 1 ，通過(guò)按位或的方式。
服務(wù)端給客戶端回消息不需要 mask，所以第二個(gè)字節(jié)就是 payload 長(zhǎng)度。
分別把這前兩個(gè)字節(jié)的數(shù)據(jù)寫到 buffer 里，指定不同的 offset：
bufferData.writeUInt8(byte1, 0); bufferData.writeUInt8(byte2, 1);
之后把 payload 數(shù)據(jù)放在后面：
payload.copy(bufferData, 2);
這樣一個(gè) websocket 的 frame 就構(gòu)造完了。
我們?cè)囈幌拢?/p>
收到客戶端消息后，每?jī)擅牖匾粋€(gè)消息。
收發(fā)消息都成功了！
就這樣，我們自己實(shí)現(xiàn)了一個(gè) websocket 服務(wù)器，實(shí)現(xiàn)了 websocket 協(xié)議的解析和生成！
完整代碼如下：
MyWebSocket:
//ws.js const { EventEmitter } = require('events'); const http = require('http'); const crypto = require('crypto'); function hashKey(key) { const sha1 = crypto.createHash('sha1'); sha1.update(key + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'); return sha1.digest('base64'); } function handleMask(maskBytes, data) { const payload = Buffer.alloc(data.length); for (let i = 0; i < data.length; i++) { payload[i] = maskBytes[i % 4] ^ data[i]; } return payload; } const OPCODES = { CONTINUE: 0, TEXT: 1, BINARY: 2, CLOSE: 8, PING: 9, PONG: 10, }; function encodeMessage(opcode, payload) { //payload.length < 126 let bufferData = Buffer.alloc(payload.length + 2 + 0);; let byte1 = parseInt('10000000', 2) | opcode; // 設(shè)置 FIN 為 1 let byte2 = payload.length; bufferData.writeUInt8(byte1, 0); bufferData.writeUInt8(byte2, 1); payload.copy(bufferData, 2); return bufferData; } class MyWebsocket extends EventEmitter { constructor(options) { super(options); const server = http.createServer(); server.listen(options.port || 8080); server.on('upgrade', (req, socket) => { this.socket = socket; socket.setKeepAlive(true); const resHeaders = [ 'HTTP/1.1 101 Switching Protocols', 'Upgrade: websocket', 'Connection: Upgrade', 'Sec-WebSocket-Accept: ' + hashKey(req.headers['sec-websocket-key']), '', '' ].join('rn'); socket.write(resHeaders); socket.on('data', (data) => { this.processData(data); // console.log(data); }); socket.on('close', (error) => { this.emit('close'); }); }); } handleRealData(opcode, realDataBuffer) { switch (opcode) { case OPCODES.TEXT: this.emit('data', realDataBuffer.toString('utf8')); break; case OPCODES.BINARY: this.emit('data', realDataBuffer); break; default: this.emit('close'); break; } } processData(bufferData) { const byte1 = bufferData.readUInt8(0); let opcode = byte1 & 0x0f; const byte2 = bufferData.readUInt8(1); const str2 = byte2.toString(2); const MASK = str2[0]; let curByteIndex = 2; let payloadLength = parseInt(str2.substring(1), 2); if (payloadLength === 126) { payloadLength = bufferData.readUInt16BE(2); curByteIndex += 2; } else if (payloadLength === 127) { payloadLength = bufferData.readBigUInt64BE(2); curByteIndex += 8; } let realData = null; if (MASK) { const maskKey = bufferData.slice(curByteIndex, curByteIndex + 4); curByteIndex += 4; const payloadData = bufferData.slice(curByteIndex, curByteIndex + payloadLength); realData = handleMask(maskKey, payloadData); } this.handleRealData(opcode, realData); } send(data) { let opcode; let buffer; if (Buffer.isBuffer(data)) { opcode = OPCODES.BINARY; buffer = data; } else if (typeof data === 'string') { opcode = OPCODES.TEXT; buffer = Buffer.from(data, 'utf8'); } else { console.error('暫不支持發(fā)送的數(shù)據(jù)類型') } this.doSend(opcode, buffer); } doSend(opcode, bufferDatafer) { this.socket.write(encodeMessage(opcode, bufferDatafer)); } } module.exports = MyWebsocket;
Index：
const MyWebSocket = require('./ws'); const ws = new MyWebSocket({ port: 8080 }); ws.on('data', (data) => { console.log('receive data:' + data); setInterval(() => { ws.send(data + ' ' + Date.now()); }, 2000) }); ws.on('close', (code, reason) => { console.log('close:', code, reason); });
html:
<!DOCTYPE HTML> <html> <body> <script> const ws = new WebSocket("ws://localhost:8080"); ws.onopen = function () { ws.send("發(fā)送數(shù)據(jù)"); setTimeout(() => { ws.send("發(fā)送數(shù)據(jù)2"); }, 3000) }; ws.onmessage = function (evt) { console.log(evt) }; ws.onclose = function () { }; </script> </body> </html>
總結(jié)
實(shí)時(shí)性較高的需求，我們會(huì)用 websocket 實(shí)現(xiàn)，比如即時(shí)通訊、游戲等場(chǎng)景。
websocket 和 http 沒(méi)什么關(guān)系，但從 http 到 websocket 需要一次切換的過(guò)程。
這個(gè)切換過(guò)程除了要帶 upgrade 的 header 外，還要帶 sec-websocket-key，服務(wù)端根據(jù)這個(gè) key 算出結(jié)果，通過(guò) sec-websocket-accept 返回。響應(yīng)是 101 Switching Protocols 的狀態(tài)碼。
這個(gè)計(jì)算過(guò)程比較固定，就是 key + 固定的字符串 通過(guò) sha1 加密后再 base64 的結(jié)果。
加這個(gè)機(jī)制是為了確保對(duì)方一定是 websocket 服務(wù)器，而不是隨意返回了個(gè) 101 狀態(tài)碼。
之后就是 websocket 協(xié)議了，這是個(gè)二進(jìn)制協(xié)議，我們根據(jù)格式完成了 websocket 幀的解析和生成。
這樣就是一個(gè)完整的 websocket 協(xié)議的實(shí)現(xiàn)了。
我們自己手寫了一個(gè) websocket 服務(wù)，有沒(méi)有感覺(jué)對(duì) websocket 的理解更深了呢？
