聊聊Node.js中的多进程和多线程

来自:互联网
时间:2022-07-26
阅读:

我们都知道 Node.js 采用的是单线程、基于事件驱动的异步 I/O 模型,其特性决定了它无法利用 CPU 多核的优势,也不善于完成一些非 I/O 类型的操作(比如执行脚本、AI 计算、图像处理等),为了解决此类问题,Node.js 提供了常规的多进(线程)方案(关于进程、线程的讨论,可参见笔者的另一篇文章 Node.js 与并发模型),本文便为大家介绍 Node.js 的多进(线)程机制。

child_process

我们可使用 child_process 模块创建 Node.js 的子进程,来完成一些特殊的任务(比如执行脚本),该模块主要提供了 execexecFileforkspwan 等方法,下面我们就简单介绍下这些方法的使用。

exec

const { exec } = require('child_process');

exec('ls -al', (error, stdout, stderr) => {
  console.log(stdout);
});

该方法根据 options.shell 指定的可执行文件处理命令字符串,在命令的执行过程中缓存其输出,直到命令执行完成后,再将执行结果以回调函数参数的形式返回。

该方法的参数解释如下:

  • command:将要执行的命令(比如 ls -al);

  • options:参数设置(可不指定),相关属性如下:

    • cwd:子进程的当前工作目录,默认取 process.cwd() 的值;

    • env:环境变量设置(为键值对对象),默认取 process.env 的值;

    • encoding:字符编码,默认值为:utf8

    • shell:处理命令字符串的可执行文件,Unix 上默认值为 /bin/shWindows 上默认值取 process.env.ComSpec 的值(如为空则为 cmd.exe);比如:

      const { exec } = require('child_process');
      
      exec("print('Hello World!')", { shell: 'python' }, (error, stdout, stderr) => {
        console.log(stdout);
      });

      运行上面的例子将输出 Hello World!,这等同于子进程执行了 python -c "print('Hello World!')" 命令,因此在使用该属性时需要注意,所指定的可执行文件必须支持通过 -c 选项来执行相关语句。

      注:碰巧 Node.js 也支持 -c 选项,但它等同于 --check 选项,只用来检测指定的脚本是否存在语法错误,并不会执行相关脚本。

    • signal:使用指定的 AbortSignal 终止子进程,该属性在 v14.17.0 以上可用,比如:

      const { exec } = require('child_process');
      
      const ac = new AbortController();
      exec('ls -al', { signal: ac.signal }, (error, stdout, stderr) => {});

      上例中,我们可通过调用 ac.abort() 来提前终止子进程。

    • timeout:子进程的超时时间(如果该属性的值大于 0,那么当子进程运行时间超过指定值时,将会给子进程发送属性 killSignal 指定的终止信号),单位毫米,默认值为 0

    • maxBuffer:stdout 或 stderr 所允许的最大缓存(二进制),如果超出,子进程将会被杀死,并且将会截断任何输出,默认值为 1024 * 1024

    • killSignal:子进程终止信号,默认值为 SIGTERM

    • uid:执行子进程的 uid

    • gid:执行子进程的 gid

    • windowsHide:是否隐藏子进程的控制台窗口,常用于 Windows 系统,默认值为 false

  • callback:回调函数,包含 errorstdoutstderr 三个参数:

    • error:如果命令行执行成功,值为 null,否则值为 Error 的一个实例,其中 error.code 为子进程的退出的错误码,error.signal 为子进程终止的信号;
    • stdout 和 stderr:子进程的 stdout 和 stderr,按照 encoding 属性的值进行编码,如果 encoding 的值为 buffer,或者 stdoutstderr 的值是一个无法识别的字符串,将按照 buffer 进行编码。

execFile

const { execFile } = require('child_process');

execFile('ls', ['-al'], (error, stdout, stderr) => {
  console.log(stdout);
});

该方法的功能类似于 exec,唯一的区别是 execFile 在默认情况下直接用指定的可执行文件(即参数 file 的值)处理命令,这使得其效率略高于 exec(如果查看 shell 的处理逻辑,笔者感觉这效率可忽略不计)。

该方法的参数解释如下:

  • file:可执行文件的名字或路径;

  • args:可执行文件的参数列表;

  • options:参数设置(可不指定),相关属性如下:

    • shell:值为 false 时表示直接用指定的可执行文件(即参数 file 的值)处理命令,值为 true 或其它字符串时,作用等同于 exec 中的 shell,默认值为 false
    • windowsVerbatimArguments:在 Windows 中是否对参数进行引号或转义处理,在 Unix 中将忽略该属性,默认值为 false
    • 属性 cwdenvencodingtimeoutmaxBufferkillSignaluidgidwindowsHidesignal 在上文中已介绍,此处不再重述。
  • callback:回调函数,等同于 exec 中的 callback,此处不再阐述。

fork

const { fork } = require('child_process');

const echo = fork('./echo.js', {
  silent: true
});
echo.stdout.on('data', (data) => {
  console.log(`stdout: ${data}`);
});

echo.stderr.on('data', (data) => {
  console.error(`stderr: ${data}`);
});

echo.on('close', (code) => {
  console.log(`child process exited with code ${code}`);
});

该方法用于创建新的 Node.js 实例以执行指定的 Node.js 脚本,与父进程之间以 IPC 方式进行通信。

该方法的参数解释如下:

  • modulePath:要运行的 Node.js 脚本路径;

  • args:传递给 Node.js 脚本的参数列表;

  • options:参数设置(可不指定),相关属性如:

    • detached:参见下文对 spwan 中 options.detached 的说明;

    • execPath:创建子进程的可执行文件;

    • execArgv:传递给可执行文件的字符串参数列表,默认取 process.execArgv 的值;

    • serialization:进程间消息的序列号类型,可用值为 json 和 advanced,默认值为 json

    • slient: 如果为 true,子进程的 stdinstdout 和 stderr 将通过管道传递给父进程,否则将继承父进程的 stdinstdout 和 stderr;默认值为 false

    • stdio:参见下文对 spwan 中 options.stdio 的说明。这里需要注意的是:

      • 如果指定了该属性,将忽略 slient 的值;
      • 必须包含一个值为 ipc 的选项(比如 [0, 1, 2, 'ipc']),否则将抛出异常。
    • 属性 cwdenvuidgidwindowsVerbatimArgumentssignaltimeoutkillSignal 在上文中已介绍,此处不再重述。

spwan

const { spawn } = require('child_process');

const ls = spawn('ls', ['-al']);
ls.stdout.on('data', (data) => {
  console.log(`stdout: ${data}`);
});

ls.stderr.on('data', (data) => {
  console.error(`stderr: ${data}`);
});

ls.on('close', (code) => {
  console.log(`child process exited with code ${code}`);
});

该方法为 child_process 模块的基础方法,execexecFilefork 最终都会调用 spawn 来创建子进程。

该方法的参数解释如下:

  • command:可执行文件的名字或路径;

  • args:传递给可执行文件的参数列表;

  • options:参数设置(可不指定),相关属性如下:

    • argv0:发送给子进程 argv[0] 的值,默认取参数 command 的值;

    • detached:是否允许子进程可以独立于父进程运行(即父进程退出后,子进程可以继续运行),默认值为 false,其值为 true 时,各平台的效果如下所述:

      • 在 Windows 系统中,父进程退出后,子进程可以继续运行,并且子进程拥有自己的控制台窗口(该特性一旦启动后,在运行过程中将无法更改);
      • 在非 Windows 系统中,子进程将作为新进程会话组的组长,此刻不管子进程是否与父进程分离,子进程都可以在父进程退出后继续运行。

      需要注意的是,如果子进程需要执行长时间的任务,并且想要父进程提前退出,需要同时满足以下几点:

      • 调用子进程的 unref 方法从而将子进程从父进程的事件循环中剔除;
      • detached 设置为 true
      • stdio 为 ignore

      比如下面的例子:

      // hello.js
      const fs = require('fs');
      let index = 0;
      function run() {
        setTimeout(() => {
          fs.writeFileSync('./hello', `index: ${index}`);
          if (index < 10) {
            index += 1;
            run();
          }
        }, 1000);
      }
      run();
      
      // main.js
      const { spawn } = require('child_process');
      const child = spawn('node', ['./hello.js'], {
        detached: true,
        stdio: 'ignore'
      });
      child.unref();
    • stdio:子进程标准输入输出配置,默认值为 pipe,值为字符串或数组:

      • 值为字符串时,会将其转换为含有三个项的数组(比如 pipe 被转换为 ['pipe', 'pipe', 'pipe']),可用值为 pipeoverlappedignoreinherit
      • 值为数组时,其中数组的前三项分别代表对 stdinstdout 和 stderr 的配置,每一项的可用值为 pipeoverlappedignoreinheritipcStream 对象、正整数(在父进程打开的文件描述符)、null(如位于数组的前三项,等同于 pipe,否则等同于 ignore)、undefined(如位于数组的前三项,等同于 pipe,否则等同于 ignore)。
    • 属性 cwdenvuidgidserializationshell(值为 boolean 或 string)、windowsVerbatimArgumentswindowsHidesignaltimeoutkillSignal 在上文中已介绍,此处不再重述。

小结

上文对 child_process 模块中主要方法的使用进行了简短介绍,由于 execSyncexecFileSyncforkSyncspwanSync 方法是 execexecFilespwan 的同步版本,其参数并无任何差异,故不再重述。

cluster

通过 cluster 模块我们可以创建 Node.js 进程集群,通过 Node.js 进程进群,我们可以更加充分地利用多核的优势,将程序任务分发到不同的进程中以提高程序的执行效率;下面将通过例子为大家介绍 cluster 模块的使用:

const http = require('http');
const cluster = require('cluster');
const numCPUs = require('os').cpus().length;

if (cluster.isPrimary) {
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }
} else {
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end(`${process.pid}\n`);
  }).listen(8000);
}

上例通过 cluster.isPrimary 属性判断(即判断当前进程是否为主进程)将其分为两个部分:

  • 为真时,根据 CPU 内核的数量并通过 cluster.fork 调用来创建相应数量的子进程;
  • 为假时,创建一个 HTTP server,并且每个 HTTP server 都监听同一个端口(此处为 8000)。

运行上面的例子,并在浏览器中访问 http://localhost:8000/,我们会发现每次访问返回的 pid 都不一样,这说明了请求确实被分发到了各个子进程。Node.js 默认采用的负载均衡策略是轮询调度,可通过环境变量 NODE_CLUSTER_SCHED_POLICY 或 cluster.schedulingPolicy 属性来修改其负载均衡策略:

NODE_CLUSTER_SCHED_POLICY = rr // 或 none

cluster.schedulingPolicy = cluster.SCHED_RR; // 或 cluster.SCHED_NONE

另外需要注意的是,虽然每个子进程都创建了 HTTP server,并都监听了同一个端口,但并不代表由这些子进程自由竞争用户请求,因为这样无法保证所有子进程的负载达到均衡。所以正确的流程应该是由主进程监听端口,然后将用户请求根据分发策略转发到具体的子进程进行处理。

由于进程之间是相互隔离的,因此进程之间一般通过共享内存消息传递管道等机制进行通讯。Node.js 则是通过消息传递来完成父子进程之间的通信,比如下面的例子:

const http = require('http');
const cluster = require('cluster');
const numCPUs = require('os').cpus().length;

if (cluster.isPrimary) {
  for (let i = 0; i < numCPUs; i++) {
    const worker = cluster.fork();
    worker.on('message', (message) => {
      console.log(`I am primary(${process.pid}), I got message from worker: "${message}"`);
      worker.send(`Send message to worker`)
    });
  }
} else {
  process.on('message', (message) => {
    console.log(`I am worker(${process.pid}), I got message from primary: "${message}"`)
  });
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end(`${process.pid}\n`);
    process.send('Send message to primary');
  }).listen(8000);
}

运行上面的例子,并访问 http://localhost:8000/,再查看终端,我们会看到类似下面的输出:

I am primary(44460), I got message from worker: "Send message to primary"
I am worker(44461), I got message from primary: "Send message to worker"
I am primary(44460), I got message from worker: "Send message to primary"
I am worker(44462), I got message from primary: "Send message to worker"

利用该机制,我们可以监听各子进程的状态,以便在某个子进程出现意外后,能够及时对其进行干预,以保证服务的可用性。

cluster 模块的接口非常简单,为了节省篇幅,这里只对 cluster.setupPrimary 方法做一些特别声明,其它方法请查看官方文档

  • cluster.setupPrimary 调用后,相关设置将同步到在 cluster.settings 属性中,并且每次调用都基于当前 cluster.settings 属性的值;
  • cluster.setupPrimary 调用后,对已运行的子进程没有影响,只影响后续的 cluster.fork 调用;
  • cluster.setupPrimary 调用后,不影响后续传递给 cluster.fork 调用的 env 参数;
  • cluster.setupPrimary 只能在主进程中使用。

worker_threads

前文我们对 cluster 模块进行了介绍,通过它我们可以创建 Node.js 进程集群以提高程序的运行效率,但 cluster 基于多进程模型,进程间高成本的切换以及进程间资源的隔离,会随着子进程数量的增加,很容易导致因系统资源紧张而无法响应的问题。为解决此类问题,Node.js 提供了 worker_threads,下面我们通过具体的例子对该模块的使用进行简单介绍:

// server.js
const http = require('http');
const { Worker } = require('worker_threads');

http.createServer((req, res) => {
  const httpWorker = new Worker('./http_worker.js');
  httpWorker.on('message', (result) => {
    res.writeHead(200);
    res.end(`${result}\n`);
  });
  httpWorker.postMessage('Tom');
}).listen(8000);

// http_worker.js
const { parentPort } = require('worker_threads');

parentPort.on('message', (name) => {
  parentPort.postMessage(`Welcone ${name}!`);
});

上例展示了 worker_threads 的简单使用,在使用 worker_threads 的过程中,需要注意以下几点:

  • 通过 worker_threads.Worker 创建 Worker 实例,其中 Worker 脚本既可以为一个独立的 JavaScript 文件,也可以为字符串,比如上例可修改为:

    const code = "const { parentPort } = require('worker_threads'); parentPort.on('message', (name) => {parentPort.postMessage(`Welcone ${name}!`);})";
    const httpWorker = new Worker(code, { eval: true });
  • 通过 worker_threads.Worker 创建 Worker 实例时,可以通过指定 workerData 的值来设置 Worker 子线程的初始元数据,比如:

    // server.js
    const { Worker } = require('worker_threads');
    const httpWorker = new Worker('./http_worker.js', { workerData: { name: 'Tom'} });
    
    // http_worker.js
    const { workerData } = require('worker_threads');
    console.log(workerData);
  • 通过 worker_threads.Worker 创建 Worker 实例时,可通过设置 SHARE_ENV 以实现在 Worker 子线程与主线程之间共享环境变量的需求,比如:

    const { Worker, SHARE_ENV } = require('worker_threads');
    const worker = new Worker('process.env.SET_IN_WORKER = "foo"', { eval: true, env: SHARE_ENV });
    worker.on('exit', () => {
      console.log(process.env.SET_IN_WORKER);
    });
  • 不同于 cluster 中进程间的通信机制,worker_threads 采用的 MessageChannel 来进行线程间的通信:

    • Worker 子线程通过 parentPort.postMessage 方法发送消息给主线程,并通过监听 parentPort 的 message 事件来处理来自主线程的消息;
    • 主线程通过 Worker 子线程实例(此处为 httpWorker,以下均以此代替 Worker 子线程)的 postMessage 方法发送消息给 httpWorker,并通过监听 httpWorker 的 message 事件来处理来自 Worker 子线程的消息。

在 Node.js 中,无论是 cluster 创建的子进程,还是 worker_threads 创建的 Worker 子线程,它们都拥有属于自己的 V8 实例以及事件循环,所不同的是:

  • 子进程之间的内存空间是互相隔离的,而 Worker 子线程共享所属进程的内存空间;
  • 子进程之间的切换成本要远远高于 Worker 子线程之间的切换成本。

尽管看起来 Worker 子线程比子进程更高效,但 Worker 子线程也有不足的地方,即cluster 提供了负载均衡,而 worker_threads 则需要我们自行完成负载均衡的设计与实现。

总结

本文介绍了 Node.js 中 child_processcluster 和 worker_threads 三个模块的使用,通过这三个模块,我们可以充分利用 CPU 多核的优势,并以多进(线)程的模式来高效地解决一些特殊任务(比如 AI、图片处理等)的运行效率。每个模块都有其适用的场景,文中仅对其基本使用进行了说明,如何结合自己的问题进行高效地运用,还需要大家自行摸索。最后,本文若有纰漏之处,还望大家能够指正,祝大家快乐编码每一天。

返回顶部
顶部