node 利用进程通信实现Cluster共享内存

Node.js的标准API没有提供进程共享内存,然而通过IPC接口的send方法和对message事件的监听,就可以实现一个多进程之间的协同机制,通过通信来操作共享内存。

##IPC的基本用法:

// worker进程 发送消息
process.send(‘读取共享内存');

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

在Node.js中,通过send和on(‘message', callback)实现的IPC通信有几个特点。首先,master和worker之间可以互相通信,而各个worker之间不能直接通信,但是worker之间可以通过master转发实现间接通信。另外,通过send方法传递的数据,会先被JSON.stringify处理后再传递,接收后会再用JSON.parse解析。所以Buffer对象传递后会变成数组,而function则无法直接传递。反过来说,就是可以直接传递除了buffer和function之外的所有数据类型(已经很强大了,而且buffer和function也可以用变通的方法实现传递)。

基于以上特点,我们可以设计一个通过IPC来共享内存的方案:

1、worker进程作为共享内存的使用者,并不直接操作共享内存,而是通过send方法通知master进程进行写入(set)或者读取(get)操作。

2、master进程初始化一个Object对象作为共享内存,并根据worker发来的message,对Object的键值进行读写。

3、由于要使用跨进程通信,所以worker发起的set和get都是异步操作,master根据请求进行实际读写操作,然后将结果返回给worker(即把结果数据send给worker)。

##数据格式

为了实现进程间异步的读写功能,需要对通信数据的格式做一点规范。

首先是worker的请求数据:

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

master在接到数据后,会根据method执行相应操作,然后根据requestMessage.id将结果数据发给对应的worker,数据格式如下:

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

规范数据格式的意义在于,master在接收到请求后,能够将处理结果发送给对应的worker,而worker在接到回传的结果后,能够调用此次通信对应的callback,从而实现协同。

规范数据格式后,接下来要做的就是设计两套代码,分别用于master进程和worker进程,监听通信并处理通信数据,实现共享内存的功能。

##User类

User类的实例在worker进程中工作,负责发送操作共享内存的请求,并监听master的回信。

var User = function() {
  var self = this;
  self.__uuid__ = 0;

  // 缓存回调函数
  self.__getCallbacks__ = {};

  // 接收每次操作请求的回信
  process.on('message', function(data) {

    if (!data.isSharedMemoryMessage) return;
    // 通过uuid找到相应的回调函数
    var cb = self.__getCallbacks__[data.uuid];
    if (cb && typeof cb == 'function') {
      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('set', key, value, callback);
};

User.prototype.get = function(key, callback) {
  this.handle('get', key, null, callback);
};

##Manager类

Manager类的实例在master进程中工作,用于初始化一个Object作为共享内存,并根据User实例的请求,在共享内存中增加键值对,或者读取键值,然后将结果发送回去。

var Manager = function() {

  var self = this;

  // 初始化共享内存
  self.__sharedMemory__ = {};

  // 监听并处理来自worker的请求
  cluster.on('online', function(worker) {
    worker.on('message', 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 'OK';
};

// 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('a', [0, 1, 2, 3]);
  }

  if (cluster.worker.id == 2) {
    // 第二个worker从共享内存读取a的值
    sharedMemoryUser.get('a', function(data) {
      console.log(data); // => [0, 1, 2, 3]
    });
  }

}

以上就是一个通过IPC通信实现的多进程共享内存功能,需要注意的是,这种方法是直接在master进程的内存里缓存数据,必须注意内存的使用情况,这里可以考虑加入一些简单的淘汰策略,优化内存的使用。另外,如果单次读写的数据比较大,IPC通信的耗时也会相应增加。

完整代码:https://github.com/x6doooo/sharedmemory

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

时间: 2017-10-26

使用cluster 将自己的Node服务器扩展为多线程服务器

用nodejs的朋友都有了解,node是单线程的,也就是说跑在8核CPU上,只能使用一个核的算力. 单线程一直是node的一个诟病,但随着0.6版本中引入cluster之后,这个情况则得到了改变,开发人员可以依靠cluster很轻松的将自己的Node服务器扩展为多线程服务器了. 什么是Cluster cluster是node提供的一个多线程库,用户可以使用它来创建多个线程,线程之间共享一个监听端口,当有外部请求这个端口时,cluster会将请求转发到随机线程里.因为每个node线程都会占用几十兆

node.js使用cluster实现多进程

首先郑重声明: nodeJS 是一门单线程!异步!非阻塞语言! nodeJS 是一门单线程!异步!非阻塞语言! nodeJS 是一门单线程!异步!非阻塞语言! 重要的事情说3遍. 因为nodeJS天生自带buff, 所以从一出生就受到 万千 粉丝的追捧(俺,也是它的死忠). 但是,傻逼php 竟然嘲笑 我大NodeJS 的性能. 说不稳定,不可靠,只能利用单核CPU. 辣鸡 nodeJS. 艹!艹!艹! 搞mo shi~ 但,大哥就是大哥,nodeJS在v0.8 的时候就已经加入了cluster

Node.js中多进程模块Cluster的介绍与使用

前言 我们都知道nodejs最大的特点就是单进程.无阻塞运行,并且是异步事件驱动的.Nodejs的这些特性能够很好的解决一些问题,例如在服务器开发中,并发的请求处理是个大问题,阻塞式的函数会导致资源浪费和时间延迟.通过事件注册.异步函数,开发人员可以提高资源的利用率,性能也会改善.既然Node.js采用单进程.单线程模式,那么在如今多核硬件流行的环境中,单核性能出色的Nodejs如何利用多核CPU呢?创始人Ryan Dahl建议,运行多个Nodejs进程,利用某些通信机制来协调各项任务.目前,已

Nodejs中解决cluster模块的多进程如何共享数据问题

前述 nodejs在v0.6.x之后增加了一个模块cluster用于实现多进程,利用child_process模块来创建和管理进程,增加程序在多核CPU机器上的性能表现.本文将介绍利用cluster模块创建的多线程如何共享数据的问题. 进程间数据共享 首先举个简单的例子,代码如下: var cluster = require('cluster'); var data = 0;//这里定义数据不会被所有进程共享,各个进程有各自的内存区域 if (cluster.isMaster) { //主进程

node.js中cluster的使用教程

本文主要给大家介绍了关于node.js中cluster使用的相关教程,分享出来供大家参考学习,下面来看看详细的介绍: 一.使用NODE中cluster利用多核CPU var cluster = require('cluster'); var http = require('http'); var numCPUs = require('os').cpus().length; if (cluster.isMaster) { // 创建工作进程 for (var i = 0; i < numCPUs;

Node学习记录之cluster模块

在如今机器的CPU都是多核的背景下,Node的单线程设计已经没法更充分的"压榨"机器性能了.所以从v0.8开始,Node新增了一个内置模块--"cluster",故名思议,它可以通过一个父进程管理一坨子进程的方式来实现集群的功能. var cluster = require('cluster'); var http = require('http'); var numCPUs = require('os').cpus().length; // 获取CPU的个数 if

Node.js中的cluster模块深入解读

预备知识 在如今机器的CPU都是多核的背景下,Node的单线程设计已经没法更充分的"压榨"机器性能了.所以从v0.8开始,Node新增了一个内置模块--"cluster",故名思议,它可以通过一个父进程管理一坨子进程的方式来实现集群的功能. 学习cluster之前,需要了解process相关的知识,如果不了解的话建议先阅读process模块.child_process模块. cluster借助child_process模块的fork()方法来创建子进程,通过fork

node学习记录之搭建web服务器教程

web服务器的基本知识 功能:1.接收HTTP请求(get,post,delete,put)2.处理HTTP请求 常见的web服务器架构: 1. Nginx/Apache:负责接收http请求,确定谁来处理请求,并返回请求的结果 2. php-fpm/php模块 常见得到请求 1.请求文件:包括要处理的静态文件 2.完成特定的操作,如登录,获取特定数据等等 使用http创建web服务器 //引入nodejs的核心模块http var http = require('http') ; //创建一个

详解通过源码解析Node.js中cluster模块的主要功能实现

众所周知,Node.js中的JavaScript代码执行在单线程中,非常脆弱,一旦出现了未捕获的异常,那么整个应用就会崩溃.这在许多场景下,尤其是web应用中,是无法忍受的.通常的解决方案,便是使用Node.js中自带的cluster模块,以master-worker模式启动多个应用实例.然而大家在享受cluster模块带来的福祉的同时,不少人也开始好奇: 为什么我的应用代码中明明有app.listen(port);,但cluter模块在多次fork这份代码时,却没有报端口已被占用? Maste

深入剖析Node.js cluster模块

cluster模块概览 node实例是单线程作业的.在服务端编程中,通常会创建多个node实例来处理客户端的请求,以此提升系统的吞吐率.对这样多个node实例,我们称之为cluster(集群). 借助node的cluster模块,开发者可以在几乎不修改原有项目代码的前提下,获得集群服务带来的好处. 集群有以下两种常见的实现方案,而node自带的cluster模块,采用了方案二. 方案一:多个node实例+多个端口 集群内的node实例,各自监听不同的端口,再由反向代理实现请求到多个端口的分发.

Node.js学习教程之Module模块

前言 采用了 Commonjs 规范,通过 module.exports.require 来导出和导入模块.模块加载机制中,采用了延迟加载的策略.就是说在用到的情况下,系统模块才会被加载,等加载完成后会放到 binding_cache 中. 分类(模块类型) 系统模块 核心模块(native 模块),http.buffer.fs 等,底层调用的内建模块 (C/C++): C/C++ 模块(built-in 内建模块),供 native 模块调用: 第三方模块 第三方维护的模块,比如 expres

Node.js中文件操作模块File System的详细介绍

File System的缩写是fs,该模块提供本地文件的读写能力. Nodejs导入文件系统模块(fs)语法如下所示: var fs = require("fs"); 异步和同步 Node.js文件系统(fs模块)模块中的方法均有异步和同步版本,例如读取文件内容的函数有异步的fs.readFile()和同步的fs.readFileSync() . 异步的方法函数最后一个参数为回调函数,回调函数的第一个参数包含了错误信息(error).同步则没有回调函数. 建议大家是用异步方法,比起同步

MySQL学习记录之KEY分区引发的血案

需求背景 业务表tb_image部分数据如下所示,其中id唯一,image_no不唯一.image_no表示每个文件的编号,每个文件在业务系统中会生成若干个文件,每个文件的唯一ID就是字段id: 业务表tb_image的一些情况如下: 根据image_no查询和根据id查询: 存量数据2kw: 日增长4w左右: 日查询量20w左右: 非ToC系统,所以并发的天花板可见: 方案选择 根据上面对业务的分析,分库分表完全没有必要.单库分表的话,由于要根据image_no和id查询,所以,一种方案是冗余

Node.js基础入门之模块与npm包管理器使用详解

目录 require函数 模块分类 第三方模块 1. 安装第三方模块 2. 引入第三方模块 3. 示例测试 系统模块 require注意事项 exports导出对象 1. exports示例 2. exports注意事项 module模块对象 package.json包描述文件 1. 什么是package.json ? 2. 如何创建package.json文件? NPM基础 1. 常用npm命令 2. npm 示例 cnpm基础 1. 什么是cnpm ? 2. 使用cnpm 控制台输出 1.