Nodejs-Ipc的設計與實現

作者：theanarkh 2020-10-23 08:31:15

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

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

對于有繼承關系的進程，nodejs本身為我們提供了進程間通信的方式，但是對于沒有繼承關系的進程，比如兄弟進程，想要通信最簡單的方式就是通過主進程中轉，類似前端框架中子組件通過更新父組件的數據，然后父通知其他子組件。因為nodejs的進程間通信需要經過序列化和反序列化，所以這種方式可能會帶來一定的性能損耗，而且在實現上也比較麻煩。今天介紹的是兄弟進程直接通信的方式。在windows上使用tcp，在非windows上使用unix域，本機通信，unix域性能上會更好，因為tcp通信需要經過協議棧的封包和解包。下面具體介紹一下這個ipc庫的設計和實現。

設計思想主要是一個進程啟動一個服務，然后其他客戶端進程可以通過地址信息去和服務器建立長連接。這里沒有使用短連接，短連接雖然在實現上會變得容易，但是頻繁通信的進程，不斷地創建和銷毀數據結構會帶來一定的開銷，長連接雖然會一直占用內存，但是這是非常小的，而長連接帶來的效率明顯會更好。不過長連接會帶來一個難點，那就是對數據的解析，比如對于tcp來說，我們從中拿到的是一串字節流，這一串字節流中可能有幾個數據包的數據，我們需要從這一串字節流中解析出一個個數據包。這就涉及到協議的設計。所以首先我們要定義一個應用層協議。

1 應用層協議的設計和實現

null應用層協議的設計非常簡單

總長度是除了開頭標記之外的其他數據長度。因為數據部分是變長的，所以我們需要一個總長度來判斷后續的數據長度是多少。
序列號是用于關聯請求和響應，因為我們在一個連接上可能會串行發送多個數據包，當我們收到一個回包的時候，我們不知道是來自哪個請求的響應，通過響應體中的seq，我們就知道是來自哪個請求的響應。設計了通信協議后，我們就需要對協議進行封包解包。首先我們看一下封包邏輯。

function seq() { 
   return ~~(Math.random() * Math.pow(2, 31)) 
} 
function packet(data, sequnce) { 
    // 轉成buffer 
    const bufferData = Buffer.from(data, 'utf-8'); 
    // 開始標記長度 
    const startFlagLength = Buffer.from([PACKET_START]).byteLength; 
    // 序列號 
    const seq = sequnce || seq(); 
    // 分配一個buffer存儲數據 
    let buffer = Buffer.allocUnsafe(startFlagLength + TOTAL_LENGTH + SEQ_LEN); 
    // 設計開始標記 
    buffer[0] = 0x3; 
    // 寫入總長度字段的值 
    buffer.writeUIntBE(TOTAL_LENGTH + SEQ_LEN + bufferData.byteLength, 1, TOTAL_LENGTH); 
    // 寫入序列號的值 
    buffer.writeUIntBE(seq, startFlagLength + TOTAL_LENGTH, SEQ_LEN); 
    // 把協議元數據和數據組裝到一起 
    buffer = Buffer.concat([buffer, bufferData], buffer.byteLength + bufferData.byteLength); 
    return buffer; 
}

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

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

接著我們定義狀態集的轉換規則。

class StateSwitcher { 
    constructor(options) { 
        this.options = options; 
    } 
 
    [PARSE_STATE.PARSE_INIT](data) { 
        // 數據不符合預期 
        if (data[0] !== PACKET_START) { 
            // 跳過部分數據，找到開始標記 
            const position = data.indexOf(PACKET_START); 
            // 沒有開始標記，說明這部分數據無效，丟棄 
            if (position === -1) { 
                return [NEED_MORE_DATA, null]; 
            } 
            // 否則返回有效數據部分，繼續解析 
            return [PARSE_STATE.PACKET_START, data.slice(position)]; 
        } 
        // 保存當前正在解析的數據包 
        this.packet = new Packet(); 
        // 跳過開始標記的字節數，進入解析協議頭階段 
        return [PARSE_STATE.PARSE_HEADER, data.slice(Buffer.from([PACKET_START]).byteLength)]; 
    }  
 
    [PARSE_STATE.PARSE_HEADER](data) { 
        // 數據不夠頭部的大小則等待數據到來 
        if (data.length < TOTAL_LENGTH + SEQ_LEN) { 
          return [NEED_MORE_DATA, data]; 
        } 
        // 有效數據包的長度 = 整個數據包長度 - 頭部長度 
        this.packet.set('length', data.readUInt32BE() - (TOTAL_LENGTH + SEQ_LEN)); 
        // 序列號 
        this.packet.set('seq', data.readUInt32BE(TOTAL_LENGTH)); 
        // 解析完頭部了，跳過去 
        data = data.slice(TOTAL_LENGTH + SEQ_LEN); 
        // 進入解析數據階段 
        return [PARSE_STATE.PARSE_DATA, data]; 
    } 
 
    [PARSE_STATE.PARSE_DATA](data) { 
        const len = this.packet.get('length'); 
        // 數據部分的長度小于協議頭中定義的長度，則繼續等待 
        if (data.length < len) { 
            return [NEED_MORE_DATA, data]; 
        } 
        // 截取數據部分 
        this.packet.set('data', data.slice(0, len)); 
        // 解析完數據了，完成一個包的解析，跳過數據部分 
        data = data.slice(len); 
        typeof this.options.cb === 'function' && this.options.cb(this.packet); 
        this.packet = null; 
        // 解析完一個數據包，進入結束標記階段 
        return [PARSE_STATE.PARSE_INIT, data]; 
    } 
}

我們再看一下狀態機的實現

class FSM { 
    constructor(options) { 
        this.options = options; 
        // 狀態處理機，定義了狀態轉移集合 
        this.stateSwitcher = new StateSwitcher({cb: options.cb}); 
        // 當前狀態 
        this.state = PARSE_STATE.PARSE_INIT; 
        // 結束狀態 
        this.endState = PARSE_STATE.PARSE_END; 
        // 當前待解析的數據 
        this.buffer = null; 
    } 
 
    run(data) { 
        // 沒有數據或者解析結束了直接返回 
        if (this.state === this.endState || !data || !data.length) { 
            return; 
        } 
        // 保存待解析的數據 
        this.buffer = this.buffer ? Buffer.concat([this.buffer, data]) : data; 
        // 還沒結束，并且還有數據可以處理則繼續執行 
        while(this.state !== this.endState && this.buffer && this.buffer.length) { 
            // 執行狀態處理函數，返回[下一個狀態, 剩下的數據] 
            const result = this.stateSwitcher[this.state](this.buffer); 
            // 如果下一個狀態是NEED_MORE_DATA則說明需要更多的數據才能繼續解析，并保持當前狀態 
            if (result[0] === NEED_MORE_DATA) { 
                return; 
            } 
            // 記錄下一個狀態和數據 
            [this.state, this.buffer] = result; 
        } 
 
    } 
}

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

2 IPC服務器的設計與實現

首先我們實現一個Client類表示和客戶端通信的實例。

// Client代表一個和server建立連接的客戶端 
class Client extends EventEmitter { 
  constructor(options) { 
    super(); 
    this.options = options; 
  } 
  send(data) { 
    this.options.client.write(data); 
  } 
  end(data) { 
    this.options.client.end(data); 
  } 
}

然后我們開始實現真正的IPC服務器

class Server extends EventEmitter { 
    constructor(options, connectionListener) { 
      super(); 
      this.options = { ...options }; 
      // 根據平臺處理參數 
      if (os.platform() === 'win32') { 
        !~~this.options.port && (this.options.port = port); 
        delete this.options.path; 
      } else { 
        !this.options.path && (this.options.path = path);  
        delete this.options.host; 
        delete this.options.port; 
        fs.existsSync(this.options.path) && fs.unlinkSync(this.options.path); 
        process.on('exit', () => { 
          fs.existsSync(this.options.path) && fs.unlinkSync(this.options.path); 
        }); 
      } 
      this.server = net.createServer({allowHalfOpen: true}, (client) => { 
        const _client = new Client({client}); 
        typeof connectionListener === 'function' && connectionListener(_client); 
        const fsm = new FSM({ 
            cb: function(packet) { 
              _client.emit('message', packet); 
            } 
        }) 
        client.on('data', fsm.run.bind(fsm)); 
        client.on('end', () => { 
          // 觸發end事件 
          _client.emit('end'); 
          // 用戶側沒有關閉寫端，則默認關閉 
          !client.writableEnded && this.options.autoEnd !== false && client.end(); 
        }); 
        client.on('error', (error) => { 
          _client.listenerCount('error') > 0 && _client.emit('error', error); 
        }); 
      }); 
      this.server.listen(this.options, () => { 
        this.emit('listen'); 
      }); 
      this.server.on('error', (error) => { 
        this.listenerCount('error') > 0 && this.emit('error', error); 
      }); 
    } 
}

服務器是對tcp和unix域服務器的封裝，基于tcp或者unix域傳輸的數據由狀態機進行處理，狀態機解析完數據包后，通知調用方。

3 IPC客戶端的設計與實現

class Client extends EventEmitter { 
  constructor(options) { 
    super(); 
    this.options = { ...options }; 
    this.socket = null; 
    this.fsm = new FSM({ 
        cb: (packet) => { 
            this.emit('message', packet); 
        } 
    }) 
  } 
  initOnce() { 
    if (!this.socket) { 
      if (os.platform() === 'win32') { 
        !~~this.options.port && (this.options.port = port); 
        delete this.options.path; 
      } else { 
        !this.options.path && (this.options.path = path);  
        delete this.options.host; 
        delete this.options.port; 
      } 
      this.socket = net.connect({allowHalfOpen: true, ...this.options}); 
      this.socket.on('data', this.fsm.run.bind(this.fsm)); 
      this.socket.on('end', () => { 
        // 觸發end事件 
        this.emit('end'); 
        // 用戶側沒有關閉寫端，則默認關閉 
        !this.socket.writableEnded && this.options.autoEnd !== false && this.socket.end(); 
      }); 
      this.socket.on('error', (e) => { 
        this.listenerCount('error') > 0 && this.emit('error', e); 
      }); 
    } 
  } 
  send(data) { 
    this.initOnce(); 
    this.socket.write(data); 
    return this; 
  } 
  end(data) { 
    this.initOnce(); 
    this.socket.end(data); 
  } 
}

客戶端和服務器類似，也是對tcp和unix域客戶端的封裝。其中數據也是由狀態機處理。

4 使用

接下來我們看一下如何使用。server.js

const { Server, packet, seq } = require('../../'); 
// window下使用tcp，非window使用unix域，即使傳了port  
new Server({port: 80, path: '/tmp/unix.sock'}, function(client) { 
    client.on('message', (data) => { 
        console.log('receive', data); 
        client.send(packet('world', data.seq)); 
        client.send(packet('world', data.seq)); 
    }); 
    client.on('end', (data) => { 
        client.end(); 
    }); 
});

client.js

const { Client, packet, seq } = require('../../'); 
const client = new Client({port: 80, path: '/tmp/unix.sock'}) 
client.send(packet('hello', seq())); 
client.send(packet('hello', seq())); 
client.on('message', function(res) { 
  console.log('receive', res); 
})

服務器輸出

客戶端輸出

5 拓展

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

class RequestManager { 
    constructor(options) { 
        this.options = { timeout: 10000, ...options }; 
        this.map = {}; 
        this.timerId = null; 
        this.startPollTimeout(); 
    } 
    set(key, context) { 
        if (typeof context.cb !== 'function') { 
            throw new Error('cb is required'); 
        } 
        this.map[key] = { 
            startTime: Date.now(), 
            ...context, 
        }; 
    } 
    get(key) { 
        return this.map[key]; 
    } 
    del(key) { 
        return delete this.map[key]; 
    } 
    // 執行上下文 
    exec(key, data) { 
        const context = this.get(key); 
        if (context) { 
            this.del(key); 
            context.cb(data); 
        } 
    }  
    // 定時輪詢是否超時 
    startPollTimeout() { 
        this.timerId = setTimeout(() => { 
            if (!this.timerId) { 
                return; 
            } 
            const nextMap = {}; 
            for (const [key, context] of Object.entries(this.map)) { 
                if (Date.now() - context.startTime < (context.timeout || this.options.timeout)) { 
                    nextMap[key] = context; 
                } else { 
                    context.cb(new Error('timeout')); 
                } 
            } 
            this.map = nextMap; 
            this.startPollTimeout(); 
        }, 1000); 
    } 
}

該類的邏輯主要是請求的seq保存對應的上下文，然后收到響應的時候，我們根據響應的seq拿到對應的上下文，從而執行對應的回調。我們看看如何使用該類。server.js

const { Server, packet, seq } = require('../../'); 
new Server({port: 80, path: '/tmp/unix.sock'}, function(client) { 
    client.on('message', (data) => { 
        console.log('receive', data); 
        // setTimeout測試超時場景 
        //setTimeout(() => { 
            client.send(packet('world', data.seq)); 
        // }, 2000) 
    }); 
    client.on('end', (data) => { 
        client.end(); 
    }); 
});

client.js

const { Client, packet, seq, RequestManager } = require('../../'); 
const requestManager = new RequestManager({timeout: 3000}); 
const client = new Client({port: 80, path: '/tmp/unix.sock'}); 
const _seq = seq();  
// 保存seq對應的上下文 
requestManager.set(_seq, { 
  cb: function() { 
    console.log(...arguments); 
  } 
}) 
// 發送數據包 
client.send(packet('hello', _seq)); 
client.on('message', function(packet) { 
  // 根據響應的seq執行對應的上下文 
  requestManager.exec(packet.seq, packet); 
})

責任編輯：武曉燕來源：編程雜技

Nodejs-Ipc 設計實現

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看

Nodejs-Ipc的設計與實現