使用worker_threads拆分30万条数组并结合Jest执行测试的方案咨询

现有代码核心问题 #

• worker.on("message", parentResolve) 绑定错误：第一个Worker返回消息就会直接resolve外层Promise，无法收集所有Worker的执行结果
• 未做CPU核心数边界处理：当设备CPU核心数≤2时，CPUS_NUMBER_LIMIT会为0，导致数组拆分后没有线程处理任务
• eval方式加载ts入口稳定性差：容易出现工作线程编译上下文、路径匹配错误
• 未正确聚合多线程执行结果，测试数据无法统一回收校验

完整落地代码 #

1. 主线程逻辑（数组拆分+Worker调度） #

import os from 'os';
import path from 'path';
import { Worker } from 'worker_threads';

// 边界处理：最少预留1个工作线程
const CPUS_NUMBER_LIMIT = Math.max(1, os.cpus().length - 2);

// 数组拆分+多线程调度函数
const splitArrayRunTest = <T>(arr: T[]): Promise<any[]> => {
  return new Promise((parentResolve, parentReject) => {
    const segmentSize = Math.ceil(arr.length / CPUS_NUMBER_LIMIT);
    const segments: T[][] = [];

    // 拆分原始数组
    for (let index = 0; index < CPUS_NUMBER_LIMIT; index++) {
      const start = index * segmentSize;
      const end = start + segmentSize;
      segments.push(arr.slice(start, end));
    }

    // 为每个分段启动独立Worker
    Promise.all(
      segments.map(
        (seg) =>
          new Promise((segResolve, segReject) => {
            const worker = new Worker(path.join(__dirname, './worker.ts'), {
              workerData: {
                seg
              },
              // ts-node环境下运行的配置，打包后运行可删除
              execArgv: ['-r', 'ts-node/register']
            });

            // 收集当前Worker返回的测试结果
            worker.on("message", (result) => segResolve(result));
            worker.on("error", (err) => segReject(err));
            worker.on("exit", (code) => {
              if (code !== 0) {
                segReject(new Error(`Worker exited with error code: ${code}`));
              }
            });
          })
      )
    )
    .then((allResults) => parentResolve(allResults.flat()))
    .catch((err) => parentReject(err));
  });
};

2. Worker线程逻辑（worker.ts，执行分段测试） #

import { parentPort, workerData } from 'worker_threads';

// 从主线程接收拆分后的数组分段
const { seg } = workerData;

// 执行你需要的Jest测试逻辑，这里替换为你的实际测试代码
const runSegmentTest = async () => {
  const testResults = [];
  for (const item of seg) {
    // 替换为你的单条数据测试逻辑
    // 示例：const res = await yourTestFunc(item);
    // testResults.push(res);
  }
  // 测试完成后将结果发回主线程
  parentPort?.postMessage(testResults);
};

runSegmentTest().catch(err => {
  console.error('Worker test failed:', err);
  throw err;
});

Jest环境适配配置 #

• 如果你是在Jest测试用例中调用多线程逻辑，需要在jest.config.js中添加配置，避免Worker线程加载ts文件失败：

module.exports = {
  // 其他原有配置保留
  testEnvironment: 'node',
  extensionsToTreatAsEsm: ['.ts'],
  globals: {
    'ts-jest': {
      useESM: true,
    }
  }
};

• 若单条测试逻辑较轻，可适当调大segmentSize减少线程创建开销，避免线程切换消耗超过并行收益
• 测试过程中产生的公共资源（如数据库连接、全局缓存）要做好线程隔离，不要共用主线程的连接实例，避免竞态条件

内容的提问来源于stack exchange，提问作者SimoneMeta