クラスタ共有メモリのノード実装の詳細説明-jsチュートリアル-php.cn

この記事では、主にプロセス通信を使用してクラスター共有メモリを実装するノードを紹介します。編集者が非常に優れていると考えたので、参考として共有します。編集者をフォローして見てみましょう。皆さんのお役に立てれば幸いです。

Node.jsの標準APIではプロセス共有メモリを提供していませんが、IPCインターフェースのsendメソッドやメッセージイベントの監視により、複数プロセス間で通信を介して共有メモリを操作する連携機構を実装できます。

##IPC の基本的な使用法:

// worker进程 发送消息
process.send(‘读取共享内存&#39;);
 
// master进程 接收消息 -> 处理 -> 发送回信
cluster.on(&#39;online&#39;, function (worker) {
   // 有worker进程建立，即开始监听message事件
   worker.on(‘message&#39;, function(data) {
     // 处理来自worker的请求
     // 回传结果
     worker.send(‘result&#39;)
   });
});

ログイン後にコピー

Node.js では、send および on('message', callback) によって実装される IPC 通信にはいくつかの特徴があります。まず、マスターとワーカーは相互に通信できますが、ワーカー同士は直接通信できませんが、ワーカーはマスター転送を通じて間接的に通信できます。また、send メソッドで渡されたデータは、渡される前に JSON.stringify によって処理され、受信後、JSON.parse を使用して解析されます。したがって、Buffer オブジェクトは渡された後は配列になりますが、関数を直接渡すことはできません。一方、バッファと関数を除くすべてのデータ型は直接転送できます (これはすでに非常に強力であり、バッファと関数は別の方法を使用して転送することもできます)。

上記の特性に基づいて、IPC を介してメモリを共有するソリューションを設計できます。

1. 共有メモリのユーザーとして、ワーカープロセスは共有メモリを直接操作しませんが、マスタープロセスに書き込みを行うように通知します。 send メソッド (set) または read (get) 操作。

2. マスタープロセスは、オブジェクトオブジェクトを共有メモリとして初期化し、ワーカーから送信されたメッセージに従ってオブジェクトのキー値を読み書きします。

3. クロスプロセス通信が使用されるため、ワーカーによって開始される set および get は非同期操作であり、マスターはリクエストに従って実際の読み取りおよび書き込み操作を実行し、結果をワーカーに返します。結果データは作業者に送信されます）。

##データフォーマット

プロセス間の非同期読み書き機能を実現するには、通信データのフォーマットを統一する必要があります。

1 つ目はワーカーのリクエストデータです:

requestMessage = {
  isSharedMemoryMessage: true, // 表示这是一次共享内存的操作通信
  method: ‘set&#39;, // or ‘get&#39; 操作的方法
  id: cluster.worker.id, // 发起操作的进程（在一些特殊场景下，用于保证master可以回信）
  uuid: uuid, // 此次操作的（用于注册/调用回调函数）
  key: key, // 要操作的键
  value: value // 键对应的值（写入）
}

ログイン後にコピー

データを受信した後、マスターはメソッドに従って対応する操作を実行し、requestMessage.id に従って結果データを対応するワーカーに送信します。データ形式は次のとおりです。

responseMessage = {
  isSharedMemoryMessage: true, // 标记这是一次共享内存通信
  uuid: requestMessage.uuid, // 此次操作的唯一标示
  value: value // 返回值。get操作为key对应的值，set操作为成功或失败
}

ログイン後にコピー

データ形式を標準化する意義は、マスターがリクエストを受信した後、処理結果を対応するワーカーに送信し、ワーカーが返された結果を受け取った後に、処理結果を対応するワーカーに送信できることです。この通信に対応するコールバックを呼び出すことで連携を実現します。

データ形式を標準化したら、次に行うことは、共有メモリの機能を実現するために、通信を監視し、通信データを処理するためのコードをマスタープロセスとワーカープロセスのそれぞれに 2 セット設計することです。

##User クラス

User クラスのインスタンスはワーカープロセスで動作し、共有メモリを操作するためのリクエストを送信し、マスターからの応答をリッスンする責任があります。

var User = function() {
  var self = this;
  self.__uuid__ = 0;
 
  // 缓存回调函数
  self.__getCallbacks__ = {};
 
  // 接收每次操作请求的回信
  process.on(&#39;message&#39;, function(data) {
    
    if (!data.isSharedMemoryMessage) return;
    // 通过uuid找到相应的回调函数
    var cb = self.__getCallbacks__[data.uuid];
    if (cb && typeof cb == &#39;function&#39;) {
      cb(data.value)
    }
    // 卸载回调函数
    self.__getCallbacks__[data.uuid] = undefined;
  });
};
 
// 处理操作
User.prototype.handle = function(method, key, value, callback) {
 
  var self = this;
  var uuid = self.__uuid__++;
 
  process.send({
    isSharedMemoryMessage: true,
    method: method,
    id: cluster.worker.id,
    uuid: uuid,
    key: key,
    value: value
  });
 
  // 注册回调函数
  self.__getCallbacks__[uuid] = callback;
 
};
 
User.prototype.set = function(key, value, callback) {
  this.handle(&#39;set&#39;, key, value, callback);
};
 
User.prototype.get = function(key, callback) {
  this.handle(&#39;get&#39;, key, null, callback);
};

ログイン後にコピー

##Manager クラス

Manager クラスのインスタンスはマスタープロセスで動作し、オブジェクトを共有メモリとして初期化し、オブジェクトのリクエストに従って共有メモリにキーと値のペアを追加するために使用されます。ユーザーインスタンス、またはキー値を読み取り、結果を送り返します。

var Manager = function() {
 
  var self = this;
  
  // 初始化共享内存
  self.__sharedMemory__ = {};
    
  // 监听并处理来自worker的请求
  cluster.on(&#39;online&#39;, function(worker) {
    worker.on(&#39;message&#39;, function(data) {
      // isSharedMemoryMessage是操作共享内存的通信标记
      if (!data.isSharedMemoryMessage) return;
      self.handle(data);
    });
  });
};
 
Manager.prototype.handle = function(data) {
  var self = this;
  var value = this[data.method](data);
 
  var msg = {
    // 标记这是一次共享内存通信
    isSharedMemoryMessage: true,       
    // 此次操作的唯一标示
    uuid: data.uuid,
    // 返回值
    value: value
  };
 
  cluster.workers[data.id].send(msg);
};
 
// set操作返回ok表示成功
Manager.prototype.set = function(data) {
  this.__sharedMemory__[data.key] = data.value;
  return &#39;OK&#39;;
};
 
// get操作返回key对应的值
Manager.prototype.get = function(data) {
  return this.__sharedMemory__[data.key];
};

ログイン後にコピー

##使い方

if (cluster.isMaster) {
 
  // 初始化Manager的实例
  var sharedMemoryManager = new Manager();
 
  // fork第一个worker
  cluster.fork();
 
  // 1秒后fork第二个worker
  setTimeout(function() {
    cluster.fork();
  }, 1000);
   
} else {
 
  // 初始化User类的实例
  var sharedMemoryUser = new User();
 
  if (cluster.worker.id == 1) {
    // 第一个worker向共享内存写入一组数据，用a标记
    sharedMemoryUser.set(&#39;a&#39;, [0, 1, 2, 3]);
  }
 
  if (cluster.worker.id == 2) {
    // 第二个worker从共享内存读取a的值
    sharedMemoryUser.get(&#39;a&#39;, function(data) {
      console.log(data); // => [0, 1, 2, 3]
    });
  }
  
}

ログイン後にコピー

上記はIPC通信により実現されるマルチプロセス共有メモリ機能です。なお、このメソッドはデータをマスタのメモリに直接キャッシュします。ここでは、メモリの使用を最適化するためにいくつかの単純な排除戦略を追加することを検討できます。また、一度に読み書きするデータが比較的大きい場合、IPC通信にかかる時間もそれに応じて増加します。

完全なコード: https://github.com/x6doooo/sharedmemory