RPCとは何ですか?ノードにRPC通信を実装する方法について話しましょう

[関連チュートリアルの推奨事項: nodejs ビデオ チュートリアル]

RPC とは何ですか?

RPC: リモート プロシージャ コールは、リモート プロシージャ コールを指します。つまり、2 つのサーバー A と B があります。アプリケーションはサーバー A にデプロイされており、サーバー B 上のアプリケーションを呼び出したいとします。提供された関数/メソッドは、同じメモリ空間にないため、直接呼び出すことはできません。呼び出しのセマンティクスと呼び出しのデータは、ネットワークを通じて伝達される必要があります。

#サーバー間の通信

RPC と HTTP

同じ点

はすべて 2 台のコンピュータ間のネットワーク通信です。 Ajax はブラウザとサーバー間の通信であり、RPC はサーバーとサーバー間の通信です。
• データ形式について双方が同意する必要があります

相違点

アドレス指定サーバーが異なります

ajax は IP を取得するためのアドレス指定サービスとして DNS を使用しますドメイン名に対応するアドレス、参照 IP アドレスを取得した後、サーバーはデータを取得するリクエストを送信します。

RPC は通常、 イントラネット 内で相互にリクエストを行うため、通常はサービスのアドレス指定に DNS を使用しません。これは内部ネットワーク上にあるため、指定した ID または v5:8001 などの仮想 VIP を使用して、アドレス指定サーバーにアクセスして v5# に対応する IP アドレスを取得できます。 ##。

アプリケーション層プロトコルは異なります

ajax

テキスト プロトコルである http プロトコルを使用します。データを操作するとき、ファイル形式は html または json オブジェクトのいずれかで、json を使用する場合はキーと値の形式になります。

RPC

はバイナリプロトコルを採用しています。バイナリ送信を使用すると、送信されるパケットは [0001 0001 0111 0110 0010] のようになり、すべてバイナリです。通常、これらの数字はフィールドを表すために使用されます。たとえば、最初の 6 桁はフィールドなどです。 この方法では、http が json オブジェクト内のキーを送信する必要がないため、データ量が小さくなります。

送信はバイナリであるため、コンピュータが理解するのに適しており、テキスト プロトコルは人間の理解に適しているため、コンピュータが各フィールドを解釈するのにかかる時間はテキスト プロトコルよりもはるかに短くなります。

RPC はバイナリを使用することで、データ量を減らし、解釈速度を高速化します。

TCP 通信方式

シンプレックス通信: クライアントのみがサーバーにメッセージを送信できる、またはサーバーのみがメッセージを送信できるクライアントがサーバーにメッセージを送信
半二重通信: 一定期間内は、クライアントのみがサーバーにメッセージを送信できます。この期間が経過すると、サーバーはメッセージを送信できるようになります。クライアントへ。時間が多数のタイム スライスに分割されている場合、1 つのタイム スライス内では単信通信です。
全二重通信: クライアントとサーバーは相互に通信できます。
これら 3 つの通信方法のいずれかを選択する際に考慮すべき主な要素は、

実装の難しさとコスト

です。全二重通信は半二重通信よりもコストが高くなりますが、シナリオによっては半二重通信を検討することもできます。

ajax

は 半二重通信です。 http はテキスト プロトコルですが、その最下層は tcp プロトコルであり、http テキストは tcp 層でバイナリ データ フローからテキストへの変換プロセスを受けます。

RPC

を理解することは、フロントエンド テクノロジをより深く理解することです。

バッファのエンコードとバイナリ データ パケットのデコード

バッファの作成

buffer.from

: 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.write(string, offset)

: 文字列を書き込みます。
• buffer.writeInt8(value, offset)
: int8 は 8 ビット バイナリ ( 8 bit は 1 バイトで表現できる整数を表し、offset は書き込みを開始する前にスキップされるバイト数です。
• buffer.writeInt16BE(value, offset)
: int16 (2 バイト)。16 バイナリ ビットで表現できる整数、つまり 32767 を表します。この数値を超えると、プログラムはエラーを報告します。
• <pre class="brush:js;toolbar:false;">const buffer = Buffer.from([1, 2, 3, 4]) // &lt;Buffer 01 02 03 04&gt; // 往第二个字节里面写入12 buffer.writeInt8(12, 1) // &lt;Buffer 01 0c 03 04&gt;</pre>
ビッグエンディアン BE とリトルエンディアン LE

: 主な理由は、2 バイト以上のデータ配置が異なることです (writeInt8 は 1 バイトしかないため、ビッグエンディアンとリトルエンディアンはありません）、ビッグエンディアンでは下位アドレスが上位ビットに配置され、リトルエンディアンでは下位アドレスが下位ビットに配置されます。次のとおりです:

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] があります。最初の 3 バイトがフィールド値を表し、次の 2 バイトがフィールドの値を表し、最後の 2 バイトがフィールド値を表すと仮定します。フィールドの値も表します。記述方法は以下の通りです。

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: &#39;hello world&#39; }

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
})

到这里就实现了双全工通行，这样客户端和服务端随时都可以往对方发小消息了。

更多node相关知识，请访问：nodejs 教程！

以上がRPCとは何ですか?ノードにRPC通信を実装する方法について話しましょうの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。