自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

如何實現(xiàn)Nodejs進程間通信

作者： theanarkh
網絡 通信技術
對于有繼承關系的進程，nodejs本身為我們提供了進程間通信的方式，但是對于沒有繼承關系的進程，比如兄弟進程，想要通信最簡單的方式就是通過主進程中轉，類似前端框架中子組件通過更新父組件的數據，然后父通知其他子組件。

 [[350246]]

本文轉載自微信公眾號「編程雜技」，作者theanarkh 。轉載本文請聯(lián)系編程雜技公眾號。 

對于有繼承關系的進程，nodejs本身為我們提供了進程間通信的方式，但是對于沒有繼承關系的進程，比如兄弟進程，想要通信最簡單的方式就是通過主進程中轉，類似前端框架中子組件通過更新父組件的數據，然后父通知其他子組件。因為nodejs內置的進程間通信需要經過序列化和反序列化，所以這種方式可能會帶來一定的性能損耗，而且在實現(xiàn)上也比較麻煩。今天介紹的是實現(xiàn)兄弟進程通信的另外一種方式，在windows上使用命名管道，在非windows上使用unix域，另外本文還會介紹基于tcp的遠程進程通信的實現(xiàn)。下面具體介紹一下設計和實現(xiàn)。

1 IPC的實現(xiàn)

ipc的實現(xiàn)比較簡單，主要是對nodejs提供的功能進行封裝。首先我們需要處理一下path，因為在命名管道和unix域中他的格式是不一樣的。

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

接著我們看看客戶端和服務器的實現(xiàn)。

1.1 IPCClient的實現(xiàn)

  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的實現(xiàn)

  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的實現(xiàn)

我們知道tcp是面向流的服務，他本身只負責傳輸數據，不負責數據的解析和解釋。通過tcp傳輸數據時，需要自己解析數據，我們需要從一串字節(jié)流中解析出一個個數據包。這就涉及到協(xié)議的設計。所以首先我們要定義一個應用層協(xié)議。

2.1 應用層協(xié)議的設計和實現(xiàn)

null應用層協(xié)議的設計非常簡單

1 總長度是除了開頭標記之外的其他數據長度。因為數據部分是變長的，所以我們需要一個總長度來判斷后續(xù)的數據長度是多少。

2 序列號是用于關聯(lián)請求和響應，因為我們在一個連接上可能會串行發(fā)送多個數據包，當我們收到一個回包的時候，我們不知道是來自哪個請求的響應，通過響應體中的seq，我們就知道是來自哪個請求的響應。設計了通信協(xié)議后，我們就需要對協(xié)議進行封包解包。首先我們看一下封包邏輯。

  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.     // 把協(xié)議元數據和數據組裝到一起 
  21.     buffer = Buffer.concat([buffer, bufferData], buffer.byteLength + bufferData.byteLength); 
  22.     return buffer; 
  23. } 

接著我們看一下解包的邏輯，因為數據的傳輸是字節(jié)流，所以有可能多個數據包的數據會粘在一起，所以我們第一步首先要根據協(xié)議解析出一個個數據包，然后再解析每一個數據包。我們通過有限狀態(tài)機實現(xiàn)數據的解析。下面是狀態(tài)機的狀態(tài)集。

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

接著我們定義狀態(tài)集的轉換規(guī)則。

  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.             // 否則返回有效數據部分，繼續(xù)解析 
  16.             return [PARSE_STATE.PACKET_START, data.slice(position)]; 
  17.         } 
  18.         // 保存當前正在解析的數據包 
  19.         this.packet = new Packet(); 
  20.         // 跳過開始標記的字節(jié)數，進入解析協(xié)議頭階段 
  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.         // 數據部分的長度小于協(xié)議頭中定義的長度，則繼續(xù)等待 
  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. } 

我們再看一下狀態(tài)機的實現(xiàn)

  1. class FSM { 
  2.     constructor(options) { 
  3.         this.options = options; 
  4.         // 狀態(tài)處理機，定義了狀態(tài)轉移集合 
  5.         this.stateSwitcher = new StateSwitcher({cb: options.cb}); 
  6.         // 當前狀態(tài) 
  7.         this.state = PARSE_STATE.PARSE_INIT; 
  8.         // 結束狀態(tài) 
  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.         // 還沒結束，并且還有數據可以處理則繼續(xù)執(zhí)行 
  22.         while(this.state !== this.endState && this.buffer && this.buffer.length) { 
  23.             // 執(zhí)行狀態(tài)處理函數，返回[下一個狀態(tài), 剩下的數據] 
  24.             const result = this.stateSwitcher[this.state](this.buffer); 
  25.             // 如果下一個狀態(tài)是NEED_MORE_DATA則說明需要更多的數據才能繼續(xù)解析，并保持當前狀態(tài) 
  26.             if (result[0] === NEED_MORE_DATA) { 
  27.                 return; 
  28.             } 
  29.             // 記錄下一個狀態(tài)和數據 
  30.             [this.state, this.buffer] = result; 
  31.         } 
  32.  
  33.     } 
  34. } 

狀態(tài)機就是對開始狀態(tài)、結束狀態(tài)、狀態(tài)轉換集的封裝。實現(xiàn)了協(xié)議的封包和解析后我們看一下如何使用。

2.2 RPC客戶端實現(xiàn)

  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服務器實現(xiàn)

  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拓展

我們實現(xiàn)了數據的傳輸和解析，但是如何我們希望數據的請求和響應是一一對應的怎么辦呢?比如像http在tcp上可以并發(fā)發(fā)起多個請求一樣，響應是否可以亂序返回，我們又如何知道某個響應對應的是哪個請求?接下來介紹如何解決這個問題。首先我們實現(xiàn)一個請求管理的類。

  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.     // 執(zhí)行上下文 
  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拿到對應的上下文，從而執(zhí)行對應的回調。我們看看如何使用該類。

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的小型應用層協(xié)議，包含協(xié)議的定義、封包、解包功能)

 

責任編輯：武曉燕 來源： 編程雜技
Nodejs通信進程
相關推薦

2021-01-18 05:11:14

通信Nodejs進程

2013-03-28 13:14:45

AIDL進程間通信Android使用AI

2017-08-06 00:05:18

進程通信開發(fā)

2010-01-05 10:00:48

Linux進程間通信

2014-07-18 09:54:57

vlan路?由?器

2022-02-07 09:10:34

進程通信加鎖

2011-06-22 17:09:50

QT 進程 通信

2024-01-03 10:17:51

Linux通信

2021-01-22 10:58:16

網絡安全進程間碼如

2018-01-12 14:35:00

Linux進程共享內存

2017-06-19 13:36:12

Linux進程消息隊列

2019-11-08 14:47:49

TCPIP網絡

2019-05-08 11:10:05

Linux進程語言

2018-05-30 13:58:02

Linux進程通信

2011-06-24 14:01:34

Qt QCOP 協(xié)議

2020-11-18 09:06:04

Python

2020-12-01 09:22:43

進程協(xié)程開發(fā)

2022-07-04 08:29:13

electron通信

2019-11-27 10:36:11

進程通信IPC

2022-05-16 08:07:15

微服務容器通信
點贊
收藏

51CTO技術棧公眾號