Nodejs 构建Cluster集群多线程Worker threads

下面是详细的攻略，希望对您有帮助。

Node.js 构建 Cluster 集群

Cluster 是 Node.js 自带的库，可以简单的创建子进程。它可以实现 Node.js 应用程序的多进程负载平衡，提高应用程序的性能和可用性。

下面是使用 Cluster 模块创建 Node.js 应用程序的集群：

首先，需要判断当前环境是否为主进程。可以使用以下代码判断：

const cluster = require('cluster');
const os = require('os');

if (cluster.isMaster) {
    // 是主进程，执行主进程的代码
} else {
    // 是工作进程，执行工作进程的代码
}

在主进程中，需要通过 fork 方法创建多个工作进程。如下所示：

const numCPUs = os.cpus().length; // 获取系统的 CPU 核心数

for (let i = 0; i < numCPUs; i++) {
    cluster.fork(); // 创建工作进程
}

在工作进程中，需要运行应用程序的代码。例如，使用 Express 框架创建 Web 服务器：

const express = require('express');
const app = express();

app.get('/', (req, res) => {
    res.send('Hello World');
});

app.listen(3000, () => {
    console.log('Server is listening on port 3000');
});

然后，可以使用 cluster 模块的 worker 事件监听工作进程的消息：

for (const id in cluster.workers) {
    cluster.workers[id].on('message', (message) => {
        console.log(`Worker ${id} sent message: ${message}`);
    });
}

可以在工作进程中发送消息，如下所示：

process.send('Hello, I am worker');

Node.js 构建多线程 Worker Threads

Worker Threads 能够创建与主线程不同的多个线程，将负载均衡的机制运用到了 CPU 内部，使得 CPU 的多个核心可以同时工作。使用 Worker Threads 可以有效地提高 Node.js 应用程序的性能。以下是 Worker Threads 的使用方法：

首先，需要引入 Worker Threads 模块：

const { Worker } = require('worker_threads');

可以使用 new Worker 方法创建 Worker 线程。例如，在主线程中创建 Worker 线程：

const worker = new Worker('./worker.js');

worker.on('message', (message) => {
    console.log(`Received message from worker: ${message}`);
});

worker.postMessage('Hello, worker');

new Worker 方法接受一个 JavaScript 文件名作为参数。在上面的示例中，./worker.js 文件包含了 Worker 线程运行的 JavaScript 代码。

在 Worker 线程中，可以使用 parentPort 对象发送和接收消息：

const { parentPort } = require('worker_threads');

parentPort.on('message', (message) => {
    console.log(`Received message from main thread: ${message}`);
});

parentPort.postMessage('Hello, main thread');

可以看到，在 Worker 线程中使用 parentPort 对象发送消息后，在主线程中通过 worker.on('message', (message) => {...}) 捕获到消息并进行处理。

以下是一个示例，展示如何使用 Worker Threads 并发处理数组元素：

const { Worker, isMainThread } = require('worker_threads');

// 处理数组的函数
function processArray(array) {
    return new Promise((resolve, reject) => {
        if (isMainThread) {
            const totalThreads = 4; // 线程数
            const arrayLength = array.length;
            const results = new Array(arrayLength);

            // 创建 Worker 线程
            const createWorker = (index) => {
                return new Promise((resolve, reject) => {
                    const start = index * Math.floor(arrayLength / totalThreads);
                    const end = (index == totalThreads - 1) ? arrayLength : (index + 1) * Math.floor(arrayLength / totalThreads);

                    const worker = new Worker(__filename);
                    worker.postMessage({ start, end, array });
                    worker.on('message', (result) => {
                        for (let i = start; i < end; i++) {
                            results[i] = result[i - start];
                        }
                        resolve();
                    });
                    worker.on('error', reject);
                    worker.on('exit', (code) => {
                        if (code != 0) {
                            reject(new Error(`Worker stopped with exit code ${code}`));
                        }
                    })
                });
            }

            // 创建 Worker 线程池
            const workers = new Array(totalThreads);
            const pool = [...workers.keys()].map((i) => createWorker(i));

            Promise.all(pool).then(() => {
                resolve(results);
            }).catch(reject);
        } else {
            // 在 Worker 线程中处理数组
            const { start, end, array } = require('worker_threads').workerData;
            const result = new Array(end - start);

            for (let i = start; i < end; i++) {
                result[i - start] = array[i] * array[i];
            }

            require('worker_threads').parentPort.postMessage(result);
        }
    });
}

// 测试代码
(async () => {
    const a = new Array(1000000).fill(0).map(() => Math.random() * 1000); // 随机生成一个数组
    const results = await processArray(a); // 处理数组

    console.log(results.slice(0, 10)); // 输出前 10 个元素
})();

在该示例中，创建了一个 processArray 函数，用于同时处理数组的多个元素。函数会为数组分配不同的区间，并创建多个 Worker 线程来处理数组中对应区间的元素。然后，Worker 线程会使用平方函数处理数组的区间，并使用 parentPort.postMessage 方法将结果返回给主线程。最后，Promise.all 方法等待所有 Worker 线程处理完毕，并返回处理后的结果。

另外，记得在使用 Worker Threads 时要注意共享内存和线程安全问题，否则会导致程序出错。可以使用 SharedArrayBuffer 安全的共享内存。同时，为了保证程序的正确性，还要避免竞争和死锁等问题。