当前位置：主页 > 查看内容

如何实现Nodejs进程间通信

发布时间：2021-06-24 00:00| 有 位朋友查看

简介：本文转载自微信公众号「编程杂技」，作者theanarkh 。转载本文请联系编程杂技公众号。 对于有继承关系的进程，nodejs本身为我们提供了进程间通信的方式，但是对于没有继承关系的进程，比如兄弟进程，想要通信最简单的方式就是通过主进程中转，类似前端框架中……

 

本文转载自微信公众号「编程杂技」，作者theanarkh 。转载本文请联系编程杂技公众号。

对于有继承关系的进程，nodejs本身为我们提供了进程间通信的方式，但是对于没有继承关系的进程，比如兄弟进程，想要通信最简单的方式就是通过主进程中转，类似前端框架中子组件通过更新父组件的数据，然后父通知其他子组件。因为nodejs内置的进程间通信需要经过序列化和反序列化，所以这种方式可能会带来一定的性能损耗，而且在实现上也比较麻烦。今天介绍的是实现兄弟进程通信的另外一种方式，在windows上使用命名管道，在非windows上使用unix域，另外本文还会介绍基于tcp的远程进程通信的实现。下面具体介绍一下设计和实现。

1 IPC的实现

ipc的实现比较简单，主要是对nodejs提供的功能进行封装。首先我们需要处理一下path，因为在命名管道和unix域中他的格式是不一样的。

  1. const os = require('os'); 
  2.  
  3. module.exports = { 
  4.     path: os.platform() === 'win32' ? '\\\\?\\pipe\\ipc' : '/tmp/unix.sock', 
  5. }; 

接着我们看看客户端和服务器的实现。

1.1 IPCClient的实现

  1. const net = require('net'); 
  2. const { EventEmitter } = require('events'); 
  3. const { path } = require('../config'); 
  4.  
  5. class Client extends EventEmitter { 
  6.   constructor(options) { 
  7.     super(); 
  8.     this.options = { path, ...options }; 
  9.     const socket = net.connect(this.options); 
  10.     socket.on('error', (error) => { 
  11.       console.error(error); 
  12.     }); 
  13.     return socket;  
  14.   } 
  15. } 
  16. module.exports = { 
  17.     Client, 
  18. }; 

1.2 IPCServer的实现

  1. const fs = require('fs'); 
  2. const net = require('net'); 
  3. const { EventEmitter } = require('events'); 
  4. const { path } = require('../config'); 
  5.  
  6. class Server extends EventEmitter { 
  7.     constructor(options, connectionListener) { 
  8.       super(); 
  9.       if (typeof options === 'function') { 
  10.         options = { 
  11.           connectionListener: options, 
  12.         }; 
  13.       } else { 
  14.         options = { ...options, connectionListener }; 
  15.       } 
  16.       try { 
  17.         fs.existsSync(options.path) && fs.unlinkSync(options.path); 
  18.       } catch(e) { 
  19.  
  20.       } 
  21.       this.options = { path, ...options }; 
  22.       return net.createServer({allowHalfOpen: this.options.allowHalfOpen, pauseOnConnect: this.options.pauseOnConnect}, (client) => { 
  23.         client.on('error', (error) => { 
  24.           console.error(error); 
  25.         }); 
  26.         typeof this.options.connectionListener === 'function' && this.options.connectionListener(client); 
  27.       }).listen(this.options); 
  28.     } 
  29. } 
  30.  
  31. module.exports = { 
  32.     Server, 
  33. }; 

2 RPC的实现

我们知道tcp是面向流的服务，他本身只负责传输数据，不负责数据的解析和解释。通过tcp传输数据时，需要自己解析数据，我们需要从一串字节流中解析出一个个数据包。这就涉及到协议的设计。所以首先我们要定义一个应用层协议。

2.1 应用层协议的设计和实现

null应用层协议的设计非常简单

1 总长度是除了开头标记之外的其他数据长度。因为数据部分是变长的，所以我们需要一个总长度来判断后续的数据长度是多少。

2 序列号是用于关联请求和响应，因为我们在一个连接上可能会串行发送多个数据包，当我们收到一个回包的时候，我们不知道是来自哪个请求的响应，通过响应体中的seq，我们就知道是来自哪个请求的响应。设计了通信协议后，我们就需要对协议进行封包解包。首先我们看一下封包逻辑。

  1. function seq() { 
  2.    return ~~(Math.random() * Math.pow(2, 31)) 
  3. } 
  4.  
  5. function packet(data, sequnce) { 
  6.     // 转成buffer 
  7.     const bufferData = Buffer.from(data, 'utf-8'); 
  8.     // 开始标记长度 
  9.     const startFlagLength = Buffer.from([PACKET_START]).byteLength; 
  10.     // 序列号 
  11.     const _seq = sequnce || seq(); 
  12.     // 分配一个buffer存储数据 
  13.     let buffer = Buffer.allocUnsafe(startFlagLength + TOTAL_LENGTH + SEQ_LEN); 
  14.     // 设计开始标记 
  15.     buffer[0] = 0x3; 
  16.     // 写入总长度字段的值 
  17.     buffer.writeUIntBE(TOTAL_LENGTH + SEQ_LEN + bufferData.byteLength, 1, TOTAL_LENGTH); 
  18.     // 写入序列号的值 
  19.     buffer.writeUIntBE(_seq, startFlagLength + TOTAL_LENGTH, SEQ_LEN); 
  20.     // 把协议元数据和数据组装到一起 
  21.     buffer = Buffer.concat([buffer, bufferData], buffer.byteLength + bufferData.byteLength); 
  22.     return buffer; 
  23. } 

接着我们看一下解包的逻辑，因为数据的传输是字节流，所以有可能多个数据包的数据会粘在一起，所以我们第一步首先要根据协议解析出一个个数据包，然后再解析每一个数据包。我们通过有限状态机实现数据的解析。下面是状态机的状态集。

  1. const PARSE_STATE = { 
  2.   PARSE_INIT: 0, 
  3.   PARSE_HEADER: 1, 
  4.   PARSE_DATA: 2, 
  5.   PARSE_END: 3, 
  6. }; 

接着我们定义状态集的转换规则。

  1. class StateSwitcher { 
  2.     constructor(options) { 
  3.         this.options = options; 
  4.     } 
  5.  
  6.     [PARSE_STATE.PARSE_INIT](data) { 
  7.         // 数据不符合预期 
  8.         if (data[0] !== PACKET_START) { 
  9.             // 跳过部分数据，找到开始标记 
  10.             const position = data.indexOf(PACKET_START); 
  11.             // 没有开始标记，说明这部分数据无效，丢弃 
  12.             if (position === -1) { 
  13.                 return [NEED_MORE_DATA, null]; 
  14.             } 
  15.             // 否则返回有效数据部分，继续解析 
  16.             return [PARSE_STATE.PACKET_START, data.slice(position)]; 
  17.         } 
  18.         // 保存当前正在解析的数据包 
  19.         this.packet = new Packet(); 
  20.         // 跳过开始标记的字节数，进入解析协议头阶段 
  21.         return [PARSE_STATE.PARSE_HEADER, data.slice(Buffer.from([PACKET_START]).byteLength)]; 
  22.     }  
  23.  
  24.     [PARSE_STATE.PARSE_HEADER](data) { 
  25.         // 数据不够头部的大小则等待数据到来 
  26.         if (data.length < TOTAL_LENGTH + SEQ_LEN) { 
  27.           return [NEED_MORE_DATA, data]; 
  28.         } 
  29.         // 有效数据包的长度 = 整个数据包长度 - 头部长度 
  30.         this.packet.set('length', data.readUInt32BE() - (TOTAL_LENGTH + SEQ_LEN)); 
  31.         // 序列号 
  32.         this.packet.set('seq', data.readUInt32BE(TOTAL_LENGTH)); 
  33.         // 解析完头部了，跳过去 
  34.         data = data.slice(TOTAL_LENGTH + SEQ_LEN); 
  35.         // 进入解析数据阶段 
  36.         return [PARSE_STATE.PARSE_DATA, data]; 
  37.     } 
  38.  
  39.     [PARSE_STATE.PARSE_DATA](data) { 
  40.         const len = this.packet.get('length'); 
  41.         // 数据部分的长度小于协议头中定义的长度，则继续等待 
  42.         if (data.length < len) { 
  43.             return [NEED_MORE_DATA, data]; 
  44.         } 
  45.         // 截取数据部分 
  46.         this.packet.set('data', data.slice(0, len)); 
  47.         // 解析完数据了，完成一个包的解析，跳过数据部分 
  48.         data = data.slice(len); 
  49.         typeof this.options.cb === 'function' && this.options.cb(this.packet); 
  50.         this.packet = null; 
  51.         // 解析完一个数据包，进入结束标记阶段 
  52.         return [PARSE_STATE.PARSE_INIT, data]; 
  53.     } 
  54. } 

我们再看一下状态机的实现

  1. class FSM { 
  2.     constructor(options) { 
  3.         this.options = options; 
  4.         // 状态处理机，定义了状态转移集合 
  5.         this.stateSwitcher = new StateSwitcher({cb: options.cb}); 
  6.         // 当前状态 
  7.         this.state = PARSE_STATE.PARSE_INIT; 
  8.         // 结束状态 
  9.         this.endState = PARSE_STATE.PARSE_END; 
  10.         // 当前待解析的数据 
  11.         this.buffer = null; 
  12.     } 
  13.  
  14.     run(data) { 
  15.         // 没有数据或者解析结束了直接返回 
  16.         if (this.state === this.endState || !data || !data.length) { 
  17.             return; 
  18.         } 
  19.         // 保存待解析的数据 
  20.         this.buffer = this.buffer ? Buffer.concat([this.buffer, data]) : data; 
  21.         // 还没结束，并且还有数据可以处理则继续执行 
  22.         while(this.state !== this.endState && this.buffer && this.buffer.length) { 
  23.             // 执行状态处理函数，返回[下一个状态, 剩下的数据] 
  24.             const result = this.stateSwitcher[this.state](this.buffer); 
  25.             // 如果下一个状态是NEED_MORE_DATA则说明需要更多的数据才能继续解析，并保持当前状态 
  26.             if (result[0] === NEED_MORE_DATA) { 
  27.                 return; 
  28.             } 
  29.             // 记录下一个状态和数据 
  30.             [this.state, this.buffer] = result; 
  31.         } 
  32.  
  33.     } 
  34. } 

状态机就是对开始状态、结束状态、状态转换集的封装。实现了协议的封包和解析后我们看一下如何使用。

2.2 RPC客户端实现

  1. const net = require('net'); 
  2. const { EventEmitter } = require('events'); 
  3. const { FSM } = require('tiny-application-layer-protocol'); 
  4. class Client extends EventEmitter { 
  5.   constructor(options) { 
  6.     super(); 
  7.     this.options = { ...options }; 
  8.     const socket = net.connect(this.options); 
  9.     socket.on('error', (error) => { 
  10.       console.error(error); 
  11.     }); 
  12.     const fsm = new FSM({ 
  13.       cb: (packet) => { 
  14.         socket.emit('message', packet); 
  15.       } 
  16.     }); 
  17.     socket.on('data', fsm.run.bind(fsm)); 
  18.     return socket;  
  19.   } 
  20. } 
  21. module.exports = { 
  22.     Client, 
  23. }; 

我们做的事情主要是负责数据的解析。

2.3 RPC服务器实现

  1. const fs = require('fs'); 
  2. const net = require('net'); 
  3. const { EventEmitter } = require('events') 
  4. const { FSM } = require('tiny-application-layer-protocol'); 
  5.  
  6. class Server extends EventEmitter { 
  7.     constructor(options, connectionListener) { 
  8.       super(); 
  9.       if (typeof options === 'function') { 
  10.         options = { 
  11.           connectionListener: options, 
  12.         }; 
  13.       } else { 
  14.         options = { ...options, connectionListener }; 
  15.       } 
  16.       this.options = { ...options }; 
  17.       return net.createServer({allowHalfOpen: this.options.allowHalfOpen, pauseOnConnect: this.options.pauseOnConnect}, (client) => { 
  18.         const fsm = new FSM({ 
  19.             cb: function(packet) { 
  20.               client.emit('message', packet); 
  21.             } 
  22.         }) 
  23.         client.on('data', fsm.run.bind(fsm)); 
  24.         client.on('error', (error) => { 
  25.           console.error(error); 
  26.         }); 
  27.         typeof this.options.connectionListener === 'function' && this.options.connectionListener(client); 
  28.       }).listen(this.options); 
  29.     } 
  30. } 
  31.  
  32. module.exports = { 
  33.     Server, 
  34. }; 

同样，服务器也是负责数据的解析

3 使用

接下来我们看一下如何使用。

3.1 ipc的使用

server.js

  1. const { IPCServer } = require('../../src'); 
  2. const { packet } = require('tiny-application-layer-protocol'); 
  3. new IPCServer(function(client) { 
  4.     console.log(1) 
  5.     client.on('data', (data) => { 
  6.         console.log('receive', data); 
  7.         client.write(packet('world', data.seq)); 
  8.     }); 
  9. }); 

client.js

  1. const { IPCClient } = require('../../src'); 
  2. const { packet, seq } = require('tiny-application-layer-protocol'); 
  3. const client = new IPCClient(); 
  4. client.write(packet('hello', seq())); 
  5. client.on('data', function(res) { 
  6.   console.log('receive', res); 
  7. }) 

服务器输出

客户端输出

3.2 RPC的使用

server.js

  1. const { RPCServer } = require('../../src'); 
  2. const { packet } = require('tiny-application-layer-protocol'); 
  3. new RPCServer({host: '127.0.0.1', port: 80}, function(client) { 
  4.     client.on('message', (data) => { 
  5.         console.log('receive', data); 
  6.         client.write(packet('world', data.seq)); 
  7.     }); 
  8. }); 

client.js

  1. const { RPCClient } = require('../../src'); 
  2. const { packet, seq } = require('tiny-application-layer-protocol'); 
  3. const client = new RPCClient({host: '127.0.0.1', port: 80}); 
  4. client.write(packet('hello', seq())); 
  5. client.on('message', function(res) { 
  6.   console.log('receive', res); 
  7. }) 

服务器输出

客户端输出

4 RPC拓展

我们实现了数据的传输和解析，但是如何我们希望数据的请求和响应是一一对应的怎么办呢?比如像http在tcp上可以并发发起多个请求一样，响应是否可以乱序返回，我们又如何知道某个响应对应的是哪个请求?接下来介绍如何解决这个问题。首先我们实现一个请求管理的类。

  1. class RequestManager { 
  2.     constructor(options) { 
  3.         this.options = { timeout: 10000, ...options }; 
  4.         this.map = {}; 
  5.         this.timerId = null; 
  6.         this.startPollTimeout(); 
  7.     } 
  8.     set(key, context) { 
  9.         if (typeof context.cb !== 'function') { 
  10.             throw new Error('cb is required'); 
  11.         } 
  12.         this.map[key] = { 
  13.             startTime: Date.now(), 
  14.             ...context, 
  15.         }; 
  16.     } 
  17.     get(key) { 
  18.         return this.map[key]; 
  19.     } 
  20.     del(key) { 
  21.         return delete this.map[key]; 
  22.     } 
  23.     // 执行上下文 
  24.     exec(key, data) { 
  25.         const context = this.get(key); 
  26.         if (context) { 
  27.             this.del(key); 
  28.             context.cb(data); 
  29.         } 
  30.     }  
  31.     execAll(data) { 
  32.         for (const [key] of Object.entries(this.map)) { 
  33.             this.exec(key, data); 
  34.         } 
  35.     }  
  36.     // 定时轮询是否超时 
  37.     startPollTimeout() { 
  38.         this.timerId = setTimeout(() => { 
  39.             if (!this.timerId) { 
  40.                 return; 
  41.             } 
  42.             const nextMap = {}; 
  43.             for (const [key, context] of Object.entries(this.map)) { 
  44.                 if (Date.now() - context.startTime < (context.timeout || this.options.timeout)) { 
  45.                     nextMap[key] = context; 
  46.                 } else { 
  47.                     context.cb(new Error('timeout')); 
  48.                 } 
  49.             } 
  50.             this.map = nextMap; 
  51.             this.startPollTimeout(); 
  52.         }, 1000); 
  53.     } 
  54. } 

该类的逻辑主要是请求的seq保存对应的上下文，然后收到响应的时候，我们根据响应的seq拿到对应的上下文，从而执行对应的回调。我们看看如何使用该类。

server.js

  1. const { RPCServer } = require('../../src'); 
  2. const { packet } = require('tiny-application-layer-protocol'); 
  3. new RPCServer({host: '127.0.0.1', port: 80}, function(client) { 
  4.     client.on('message', (data) => { 
  5.         console.log('receive', data); 
  6.         client.end(packet('world', data.seq)); 
  7.     }); 
  8.     client.on('end', (data) => { 
  9.         client.end(); 
  10.     }); 
  11. }); 

client.js

  1. const { RPCClient, RequestManager } = require('../../src'); 
  2. const { packet, seq } = require('tiny-application-layer-protocol'); 
  3. const requestManager = new RequestManager({timeout: 3000}); 
  4. const client = new RPCClient({host: '127.0.0.1', port: 80}); 
  5. const _seq = seq();  
  6. requestManager.set(_seq, { 
  7.   cb: function() { 
  8.     console.log(...arguments); 
  9.   } 
  10. }) 
  11. client.write(packet('hello', _seq)); 
  12. client.on('message', function(packet) { 
  13.   requestManager.exec(packet.seq, packet); 
  14. }) 

输出 服务器输出

客户端输出null

github仓库：https://github.com/theanarkh/nodejs-ipc

github仓库：https://github.com/theanarkh/tiny-application-layer-protocol

npm install nodejs-i-p-c(ipc和rpc库，依赖tiny-application-layer-protocol)

npm install tiny-application-layer-protocol(基于tcp的小型应用层协议，包含协议的定义、封包、解包功能)


本文转载自网络，原文链接：https://mp.weixin.qq.com/s/OsP1rELxo66etJWRjZSniw
本站部分内容转载于网络，版权归原作者所有，转载之目的在于传播更多优秀技术内容，如有侵权请联系QQ/微信：153890879删除，谢谢！
上一篇：不用担忧，明日5G或是“今日高铁” 下一篇：手机App无授权就能监听电话 成功率高达90%：罪魁祸首你想象不到

推荐图文

  • 周排行
  • 月排行
  • 总排行

随机推荐