站長資訊網
node.js極速入門課程：進入學習
【相關教程推薦：nodejs視頻教程】
什么是RPC？
RPC：Remote Procedure Call(遠程過程調用)是指遠程過程調用，也就是說兩臺服務器A，B，一個應用部署在A服務器上，想要調用B服務器上應用提供的函數/方法，由于不在一個內存空間，不能直接調用，需要通過網絡來表達調用的語義和傳達調用的數據。
服務器和服務器之間的通信
RPC vs HTTP
相同點
- 都是兩臺計算機之間的網絡通信。ajax是瀏覽器和服務器之間的通行，RPC是服務器與服務器之間的通行
- 需要雙方約定一個數據格式
不同點
- 尋址服務器不同
ajax 是使用 DNS作為尋址服務獲取域名所對應的ip地址，瀏覽器拿到ip地址之后發送請求獲取數據。
RPC一般是在內網里面相互請求，所以它一般不用DNS做尋址服務。因為在內網，所以可以使用規定的id或者一個虛擬vip，比如v5:8001，然后到尋址服務器獲取v5所對應的ip地址。
- 應用層協議不同
ajax使用http協議，它是一個文本協議，我們交互數據的時候文件格式要么是html，要么是json對象，使用json的時候就是key-value的形式。
RPC采用二進制協議。采用二進制傳輸，它傳輸的包是這樣子的[0001 0001 0111 0110 0010]，里面都是二進制，一般采用那幾位表示一個字段，比如前6位是一個字段，依次類推。
這樣就不需要http傳輸json對象里面的key，所以有更小的數據體積。
因為傳輸的是二進制，更適合于計算機來理解，文本協議更適合人類理解，所以計算機去解讀各個字段的耗時是比文本協議少很多的。
RPC采用二進制有更小的數據體積，及更快的解讀速度。
- TCP通訊方式
-
單工通信：只能客戶端給服務端發消息，或者只能服務端給客戶端發消息
-
半雙工通信：在某個時間段內只能客戶端給服務端發消息，過了這個時間段服務端可以給客戶端發消息。如果把時間分成很多時間片，在一個時間片內就屬于單工通信
-
全雙工通信：客戶端和服務端能相互通信
選擇這三種通信方式的哪一種主要考慮的因素是：實現難度和成本。全雙工通信是要比半雙工通信的成本要高的，在某些場景下還是可以考慮使用半雙工通信。
ajax是一種半雙工通信。http是文本協議，但是它底層是tcp協議，http文本在tcp這一層會經歷從二進制數據流到文本的轉換過程。
理解RPC只是在更深入地理解前端技術。
buffer編解碼二進制數據包
創建buffer
buffer.from: 從已有的數據創建二進制
const buffer1 = Buffer.from('geekbang') const buffer2 = Buffer.from([0, 1, 2, 3, 4]) <Buffer 67 65 65 6b 62 61 6e 67> <Buffer 00 01 02 03 04>
buffer.alloc: 創建一個空的二進制
const buffer3 = Buffer.alloc(20) <Buffer 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00>
往buffer里面寫東西
buffer.write(string, offset): 寫入字符串buffer.writeInt8(value, offset): int8表示二進制8位(8位表示一個字節)所能表示的整數，offset開始寫入之前要跳過的字節數。buffer.writeInt16BE(value, offset): int16(兩個字節數)，表示16個二進制位所能表示的整數，即32767。超過這個數程序會報錯。
const buffer = Buffer.from([1, 2, 3, 4]) // <Buffer 01 02 03 04> // 往第二個字節里面寫入12 buffer.writeInt8(12, 1) // <Buffer 01 0c 03 04>
大端BE與小端LE：主要是對于2個以上字節的數據排列方式不同(writeInt8因為只有一個字節，所以沒有大端和小端)，大端的話就是低位地址放高位，小端就是低位地址放低位。如下：
const buffer = Buffer.from([1, 2, 3, 4]) buffer.writeInt16BE(512, 2) // <Buffer 01 02 02 00> buffer.writeInt16LE(512, 2) // <Buffer 01 02 00 02>
RPC傳輸的二進制如何表示傳遞的字段
PC傳輸的二進制是如何表示字段的呢？現在有個二進制包[00, 00, 00, 00, 00, 00, 00]，我們假定前三個字節表示一個字段值，后面兩個表示一個字段的值，最后兩個也表示一個字段的值。那寫法如下：
writeInt16BE(value, 0) writeInt16BE(value, 2) writeInt16BE(value, 4)
發現像這樣寫，不僅要知道寫入的值，還要知道值的數據類型，這樣就很麻煩。不如json格式那么方便。針對這種情況業界也有解決方案。npm有個庫protocol-buffers，把我們寫的參數轉化為buffer：
// test.proto 定義的協議文件 message Column { required float num = 1; required string payload = 2; } // index.js const fs = require('fs') var protobuf = require('protocol-buffers') var messages = protobuf(fs.readFileSync('test.proto')) var buf = messages.Column.encode({ num: 42, payload: 'hello world' }) console.log(buf) // <Buffer 0d 00 00 28 42 12 0b 68 65 6c 6c 6f 20 77 6f 72 6c 64> var obj = messages.Column.decode(buf) console.log(obj) // { num: 42, payload: 'hello world' }
net建立RPC通道
半雙工通信
服務端代碼:
const net = require('net') const LESSON_DATA = { 136797: '01 | 課程介紹', 136798: '02 | 內容綜述', 136799: '03 | Node.js是什么？', 136800: '04 | Node.js可以用來做什么？', 136801: '05 | 課程實戰項目介紹', 136803: '06 | 什么是技術預研？', 136804: '07 | Node.js開發環境安裝', 136806: '08 | 第一個Node.js程序：石頭剪刀布游戲', 136807: '09 | 模塊：CommonJS規范', 136808: '10 | 模塊：使用模塊規范改造石頭剪刀布游戲', 136809: '11 | 模塊：npm', 141994: '12 | 模塊：Node.js內置模塊', 143517: '13 | 異步：非阻塞I/O', 143557: '14 | 異步：異步編程之callback', 143564: '15 | 異步：事件循環', 143644: '16 | 異步：異步編程之Promise', 146470: '17 | 異步：異步編程之async/await', 146569: '18 | HTTP：什么是HTTP服務器？', 146582: '19 | HTTP：簡單實現一個HTTP服務器' } const server = net.createServer(socket => { // 監聽客戶端發送的消息 socket.on('data', buffer => { const lessonId = buffer.readInt32BE() setTimeout(() => { // 往客戶端發送消息 socket.write(LESSON_DATA[lessonId]) }, 1000) }) }) server.listen(4000)
客戶端代碼:
const net = require('net') const socket = new net.Socket({}) const LESSON_IDS = [ '136797', '136798', '136799', '136800', '136801', '136803', '136804', '136806', '136807', '136808', '136809', '141994', '143517', '143557', '143564', '143644', '146470', '146569', '146582' ] socket.connect({ host: '127.0.0.1', port: 4000 }) let buffer = Buffer.alloc(4) buffer.writeInt32BE(LESSON_IDS[Math.floor(Math.random() * LESSON_IDS.length)]) // 往服務端發送消息 socket.write(buffer) // 監聽從服務端傳回的消息 socket.on('data', buffer => { console.log(buffer.toString()) // 獲取到數據之后再次發送消息 buffer = Buffer.alloc(4) buffer.writeInt32BE(LESSON_IDS[Math.floor(Math.random() * LESSON_IDS.length)]) socket.write(buffer) })
以上半雙工通信步驟如下：
- 客戶端發送消息 socket.write(buffer)
- 服務端接受消息后往客戶端發送消息 socket.write(buffer)
- 客戶端接受消息后再次發送消息
這樣在一個時間端之內，只有一個端往另一個端發送消息，這樣就實現了半雙工通信。那如何實現全雙工通信呢，也就是在客戶端往服務端發送消息的同時，服務端還沒有消息返回給客戶端之前，客戶端又發送了一個消息給服務端。
全雙工通信
先來看一個場景:
客戶端發送了一個id1的請求，但是服務端還來不及返回，接著客戶端又發送了一個id2的請求。
等了一個之后，服務端先把id2的結果返回了，然后再把id1的結果返回。
那如何結果匹配到對應的請求上呢？
如果按照時間順序，那么id1的請求對應了id2的結果，因為id2是先返回的；id2的請求對應了id1的結果，這樣就導致請求包和返回包錯位的情況。
怎么辦呢？
我們可以給請求包和返回包都帶上序號，這樣就能對應上。
錯位處理
客戶端代碼:
socket.on('data', buffer => { // 包序號 const seqBuffer = buffer.slice(0, 2) // 服務端返回的內容 const titleBuffer = buffer.slice(2) console.log(seqBuffer.readInt16BE(), titleBuffer.toString()) }) // 包序號 let seq = 0 function encode(index) { // 請求包的長度現在是6 = 2(包序號) + 4(課程id) buffer = Buffer.alloc(6) buffer.writeInt16BE(seq) buffer.writeInt32BE(LESSON_IDS[index], 2) seq++ return buffer } // 每50ms發送一次請求 setInterval(() => { id = Math.floor(Math.random() * LESSON_IDS.length) socket.write(encode(id)) }, 50)
服務端代碼:
const server = net.createServer(socket => { socket.on('data', buffer => { // 把包序號取出 const seqBuffer = buffer.slice(0, 2) // 從第2個字節開始讀取 const lessonId = buffer.readInt32BE(2) setTimeout(() => { const buffer = Buffer.concat([ seqBuffer, Buffer.from(LESSON_DATA[lessonId]) ]) socket.write(buffer) // 這里返回時間采用隨機的，這樣就不會按順序返回，就可以測試錯位的情況 }, 10 + Math.random() * 1000) }) })
- 客戶端把包序號和對應的id給服務端
- 服務端取出包序號和對應的id，然后把包序號和id對應的內容返回給客戶端，同時設置返回的時間是隨機的，這樣就不會按照順序返回。
粘包處理
如果我們這樣發送請求:
for (let i = 0; i < 100; i++) { id = Math.floor(Math.random() * LESSON_IDS.length) socket.write(encode(id)) }
我們發現服務端接收到的信息如下:
<Buffer 00 00 00 02 16 64 00 01 00 02 16 68 00 02 00 02 31 1c 00 03 00 02 3c 96 00 04 00 02 16 68 00 05 00 02 16 5e 00 06 00 02 16 66 00 07 00 02 16 67 00 08 ... 550 more bytes>
這是因為TCP自己做的一個優化，它會把所有的請求包拼接在一起，這樣就會產生粘包的現象。
服務端需要把包進行拆分，拆分成100個小包。
那如何拆分呢？
首先客戶端發送的數據包包括兩部分：定長的包頭和不定長的包體。
包頭又分為兩部分：包序號及包體的長度。只有知道包體的長度，才能知道從哪里進行分割。
let seq = 0 function encode(data) { // 正常情況下，這里應該是使用 protocol-buffers 來encode一段代表業務數據的數據包 // 為了不要混淆重點，這個例子比較簡單，就直接把課程id轉buffer發送 const body = Buffer.alloc(4); body.writeInt32BE(LESSON_IDS[data.id]); // 一般來說，一個rpc調用的數據包會分為定長的包頭和不定長的包體兩部分 // 包頭的作用就是用來記載包的序號和包的長度，以實現全雙工通信 const header = Buffer.alloc(6); // 包序號占2個字節，包體長度占4個字節，共6個字節 header.writeInt16BE(seq) header.writeInt32BE(body.length, 2); // 包頭和包體拼起來發送 const buffer = Buffer.concat([header, body]) console.log(`包${seq}傳輸的課程id為${LESSON_IDS[data.id]}`); seq++; return buffer; } // 并發 for (let i = 0; i < 100; i++) { id = Math.floor(Math.random() * LESSON_IDS.length) socket.write(encode({ id })) }
服務端進行拆包
const server = net.createServer(socket => { let oldBuffer = null socket.on('data', buffer => { // 把上一次data事件使用殘余的buffer接上來 if (oldBuffer) { buffer = Buffer.concat([oldBuffer, buffer]) } let packageLength = 0 // 只要還存在可以解成完整包的包長 while ((packageLength = checkComplete(buffer))) { // 確定包的長度后進行slice分割 const package = buffer.slice(0, packageLength) // 剩余的包利用循環繼續分割 buffer = buffer.slice(packageLength) // 把這個包解成數據和seq const result = decode(package) // 計算得到要返回的結果，并write返回 socket.write(encode(LESSON_DATA[result.data], result.seq)) } // 把殘余的buffer記下來 oldBuffer = buffer }) })
checkComplete 函數的作用來確定一個數據包的長度，然后進行分割:
function checkComplete(buffer) { // 如果包的長度小于6個字節說明只有包頭，沒有包體，那么直接返回0 if (buffer.length <= 6) { return 0 } // 讀取包頭的第二個字節，取出包體的長度 const bodyLength = buffer.readInt32BE(2) // 請求包包括包頭(6個字節)和包體body return 6 + bodyLength }
decode對包進行解密:
function decode(buffer) { // 讀取包頭 const header = buffer.slice(0, 6) const seq = header.readInt16BE() // 讀取包體 // 正常情況下，這里應該是使用 protobuf 來decode一段代表業務數據的數據包 // 為了不要混淆重點，這個例子比較簡單，就直接讀一個Int32即可 const body = buffer.slice(6).readInt32BE() // 這里把seq和數據返回出去 return { seq, data: body } }
encode把客戶端想要的數據轉化為二進制返回，這個包同樣包括包頭和包體，包頭又包括包需要包序號和包體的長度。
function encode(data, seq) { // 正常情況下，這里應該是使用 protobuf 來encode一段代表業務數據的數據包 // 為了不要混淆重點，這個例子比較簡單，就直接把課程標題轉buffer返回 const body = Buffer.from(data) // 一般來說，一個rpc調用的數據包會分為定長的包頭和不定長的包體兩部分 // 包頭的作用就是用來記載包的序號和包的長度，以實現全雙工通信 const header = Buffer.alloc(6) header.writeInt16BE(seq) header.writeInt32BE(body.length, 2) const buffer = Buffer.concat([header, body]) return buffer }
當客戶端收到服務端發送的包之后，同樣也要進行拆包，因為所有的包同樣都粘在一起了:
<Buffer 00 00 00 00 00 1d 30 36 20 7c 20 e4 bb 80 e4 b9 88 e6 98 af e6 8a 80 e6 9c af e9 a2 84 e7 a0 94 ef bc 9f 00 01 00 00 00 1d 30 36 20 7c 20 e4 bb 80 e4 ... 539 more bytes>
因此，客戶端也需要拆包，拆包策略與服務端的拆包策略是一致的:
let oldBuffer = null socket.on('data', buffer => { // 把上一次data事件使用殘余的buffer接上來 if (oldBuffer) { buffer = Buffer.concat([oldBuffer, buffer]) } let completeLength = 0 // 只要還存在可以解成完整包的包長 while ((completeLength = checkComplete(buffer))) { const package = buffer.slice(0, completeLength) buffer = buffer.slice(completeLength) // 把這個包解成數據和seq const result = decode(package) console.log(`包${result.seq}，返回值是${result.data}`) } // 把殘余的buffer記下來 oldBuffer = buffer })
到這里就實現了雙全工通行，這樣客戶端和服務端隨時都可以往對方發小消息了。
