节点集群 fork 过于频繁地执行代码

Question

我有一些作业（函数）要在单独的线程中执行。因此，我创建了一个实用函数来执行此操作：

import {cpus} from "os"
import cluster from 'cluster'

export default function runInThread(func: Function, useAllCores = false, numThreads = useAllCores ? cpus().length : 1) {
    if(cluster.isMaster) {
        for(let i = 0; i < numThreads; i++) {
            cluster.fork()
        }
    } else {
        func()
    }
}

那么我想像这样执行作业：

import runInThread from "../helpers/threading"
import job1 from "./job1"
import job2 from "./job2"

let jobs = [
    job1, 
    job2
]

jobs.forEach(job => runInThread(job))

示例作业是：

export default function job1() {
    console.log('job1')
}

export default function job2() {
    console.log('job2')
}

不幸的是，它们似乎被执行了两次，因为输出是：

job1
job1
job2
job2

所以似乎每个作业都分叉了两次...？

Answer 1

你描述的完全正常。而且只产生了两个工人。

您可以通过在作业中打印工人 ID 来确认这一点：

import cluster from 'cluster';

export default function job1() {
  console.log(`WORKER #${cluster.worker.id}: job1`);
};

调用 fork 后，它会生成一个新的 node.js 进程（顺便说一句，不是线程）并再次执行您的代码。这意味着您的每个员工都将有效地执行如下操作：

import {cpus} from "os"
import cluster from 'cluster'
import job1 from "./job1"
import job2 from "./job2"

function runInThread(func, useAllCores = false, numThreads = useAllCores ? cpus().length : 1) {
    func()
}

let jobs = [
    job1,
    job2
]

jobs.forEach(job => runInThread(job))

因此每个工人将记录：

job1
job2

然后退出。此外，由于它们在两个不同的进程中执行，因此日志消息可能会交错。

Answer 2

主要思想 - 每个线程（fork）都像特定功能的宿主。在示例中，这只是函数名称，但在实际工作中，您可以为函数生成一些 id，并将其作为环境变量传递给 worker。

import {cpus} from "os"
import cluster from 'cluster'

export default function runInThread(func: Function, useAllCores = false, numThreads = useAllCores ? cpus().length : 1) {
    if(cluster.isMaster) {
        for(let i = 0; i < numThreads; i++) {
            cluster.fork({FUNC_ID: func.name})
        }
    } else {
       if (process.env.FUNC_ID === func.name) {
         func()
       }
    }
}

节点集群 fork 过于频繁地执行代码

node cluster fork executes code too often

worker

node.js