如何提升Node.js日志统计函数的执行效率？

核心问题在于Node.js事件循环是单线程的：尽管fs.createReadStream这类异步IO操作由libuv线程池处理，但日志分析的核心逻辑（// do something...）是同步CPU密集型任务，会阻塞主线程。当多个Promise同时执行时，这些同步任务会在主线程里排队串行执行，导致池大小设为10和1的耗时几乎一致——本质上并没有实现真正的并发。

针对不同的任务类型，有两种有效的优化方向：

1. 使用Worker Threads实现真正并行（适用于CPU密集型分析） #

Worker Threads是Node.js专门用于处理CPU密集型任务的模块，每个Worker拥有独立的V8线程，能真正并行执行分析逻辑，彻底解决单线程阻塞问题。

步骤1：编写Worker分析脚本（log-analyzer-worker.js） #

const { parentPort, workerData } = require('worker_threads');
const fs = require('fs');
const readline = require('readline');

async function analyzeLog(filePath) {
  const fileStream = fs.createReadStream(filePath);
  const readLine = readline.createInterface({
    input: fileStream,
    crlfDelay: Infinity,
  });

  // 替换为你的实际日志分析逻辑
  const stats = { errorCount: 0 };
  for await (const line of readLine) {
    if (line.includes('ERROR')) {
      stats.errorCount++;
    }
  }

  fileStream.destroy();
  return stats;
}

// 接收主线程的文件路径，执行分析后返回结果
analyzeLog(workerData.filePath)
  .then(result => parentPort.postMessage({ filePath: workerData.filePath, result }))
  .catch(err => parentPort.postMessage({ filePath: workerData.filePath, error: err.message }));

步骤2：主线程实现Worker Pool #

const { Worker } = require('worker_threads');
const filePathList = [
  "./log1.log", "./log2.log", "./log3.log", "./log4.log", "./log5.log",
  "./log6.log", "./log7.log", "./log8.log", "./log9.log", "./log10.log"
];
const poolSize = 10;

async function statistic() {
  const activeWorkers = new Set();
  const results = [];

  while (filePathList.length > 0) {
    // 填满Worker池
    while (activeWorkers.size < poolSize && filePathList.length > 0) {
      const filePath = filePathList.shift();
      const worker = new Worker('./log-analyzer-worker.js', { workerData: { filePath } });
      
      activeWorkers.add(worker);

      // 监听Worker完成事件
      worker.on('message', (msg) => {
        activeWorkers.delete(worker);
        results.push(msg);
        console.log(`${filePath} DONE!, remain workers: ${activeWorkers.size}`);
      });

      worker.on('error', (err) => {
        activeWorkers.delete(worker);
        console.error(`${filePath} FAILED: ${err.message}`);
      });

      worker.on('exit', (code) => {
        if (code !== 0) {
          console.error(`${filePath} exited with code ${code}`);
        }
      });
    }

    // 等待至少一个Worker完成，再补充新任务
    await new Promise(resolve => {
      const listener = () => {
        if (activeWorkers.size < poolSize) {
          resolve();
          activeWorkers.forEach(w => w.off('message', listener));
        }
      };
      activeWorkers.forEach(w => w.on('message', listener));
    });
  }

  // 等待所有剩余Worker完成
  await Promise.all(Array.from(activeWorkers).map(w => 
    new Promise(resolve => w.on('exit', resolve))
  ));

  return results;
}

(async () => {
  const start = Date.now();
  console.log(`Start statistic algorithm, timestamp: ${start}`);
  const results = await statistic();
  const end = Date.now();
  console.log(`End statistic algorithm, timestamp: ${end}`);
  console.log(`It took ${(end - start) / 1000} sec`);
  console.log('Analysis results:', results);
})();

2. 优化Promise Pool（仅适用于IO密集型分析） #

如果日志分析逻辑是IO密集型（比如需要调用外部API查询），可以优化原Promise Pool的实现，确保异步任务正确调度，避免主线程阻塞：

优化后的Promise Pool代码 #

const fs = require("fs");
const readline = require("readline");

const filePathList = [
  "./log1.log", "./log2.log", "./log3.log", "./log4.log", "./log5.log",
  "./log6.log", "./log7.log", "./log8.log", "./log9.log", "./log10.log"
];

const poolSize = 10;

async function processFile(filePath) {
  const fileStream = fs.createReadStream(filePath);
  const readLine = readline.createInterface({
    input: fileStream,
    crlfDelay: Infinity,
  });

  const stats = { errorCount: 0 };
  for await (const line of readLine) {
    // 若存在轻量CPU操作，每处理N行让出一次主线程，避免阻塞
    if (stats.errorCount % 1000 === 0) {
      await new Promise(resolve => setImmediate(resolve));
    }
    if (line.includes('ERROR')) {
      stats.errorCount++;
    }
  }

  fileStream.destroy();
  return { filePath, stats };
}

async function statistic() {
  let promisePool = [];
  const results = [];

  while (filePathList.length > 0) {
    // 填满任务池
    while (promisePool.length < poolSize && filePathList.length > 0) {
      const filePath = filePathList.shift();
      const promise = processFile(filePath)
        .then(result => {
          promisePool = promisePool.filter(p => p !== promise);
          results.push(result);
          console.log(`${filePath} DONE!, remain promises: ${promisePool.length}`);
          return result;
        })
        .catch(err => {
          promisePool = promisePool.filter(p => p !== promise);
          console.error(`${filePath} FAILED: ${err.message}`);
          throw err;
        });
      promisePool.push(promise);
    }

    // 等待至少一个任务完成，再补充新任务
    await Promise.race(promisePool);
  }

  // 等待所有剩余任务完成
  await Promise.all(promisePool);
  return results;
}

(async () => {
  const start = Date.now();
  console.log(`Start statistic algorithm, timestamp: ${start}`);
  const results = await statistic();
  const end = Date.now();
  console.log(`End statistic algorithm, timestamp: ${end}`);
  console.log(`It took ${(end - start) / 1000} sec`);
})();

• 原代码中new Promise(async (resolve, reject) => {})的写法存在问题：会导致Promise嵌套，且reject可能无法被正确捕获，建议直接使用async函数返回Promise。
• CPU密集型任务必须用Worker Threads，单线程事件循环无法突破并发瓶颈；IO密集型任务可通过优化Promise Pool实现高效并发。

内容的提问来源于stack exchange，提问作者Carl.t_