Node.js高級編程使用RPC通信示例詳解

更新時間：2023年01月12日 11:45:02 作者：Aaaaaaaaaaayou

這篇文章主要為大家介紹了Node.js高級編程使用RPC通信示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前言

在構(gòu)建微服務(wù)時，為了追求極致的效率，服務(wù)間一般會使用 RPC（Remote Procedure Call）來進(jìn)行通信。本文通過 Node.js 來實踐一下。

Node.js 樸素 RPC

首先我們來構(gòu)建一下 server：

// server.js
const net = require('net')
const {msgBuffer} = require('../utils')
const server = net.createServer((clientSocket) => {
  clientSocket.on('data', (data) => {
    msgBuffer.push(data)
    while (!msgBuffer.isFinished()) {
      const message = JSON.parse(msgBuffer.handleData())
      clientSocket.write(
        JSON.stringify(fnMap[message.cmd].apply(null, message.params)) + '\n'
      )
    }
  })
})
server.listen(9999, () => console.log('Listening on 9999'))
const fnMap = {
  add: (...args) => {
    let s = 0
    for (let i = 0; i < args.length; i++) {
      s += args[i]
    }
    return s
  },
  multiply: (...args) => {
    let p = 1
    for (let i = 0; i < args.length; i++) {
      p *= args[i]
    }
    return p
  },
}
// MessageBuffer
class MessageBuffer {
  constructor(delimiter) {
    this.delimiter = delimiter
    this.buffer = ''
  }
  isFinished() {
    if (
      this.buffer.length === 0 ||
      this.buffer.indexOf(this.delimiter) === -1
    ) {
      return true
    }
    return false
  }
  push(data) {
    this.buffer += data
  }
  getMessage() {
    const delimiterIndex = this.buffer.indexOf(this.delimiter)
    if (delimiterIndex !== -1) {
      const message = this.buffer.slice(0, delimiterIndex)
      this.buffer = this.buffer.replace(message + this.delimiter, '')
      return message
    }
    return null
  }
  handleData() {
    const message = this.getMessage()
    return message
  }
}
exports.msgBuffer = new MessageBuffer('\n')

我們新建了一個 TCP 的服務(wù)，并監(jiān)聽來自客戶端的數(shù)據(jù)，注意這里我們通過一個 MessageBuffer 類來對數(shù)據(jù)進(jìn)行解析（至于為什么這么做可參考考文末補充內(nèi)容：關(guān)于 TCP “粘包”問題說明），將 TCP 數(shù)據(jù)流解析成我們的消息體。然后調(diào)用服務(wù)端預(yù)先配置好的方法，最后將返回值返回給客戶端。

客戶端相對比較簡單，將函數(shù)調(diào)用相關(guān)數(shù)據(jù)按照事先規(guī)定好的格式發(fā)送給服務(wù)端即可：

const net = require('net')
const {msgBuffer} = require('../utils')
const client = net.connect({port: 9999}, () => {
  client.write(JSON.stringify({cmd: 'add', params: [1, 2, 3]}) + '\n')
  client.write(JSON.stringify({cmd: 'multiply', params: [1, 2, 3]}) + '\n')
})
client.on('data', (data) => {
  msgBuffer.push(data)
  while (!msgBuffer.isFinished()) {
    const message = JSON.parse(msgBuffer.handleData())
    console.log(message)
  }
})

這樣，一個非常簡單的 RPC 雛形就出來了，不過目前這種方式還不是 RPC。所謂的 RPC，就是客戶端必須像調(diào)用本地方法一樣來調(diào)用遠(yuǎn)端的方法，而不是還需要自己組裝消息體，并監(jiān)聽事件獲取返回值。理想中的方式應(yīng)該像這樣：

const result = await client.add(1, 2, 3)

我們來改造一下。首先，我們定義一份配置文件，用來描述我們的 services：

// services/index.js
class Calculator {
  add(arr) {
    let s = 0
    for (let i = 0; i < arr.length; i++) {
      s += arr[i]
    }
    return s
  }
  multiply(arr) {
    let p = 1
    for (let i = 0; i < arr.length; i++) {
      p *= arr[i]
    }
    return p
  }
}
module.exports = {
  calculator: {
    cls: Calculator,
    methods: {
      add: {
        params: [{type: 'number[]', optional: false}],
        return: {
          type: 'number',
        },
      },
      multiply: {
        params: [{type: 'number[]', optional: false}],
        return: {
          type: 'number',
        },
      },
    },
  },
}

services 描述文件中包含了類以及它擁有的方法，方法參數(shù)（類型，是否可選），返回值類型等信息。為了簡單一點，我們先不校驗參數(shù)和返回值的類型。

然后就是我們的 server：

const net = require('net')
const {msgBuffer} = require('../utils')
const services = require('../services')
class Server {
  constructor(services) {
    this.tcpServer = net.createServer((clientSocket) => {
      const serviceMap = this.createServiceMap(services)
      clientSocket.on('data', (data) => {
        msgBuffer.push(data)
        while (!msgBuffer.isFinished()) {
          const {seqId, service, method, params} = JSON.parse(
            msgBuffer.handleData()
          )
          clientSocket.write(
            JSON.stringify({
              seqId,
              result: serviceMap[service][method].apply(null, params),
            }) + '\n'
          )
        }
      })
    })
  }
  createServiceMap(services) {
    const serviceMap = {}
    Object.keys(services).forEach((serviceKey) => {
      serviceMap[serviceKey] = new services[serviceKey].cls()
    })
    return serviceMap
  }
  listen(...args) {
    this.tcpServer.listen(...args)
  }
}
new Server(services).listen(9999)

server 中會監(jiān)聽 client 的連接，一旦有 client 進(jìn)來，就根據(jù) services 配置文件為其實例化所有 services。之后開始接受 client 的數(shù)據(jù)，并根據(jù) client 的消息調(diào)用相應(yīng)的 service 中的方法，并返回結(jié)果。

注意到消息體中有個 seqId，用來標(biāo)識包的序號，必須將其返回給 client，這樣 client 才能知道返回的結(jié)果是跟哪個請求對應(yīng)的。

最后就是我們的 client：

const net = require('net')
const EventEmitter = require('events')
const {msgBuffer} = require('../utils')
const services = require('../services')
class Client {
  constructor({port, services}) {
    this.rspResolve = {}
    this.seqId = 0
    this.port = port
    this.parseServices(services)
  }
  init() {
    return new Promise((resolve, reject) => {
      this.client = net.connect({port: this.port}, () => {
        resolve()
      })
      this.client.on('data', (data) => {
        msgBuffer.push(data)
        while (!msgBuffer.isFinished()) {
          const {seqId, result} = JSON.parse(msgBuffer.handleData())
          this.rspResolve[seqId](result)
        }
      })
    })
  }
  parseServices(services) {
    for (const serviceKey in services) {
      const service = services[serviceKey]
      this[serviceKey] = {}
      for (const method in service.methods) {
        this[serviceKey][method] = (...params) => {
          this.client.write(
            JSON.stringify({
              seqId: this.seqId,
              service: serviceKey,
              method,
              params,
            }) + '\n'
          )
          return new Promise((resolve, reject) => {
            this.rspResolve[this.seqId++] = resolve
          })
        }
      }
    }
  }
}
const client = new Client({port: 9999, services})
client.init().then(async () => {
  console.log(await client.calculator.add([1, 2, 3, 4, 5]))
  console.log(await client.calculator.multiply([1, 2, 3, 4, 5]))
})

初始化一個 client 時，會解析 services，并在當(dāng)前 client 實例上添加 services 的方法。方法中會將函數(shù)調(diào)用封裝成消息發(fā)送給服務(wù)端并返回 Promise 對象，同時將 Promise 對象的 resolve 方法緩存在 resResolve 這個 Map 中，此時 Promise 對象還處于 pending 狀態(tài)。

當(dāng) server 返回相應(yīng)的 seqId 的結(jié)果時，resResolve 中對應(yīng)的 resolve 方法會調(diào)用，從而將 Promise 對象狀態(tài)設(shè)為 fulfilled，此時 client 則可以獲取到結(jié)果。

這樣我們就實現(xiàn)了一個非常樸素的 RPC 框架。接下來我們簡單看看業(yè)界常用的 RPC 框架是怎么做的吧，這里以 Thrift 為例。

Thrift RPC Demo

我們先準(zhǔn)備一個 calculator.thrift 文件，用來描述 service：

service Calculator {
  i32 add(1:list&lt;i32&gt; arr),
  i32 multiply(1:list&lt;i32&gt; arr)
}

由于 thrift 文件是語言無關(guān)的，所以我們需要通過它生成對應(yīng) Calculator.js 文件：

thrift -r --gen js:node calculator.thrift

這個文件包含 server 端和 client 相關(guān)的代碼，在 client 端負(fù)責(zé)將函數(shù)調(diào)用轉(zhuǎn)為消息發(fā)送給 server，在 server 端負(fù)責(zé)讀取消息，調(diào)用方法，返回結(jié)果給 client。

然后 server 和 client 分別按照如下方式進(jìn)行使用即可：

// server.js
var thrift = require('thrift')
var Calculator = require('./gen-nodejs/Calculator')
var server = thrift.createServer(Calculator, {
  add(arr, result) {
    let s = 0
    for (let i = 0; i &lt; arr.length; i++) {
      s += arr[i]
    }
    result(null, s)
  },
  multiply(arr, result) {
    let p = 1
    for (let i = 0; i &lt; arr.length; i++) {
      p *= arr[i]
    }
    result(p)
  },
})
server.listen(9090)
// client.js
var thrift = require('thrift')
var Calculator = require('./gen-nodejs/Calculator')
var transport = thrift.TBufferedTransport
var protocol = thrift.TBinaryProtocol
var connection = thrift.createConnection('localhost', 9090, {
  transport: transport,
  protocol: protocol,
})
var client = thrift.createClient(Calculator, connection)
client.add([1, 2], function (err, response) {
  console.log(response)
})

下面，我們通過 Wireshark 來看看 thrift 通信的過程。

打開 Wireshark，選擇 Capturing from Loopback: lo0，然后在 filter 中輸入 tcp.port == 9090。分別運行上面的 server 和 client，則可抓包到如下內(nèi)容：

我們先來看看第五行，可以看到 Wireshark 自動識別了 thrift 協(xié)議，并解析出這是一個 CALL 類型的消息，調(diào)用的方法為 add。接下來我們再仔細(xì)看看 thrift 協(xié)議：

thrift 協(xié)議格式如上圖所示，這里是一個參數(shù)的場景，如果有多個參數(shù)的話則可以在 Data -> List 后面繼續(xù)添加，比如我們給 add 方法增加第二個參數(shù)，表示是否打印日志：

i32 add(1:list<i32> arr, 2:bool printLog)

抓包得到的內(nèi)容如下：

返回的消息格式也類似，這里就不贅述了。

關(guān)于 RPC 的內(nèi)容就先介紹到這，后面計劃基于 Nest.js 再實戰(zhàn)一下。

補充內(nèi)容

關(guān)于 TCP “粘包”問題說明

首先聲明一下，所謂的 TCP “粘包問題”其實并不是一個問題。

先看一個簡單的例子：

// server.js
const net = require('net')
const server = net.createServer((clientSocket) => {
  console.log('Client connected')
  clientSocket.on('data', (data) => {
    console.log('-------------------')
    console.log(data.toString())
  })
})
server.listen(9999, () => console.log('Listening on 9999'))
// client.js
const net = require('net')
const client = net.connect({port: 9999}, () => {
  client.write(JSON.stringify({cmd: 'add', params: [1, 2]}))
  client.write(JSON.stringify({cmd: 'multiply', params: [1, 2, 3]}))
})

啟動 server 后再運行 client，則 server 有可能會打印如下日志：

-------------------
{"cmd":"add","params":[1,2]}{"cmd":"multiply","params":[1,2,3]}

如上所示，客戶端調(diào)用了兩次 write，但是服務(wù)端卻只打印了一次。也就是說，兩次發(fā)送的數(shù)據(jù)在服務(wù)端被一次性取出來了。即，使用方層面的兩個包“粘在”了一起。原因在于 TCP 是面向字節(jié)流的，并沒有包的概念，所以開發(fā)者需要對 data 事件獲取到的數(shù)據(jù)進(jìn)行解析。

以上就是Node.js高級編程使用RPC通信示例詳解的詳細(xì)內(nèi)容，更多關(guān)于Node.js高級編程RPC通信的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: