worker_threads

何时用

Node 单线程跑 JS——CPU 密集任务（图像处理、加密、复杂计算）会阻塞事件循环，所有请求都等。

// ❌ 阻塞
function fibonacci(n) {
    if (n < 2) return n;
    return fibonacci(n-1) + fibonacci(n-2);
}

server.on('request', () => {
    fibonacci(40);    // 阻塞 1+ 秒
});

解决方案优先级：

1. 算法优化（最有效）
2. 拆任务（每几 ms 让出循环）
3. worker_threads（同进程多线程，下面讲）
4. child_process / cluster（多进程，更重）

基础

// main.js
import { Worker } from 'worker_threads';

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

worker.on('message', result => {
    console.log('Result:', result);
});

worker.postMessage({ task: 'fib', n: 40 });

// worker.js
import { parentPort } from 'worker_threads';

function fibonacci(n) {
    if (n < 2) return n;
    return fibonacci(n-1) + fibonacci(n-2);
}

parentPort.on('message', msg => {
    if (msg.task === 'fib') {
        const result = fibonacci(msg.n);
        parentPort.postMessage(result);
    }
});

内联代码（不用单独文件）

import { Worker } from 'worker_threads';

const worker = new Worker(`
    const { parentPort } = require('worker_threads');
    parentPort.on('message', n => {
        let result = 0;
        for (let i = 0; i < n; i++) result += i;
        parentPort.postMessage(result);
    });
`, { eval: true });

worker.postMessage(1000000);
worker.on('message', console.log);

Promise 包装

实战中常包成 Promise：

function runInWorker(task) {
    return new Promise((resolve, reject) => {
        const worker = new Worker('./worker.js');
        worker.on('message', result => {
            resolve(result);
            worker.terminate();
        });
        worker.on('error', reject);
        worker.postMessage(task);
    });
}

const result = await runInWorker({ task: 'heavy-compute', data: [...] });

数据传递

Worker 间通信通过 postMessage——数据被结构化克隆（深拷贝）：

worker.postMessage({ name: 'Alice', tags: ['a', 'b'] });

⚠ 不能传函数 / Symbol / DOM。

大数据用 Transferable（零拷贝）

const buffer = new ArrayBuffer(100 * 1024 * 1024);   // 100 MB

worker.postMessage(buffer, [buffer]);
// 第二参数 = transferable list
// → buffer 所有权转给 worker，主线程不能再访问

SharedArrayBuffer（线程共享内存）

const sab = new SharedArrayBuffer(4);          // 4 字节
const view = new Int32Array(sab);

worker.postMessage(sab);                         // 不复制，共享

// 两端都能读写 view
view[0] = 42;

⚠ 需要正确用 Atomics 处理同步——否则有数据竞争。

线程池：worker pool

每次新建 worker 慢——生产用线程池：

import { Worker } from 'worker_threads';

class WorkerPool {
    constructor(size, script) {
        this.workers = Array.from({ length: size }, () => new Worker(script));
        this.queue = [];
        this.idle = [...this.workers];
    }

    run(task) {
        return new Promise((resolve, reject) => {
            const job = { task, resolve, reject };
            const w = this.idle.pop();
            if (w) this.assign(w, job);
            else this.queue.push(job);
        });
    }

    assign(worker, job) {
        worker.once('message', result => {
            job.resolve(result);
            const next = this.queue.shift();
            if (next) this.assign(worker, next);
            else this.idle.push(worker);
        });
        worker.postMessage(job.task);
    }
}

const pool = new WorkerPool(4, './worker.js');
const results = await Promise.all(tasks.map(t => pool.run(t)));

或用现成库：piscina（最流行）：

import Piscina from 'piscina';

const pool = new Piscina({ filename: './worker.js' });
const result = await pool.run({ a: 1, b: 2 });

worker_threads vs child_process vs cluster

坑

• worker 启动比想象的慢（几十 ms）—— 频繁短任务用线程池
• worker 不共享主进程内存——除非 SharedArrayBuffer
• worker 里不能用 process.exit（用 parentPort.close() 或退出 worker）
• 异常未捕获 → worker 退出 + main error 事件

下一篇：cluster 多进程。