Node.js Workerpool CPU 管理策略：单实例池的优势与实践

// route1.js
const workerpool = require('workerpool');
const pool1 = workerpool.pool(__dirname + '/job1.js');
// pool1.exec(...)

// route2.js
const workerpool = require('workerpool');
const pool2 = workerpool.pool(__dirname + '/job2.js');
// pool2.exec(...)

// route3.js
const workerpool = require('workerpool');
const pool3 = workerpool.pool(__dirname + '/job3.js');
// pool3.exec(...)

// worker.js
// 这个文件将在单独的worker线程中运行
module.exports = {
  /**
   * 模拟一个CPU密集型的任务1
   * @param {string} inputData - 输入数据
   * @returns {string} 处理结果
   */
  job1: async (inputData) => {
    console.log(`Worker: Executing job1 with ${inputData}`);
    // 模拟耗时计算
    let result = 0;
    for (let i = 0; i < 1e7; i++) {
      result += Math.sqrt(i);
    }
    return `Job1 completed for: ${inputData}, result: ${result.toFixed(2)}`;
  },

  /**
   * 模拟一个CPU密集型的任务2
   * @param {number} num - 输入数字
   * @returns {number} 斐波那契数列结果
   */
  job2: async (num) => {
    console.log(`Worker: Executing job2 (Fibonacci) for ${num}`);
    // 模拟递归计算斐波那契数列
    const fib = (n) => (n < 2 ? n : fib(n - 1) + fib(n - 2));
    const result = fib(num);
    return `Job2 (Fibonacci) completed for ${num}, result: ${result}`;
  },

  /**
   * 模拟一个CPU密集型的任务3
   * @param {string} text - 输入文本
   * @returns {string} 反转文本
   */
  job3: async (text) => {
    console.log(`Worker: Executing job3 (Reverse Text) for "${text}"`);
    // 模拟字符串处理
    const reversedText = text.split('').reverse().join('');
    return `Job3 (Reverse Text) completed for "${text}", result: "${reversedText}"`;
  },
};

// poolManager.js
const workerpool = require('workerpool');
const path = require('path');

let poolInstance = null;

/**
 * 获取或创建唯一的 Worker Pool 实例。
 * @returns {workerpool.Pool} 唯一的 workerpool 实例。
 */
function getWorkerPool() {
  if (!poolInstance) {
    // 初始化 workerpool，指定包含所有任务的worker文件
    // maxWorkers: 'max' 表示根据CPU核心数创建最大数量的worker线程
    poolInstance = workerpool.pool(path.resolve(__dirname, 'worker.js'), {
      minWorkers: 'max', // 初始创建的worker数量
      maxWorkers: 'max', // 最大允许的worker数量
      workerType: 'thread', // 使用Node.js的worker_threads
      errorHandler: (err) => {
        console.error('Worker Pool Error:', err);
      },
    });

    // 监听workerpool事件，便于调试和监控
    poolInstance.on('workerCreated', (worker) => console.log(`Worker created: ${worker.id}`));
    poolInstance.on('workerDestroyed', (worker) => console.log(`Worker destroyed: ${worker.id}`));
    poolInstance.on('taskCompleted', (task) => console.log(`Task completed: ${task.id}`));
    poolInstance.on('taskFailed', (task, err) => console.error(`Task failed: ${task.id}, Error: ${err.message}`));

    // 在应用关闭时优雅地终止workerpool
    process.on('SIGINT', async () => {
      console.log('Shutting down worker pool...');
      await poolInstance.terminate();
      console.log('Worker pool terminated.');
      process.exit(0);
    });
  }
  return poolInstance;
}

module.exports = { getWorkerPool };

// route1.js (使用Express作为示例)
const express = require('express');
const router = express.Router();
const { getWorkerPool } = require('./poolManager'); // 导入单一的pool实例

router.get('/process-task1', async (req, res) => {
  try {
    const pool = getWorkerPool();
    const data = req.query.data || 'default_input';
    // 调用 worker.js 中暴露的 job1 函数
    const result = await pool.exec('job1', [data]);
    res.status(200).send(`Task 1 processed: ${result}`);
  } catch (error) {
    console.error('Error processing task 1:', error);
    res.status(500).send('Failed to process task 1.');
  }
});

module.exports = router;

// route2.js
const express = require('express');
const router = express.Router();
const { getWorkerPool } = require('./poolManager');

router.get('/process-task2', async (req, res) => {
  try {
    const pool = getWorkerPool();
    const num = parseInt(req.query.num) || 35; // 斐波那契数列的输入
    // 调用 worker.js 中暴露的 job2 函数
    const result = await pool.exec('job2', [num]);
    res.status(200).send(`Task 2 processed: ${result}`);
  } catch (error) {
    console.error('Error processing task 2:', error);
    res.status(500).send('Failed to process task 2.');
  }
});

module.exports = router;

// route3.js
const express = require('express');
const router = express.Router();
const { getWorkerPool } = require('./poolManager');

router.get('/process-task3', async (req, res) => {
  try {
    const pool = getWorkerPool();
    const text = req.query.text || 'hello world';
    // 调用 worker.js 中暴露的 job3 函数
    const result = await pool.exec('job3', [text]);
    res.status(200).send(`Task 3 processed: ${result}`);
  } catch (error) {
    console.error('Error processing task 3:', error);
    res.status(500).send('Failed to process task 3.');
  }
});

module.exports = router;

// app.js
const express = require('express');
const app = express();
const port = 3000;

// 导入路由
const route1 = require('./route1');
const route2 = require('./route2');
const route3 = require('./route3');

// 注册路由
app.use('/api', route1);
app.use('/api', route2);
app.use('/api', route3);

app.get('/', (req, res) => {
  res.send('Workerpool Demo Server is running. Try /api/process-task1, /api/process-task2, /api/process-task3');
});

app.listen(port, () => {
  console.log(`Server listening at http://localhost:${port}`);
});