1. 首页
  2. Node.js
  3. Node.js 之 RPC 调用

Node.js 之 RPC 调用

2022/07/29 22:15:51 发表 2022/07/29 22:15:51 更新 原创 雨辰 730 1 0

什么是 RPC 调用

RPC 全称为 Remote Procedure Call（远程过程调用）。

  • 是指两个计算机之间的网络通信。
  • 客户端与服务端需约定一个数据格式。
  • 基于 TCP 或 UDP 协议。

TCP 通信方式

单工通信

Client 端和 Server 端，只能单方向传输数据，不能双向传输。

Client --> Server或者Client <-- Server

单工通信

半双工通信

Client 端和 Server 端，可以双向交替传输数据。在同一时间段内，只能由其中一端向另外一端传输数据。

半双工通信

全双工通信

Client 端和 Server 端，可以双向同时传输数据。在同一时间段内，可双向交叉传输数据。

全双工通信

Node.js Buffer 模块

使用 Node.js Buffer 模块，编码/解码二进制数据包。

官方文档：Buffer | Node.js v14.20.0 Documentation

知识扩展：

Node.js Net 模块

使用 Node.js Net 模块，搭建多路复用的 RPC 通道。

示例：单工通信

由客户端向服务端传送数据。

客户端 Client

  1. const net = require('net');
  3. const socket = new net.Socket({});
  5. socket.connect({
  6.     host: '127.0.0.1',
  7.     port: 5000,
  8. });
  10. socket.write('Hello World!');

服务端 Server

  1. const net = require('net');
  3. const server = net.createServer((socket) => {
  4.     // 监听 data 事件
  5.     socket.on('data', (data) => {
  6.         // 参数 data 为 Buffer 类型
  7.         console.log(data, data.toString());
  8.     });
  9. });
  11. server.listen(5000);

示例：半双工通信

客户端 Client

  1. const net = require('net');
  2. const socket = new net.Socket({});
  4. // 建立连接
  5. socket.connect({
  6.     host: '127.0.0.1',
  7.     port: 5000,
  8. });
  10. // id 集合
  11. const ids = [
  12.     10001,
  13.     10002,
  14.     10003,
  15.     10004,
  16.     10005,
  17.     10006,
  18.     10007,
  19.     10008,
  20.     10009,
  21.     10010,
  22. ];
  24. // id 随机索引
  25. let index = Math.floor(Math.random() * ids.length);
  27. // 传送第一条数据
  28. socket.write(encode(index));
  30. // 监听从服务端返回的数据
  31. socket.on('data', (buffer) => {
  32.     console.log(buffer.toString());
  33.     // 接收到数据之后，按照半双工通信逻辑，开始下一次请求。
  34.     index = Math.floor(Math.random() * ids.length);
  35.     socket.write(encode(index));
  36. });
  38. // 编码请求包
  39. function encode(index) {
  40.     buffer = Buffer.alloc(4);
  41.     buffer.writeInt32BE(
  42.         ids[index]
  43.     );
  44.     return buffer;
  45. }

服务端 Server

  1. const net = require('net');
  3. const dataList = {
  4.     10001: '第 1 条数据',
  5.     10002: '第 2 条数据',
  6.     10003: '第 3 条数据',
  7.     10004: '第 4 条数据',
  8.     10005: '第 5 条数据',
  9.     10006: '第 6 条数据',
  10.     10007: '第 7 条数据',
  11.     10008: '第 8 条数据',
  12.     10009: '第 9 条数据',
  13.     10010: '第 10 条数据',
  14. };
  16. const server = net.createServer((socket) => {
  17.     socket.on('data', (buffer) => {
  18.         const id = buffer.readInt32BE();
  19.         // 伪造延迟 100 毫秒
  20.         setTimeout(() => {
  21.             socket.write(
  22.                 Buffer.from(dataList[id])
  23.             );
  24.         }, 100);
  25.     })
  26. });
  28. server.listen(5000);

示例：全双工通信

客户端 Client

  1. const net = require('net');
  2. const socket = new net.Socket({});
  4. socket.connect({
  5.     host: '127.0.0.1',
  6.     port: 4000
  7. });
  9. const LESSON_IDS = [
  10.     10001,
  11.     10002,
  12.     10003,
  13.     10004,
  14.     10005,
  15.     10006,
  16.     10007,
  17.     10008,
  18.     10009,
  19.     10010,
  20. ]
  22. let id = Math.floor(Math.random() * LESSON_IDS.length);
  24. let oldBuffer = null;
  25. socket.on('data', (buffer) => {
  26.     // 把上一次data事件使用残余的buffer接上来
  27.     if (oldBuffer) {
  28.         buffer = Buffer.concat([oldBuffer, buffer]);
  29.     }
  30.     let completeLength = 0;
  32.     // 只要还存在可以解成完整包的包长
  33.     while (completeLength = checkComplete(buffer)) {
  34.         const package = buffer.slice(0, completeLength);
  35.         buffer = buffer.slice(completeLength);
  37.         // 把这个包解成数据和seq
  38.         const result = decode(package);
  39.         console.log(`包${result.seq}，返回值是${result.data}`);
  40.     }
  42.     // 把残余的buffer记下来
  43.     oldBuffer = buffer;
  44. })
  47. let seq = 0;
  48. /**
  49.  * 二进制包编码函数
  50.  * 在一段rpc调用里，客户端需要经常编码rpc调用时，业务数据的请求包
  51.  */
  52. function encode(data) {
  53.     // 正常情况下，这里应该是使用 protobuf 来encode一段代表业务数据的数据包
  54.     // 为了不要混淆重点，这个例子比较简单，就直接把课程id转buffer发送
  55.     const body = Buffer.alloc(4);
  56.     body.writeInt32BE(LESSON_IDS[data.id]);
  58.     // 一般来说，一个rpc调用的数据包会分为定长的包头和不定长的包体两部分
  59.     // 包头的作用就是用来记载包的序号和包的长度，以实现全双工通信
  60.     const header = Buffer.alloc(6);
  61.     header.writeInt16BE(seq)
  62.     header.writeInt32BE(body.length, 2);
  64.     // 包头和包体拼起来发送
  65.     const buffer = Buffer.concat([header, body])
  67.     console.log(`包${seq}传输的课程id为${LESSON_IDS[data.id]}`);
  68.     seq++;
  69.     return buffer;
  70. }
  72. /**
  73.  * 二进制包解码函数
  74.  * 在一段rpc调用里，客户端需要经常解码rpc调用时，业务数据的返回包
  75.  */
  76. function decode(buffer) {
  77.     const header = buffer.slice(0, 6);
  78.     const seq = header.readInt16BE();
  79.     const body = buffer.slice(6)
  81.     return {
  82.         seq,
  83.         data: body.toString()
  84.     }
  85. }
  87. /**
  88.  * 检查一段buffer是不是一个完整的数据包。
  89.  * 具体逻辑是：判断header的bodyLength字段，看看这段buffer是不是长于header和body的总长
  90.  * 如果是，则返回这个包长，意味着这个请求包是完整的。
  91.  * 如果不是，则返回0，意味着包还没接收完
  92.  * @param {} buffer 
  93.  */
  94. function checkComplete(buffer) {
  95.     if (buffer.length < 6) {
  96.         return 0;
  97.     }
  98.     const bodyLength = buffer.readInt32BE(2);
  99.     return 6 + bodyLength
  100. }
  103. for (let k = 0; k < 100; k++) {
  104.     id = Math.floor(Math.random() * LESSON_IDS.length);
  105.     socket.write(encode({ id }));
  106. }

服务端 Server

  1. const net = require('net');
  3. const server = net.createServer((socket) => {
  5.     let oldBuffer = null;
  6.     socket.on('data', function (buffer) {
  7.         // 把上一次data事件使用残余的buffer接上来
  8.         if (oldBuffer) {
  9.             buffer = Buffer.concat([oldBuffer, buffer]);
  10.         }
  12.         let packageLength = 0;
  13.         // 只要还存在可以解成完整包的包长
  14.         while (packageLength = checkComplete(buffer)) {
  15.             const package = buffer.slice(0, packageLength);
  16.             buffer = buffer.slice(packageLength);
  18.             // 把这个包解成数据和seq
  19.             const result = decode(package);
  21.             // 计算得到要返回的结果，并write返回
  22.             socket.write(
  23.                 encode(LESSON_DATA[result.data], result.seq)
  24.             );
  25.         }
  27.         // 把残余的buffer记下来
  28.         oldBuffer = buffer;
  29.     })
  31. });
  33. server.listen(4000);
  35. /**
  36.  * 二进制包编码函数
  37.  * 在一段rpc调用里，服务端需要经常编码rpc调用时，业务数据的返回包
  38.  */
  39. function encode(data, seq) {
  40.     // 正常情况下，这里应该是使用 protobuf 来encode一段代表业务数据的数据包
  41.     // 为了不要混淆重点，这个例子比较简单，就直接把课程标题转buffer返回
  42.     const body = Buffer.from(data)
  44.     // 一般来说，一个rpc调用的数据包会分为定长的包头和不定长的包体两部分
  45.     // 包头的作用就是用来记载包的序号和包的长度，以实现全双工通信
  46.     const header = Buffer.alloc(6);
  47.     header.writeInt16BE(seq)
  48.     header.writeInt32BE(body.length, 2);
  50.     const buffer = Buffer.concat([header, body])
  52.     return buffer;
  53. }
  55. /**
  56.  * 二进制包解码函数
  57.  * 在一段rpc调用里，服务端需要经常解码rpc调用时，业务数据的请求包
  58.  */
  59. function decode(buffer) {
  60.     const header = buffer.slice(0, 6);
  61.     const seq = header.readInt16BE();
  63.     // 正常情况下，这里应该是使用 protobuf 来decode一段代表业务数据的数据包
  64.     // 为了不要混淆重点，这个例子比较简单，就直接读一个Int32即可
  65.     const body = buffer.slice(6).readInt32BE()
  67.     // 这里把seq和数据返回出去
  68.     return {
  69.         seq,
  70.         data: body
  71.     }
  72. }
  74. /**
  75.  * 检查一段buffer是不是一个完整的数据包。
  76.  * 具体逻辑是：判断header的bodyLength字段，看看这段buffer是不是长于header和body的总长
  77.  * 如果是，则返回这个包长，意味着这个请求包是完整的。
  78.  * 如果不是，则返回0，意味着包还没接收完
  79.  * @param {} buffer 
  80.  */
  81. function checkComplete(buffer) {
  82.     if (buffer.length < 6) {
  83.         return 0;
  84.     }
  85.     const bodyLength = buffer.readInt32BE(2);
  86.     return 6 + bodyLength
  87. }
  89. // 假数据
  90. const LESSON_DATA = {
  91.     10001: '第 1 条数据',
  92.     10002: '第 2 条数据',
  93.     10003: '第 3 条数据',
  94.     10004: '第 4 条数据',
  95.     10005: '第 5 条数据',
  96.     10006: '第 6 条数据',
  97.     10007: '第 7 条数据',
  98.     10008: '第 8 条数据',
  99.     10009: '第 9 条数据',
  100.     10010: '第 10 条数据',
  101. }

（完）