14.2 线程池、任务队列与 Atomics 同步
14.1 证明了多核确实能吃下大头，但也留下一个问题：每提交一批任务就 new 一批 worker，冷启动的开销比任务本身还大。正确做法是把 worker 养起来复用——这就是线程池。本节从零写一个：固定 N 个 worker、一条所有 worker 共享的任务队列、一套用 Atomics 写出来的同步原语，最后用 1000 个任务量出串行与并行的差距。
共享内存布局：一个 Int32Array 装下整条队列
线程池的共享状态只有一块 SharedArrayBuffer，被切成一排 Int32 字：
| 字下标 | 字段 | 作用 |
|---|---|---|
| 0 | LOCK | 互斥锁：0 空闲 / 1 占用 |
| 1 | HEAD | 队头下标（下一个出队的槽位） |
| 2 | TAIL | 队尾下标（下一个入队的槽位） |
| 3 | CAP | 环形缓冲容量 |
| 4 | PENDING | 已提交但未完成的任务数 |
| 5 | SIGNAL | 唤醒信号：入队时自增 |
| 6 | SHUTDOWN | 关闭标志 |
| 8–71 | STAT_BASE | 每个 worker 的处理计数（最多 64 个 worker） |
| 72 起 | SLOT_BASE | 环形槽位区，每个槽位存一个任务号 |
为什么不直接传函数？因为 postMessage 只能传可结构化克隆的数据，函数传不过去。所以任务在队列里只是一个整数号，worker 拿到号之后自己查表执行对应的计算。这是所有基于 worker 的任务系统的共同约束，Go 里没这个问题（闭包能直接 go f()），JS 里必须绕。
可运行产物：worker 与 JobSystem
worker 的逻辑是：取任务、执行、自减未完成数；队列空了就阻塞在唤醒信号上。这是 src/core/job-worker.mjs，层级 L1。
// file: src/core/job-worker.mjs
// tier: L1
// verify: npx vitest run src/core/job-system.test.ts
//
// 线程池 worker：从共享环形队列取任务号，跑确定性的计算，再自减未完成数。
// 队列空时用 Atomics.wait 阻塞在唤醒信号上，由主线程入队后唤醒。
import { parentPort, workerData } from 'node:worker_threads';
import { HEAD, LOCK, PENDING, SHUTDOWN, SIGNAL, SLOT_BASE, STAT_BASE, TAIL } from './job-system.ts';
const { sab, workerIndex, iter } = workerData;
const words = new Int32Array(sab, 0, SLOT_BASE);
const cap = Atomics.load(words, 3);
const slots = new Int32Array(sab, SLOT_BASE * 4, cap);
function lock() {
for (let spin = 0; ; spin++) {
if (Atomics.compareExchange(words, LOCK, 0, 1) === 0) return;
if (spin > 32) Atomics.wait(words, LOCK, 1, 1);
}
}
function unlock() {
Atomics.store(words, LOCK, 0);
Atomics.notify(words, LOCK, 1);
}
function tryPop() {
lock();
const h = Atomics.load(words, HEAD);
const t = Atomics.load(words, TAIL);
let id = -1;
if (h < t) {
id = slots[h % cap];
Atomics.store(words, HEAD, h + 1);
}
unlock();
return id;
}
// 任务体：由任务号派生的确定性计算，worker 侧自己实现（跨线程只能传整数）。
function runTask(id) {
let x = (id * 2654435761) >>> 0;
for (let k = 0; k < iter; k++) {
x ^= x << 13; x >>>= 0;
x ^= x >>> 17;
x ^= x << 5; x >>>= 0;
}
return x;
}
for (;;) {
const id = tryPop();
if (id >= 0) {
runTask(id);
Atomics.add(words, STAT_BASE + workerIndex, 1);
Atomics.sub(words, PENDING, 1);
continue;
}
if (Atomics.load(words, SHUTDOWN) === 1) break;
const s = Atomics.load(words, SIGNAL);
if (Atomics.load(words, SHUTDOWN) === 1) break;
Atomics.wait(words, SIGNAL, s, 50);
}
parentPort.postMessage({ workerIndex });
主线程侧是 JobSystem：构造时起 N 个 worker 并建好共享缓冲，submit 往环形队列里塞任务号，run 阻塞到全部完成。这是 src/core/job-system.ts：
// file: src/core/job-system.ts
// tier: L1
// verify: npx vitest run src/core/job-system.test.ts
//
// helio 内核：基于 worker_threads 的线程池 + 共享环形任务队列。
// 队列由一把 Atomics 互斥锁保护；空闲 worker 阻塞在 SIGNAL 上。
// 共享内存布局（Int32 字）：[0]锁 [1]队头 [2]队尾 [3]容量 [4]未完成数
// [5]唤醒信号 [6]关闭标志 [7]保留 [8..72)各 worker 处理计数 [72..]槽位区。
import { Worker } from 'node:worker_threads';
import { availableParallelism } from 'node:os';
import { performance } from 'node:perf_hooks';
export const LOCK = 0;
export const HEAD = 1;
export const TAIL = 2;
export const CAP = 3;
export const PENDING = 4;
export const SIGNAL = 5;
export const SHUTDOWN = 6;
export const STAT_BASE = 8;
export const MAX_WORKERS = 64;
export const SLOT_BASE = STAT_BASE + MAX_WORKERS; // 槽位区起点 = 72
export interface RunStats {
ms: number;
perWorker: number[];
total: number;
}
export class JobSystem {
#words: Int32Array;
#slots: Int32Array;
#cap: number;
#workerCount: number;
#workers: Worker[] = [];
constructor(workerCount = Math.max(1, availableParallelism() - 1), capacity = 4096, taskIterations = 200_000) {
this.#workerCount = workerCount;
this.#cap = capacity;
const sab = new SharedArrayBuffer((SLOT_BASE + capacity) * 4);
this.#words = new Int32Array(sab, 0, SLOT_BASE);
this.#slots = new Int32Array(sab, SLOT_BASE * 4, capacity);
Atomics.store(this.#words, CAP, capacity);
for (let i = 0; i < workerCount; i++) {
this.#workers.push(
new Worker(new URL('./job-worker.mjs', import.meta.url), {
workerData: { sab, workerIndex: i, iter: taskIterations },
}),
);
}
}
get workerCount(): number {
return this.#workerCount;
}
#lock(): void {
for (let spin = 0; ; spin++) {
if (Atomics.compareExchange(this.#words, LOCK, 0, 1) === 0) return;
if (spin > 32) Atomics.wait(this.#words, LOCK, 1, 1); // 自旋失败后让出
}
}
#unlock(): void {
Atomics.store(this.#words, LOCK, 0);
Atomics.notify(this.#words, LOCK, 1);
}
/** 提交一个任务号（必须是 [0, 2^31) 的整数）。 */
submit(id: number): void {
this.#lock();
const t = Atomics.load(this.#words, TAIL);
this.#slots[t % this.#cap] = id;
Atomics.store(this.#words, TAIL, t + 1);
this.#unlock();
}
/** 提交 0..count-1 共 count 个任务，阻塞到全部完成，返回统计。 */
async run(count: number): Promise<RunStats> {
const t0 = performance.now();
// 上一轮已全部完成（PENDING==0），此刻重置计数是安全的
for (let i = 0; i < this.#workerCount; i++) Atomics.store(this.#words, STAT_BASE + i, 0);
Atomics.store(this.#words, PENDING, count);
for (let i = 0; i < count; i++) this.submit(i);
Atomics.add(this.#words, SIGNAL, 1);
Atomics.notify(this.#words, SIGNAL, this.#workerCount);
while (Atomics.load(this.#words, PENDING) > 0) {
await new Promise((r) => setTimeout(r, 1));
}
const ms = performance.now() - t0;
const perWorker = Array.from({ length: this.#workerCount }, (_, i) =>
Atomics.load(this.#words, STAT_BASE + i),
);
return { ms, perWorker, total: perWorker.reduce((a, b) => a + b, 0) };
}
async terminate(): Promise<void> {
Atomics.store(this.#words, SHUTDOWN, 1);
Atomics.add(this.#words, SIGNAL, 1);
Atomics.notify(this.#words, SIGNAL, this.#workerCount);
await Promise.all(this.#workers.map((w) => w.terminate()));
this.#workers = [];
}
}
两个细节值得点出。第一，锁用自旋 + 让出：Atomics.compareExchange 抢锁，抢不到就自旋，自旋超过 32 次再 Atomics.wait 挂起——避免短临界区里频繁陷入内核态。第二，完成检测靠 PENDING 计数：主线程只等这一个数归零，不必挨个 await worker 的消息。
正确性：vitest 用例
先证明它算得对，再看它跑得快不快。这是 src/core/job-system.test.ts：
// file: src/core/job-system.test.ts
// tier: L1
// verify: npx vitest run src/core/job-system.test.ts
import { afterAll, describe, expect, it } from 'vitest';
import { JobSystem } from './job-system.ts';
const js = new JobSystem(4, 256);
describe('JobSystem', () => {
it('200 个任务全部完成，计数之和等于任务数', async () => {
const r = await js.run(200);
expect(r.total).toBe(200);
expect(r.perWorker.reduce((a, b) => a + b, 0)).toBe(200);
});
it('再次提交仍能跑完（队列复用）', async () => {
const r = await js.run(300);
expect(r.total).toBe(300);
});
it('没有任何 worker 闲置（都被派到活）', async () => {
const r = await js.run(400);
expect(r.perWorker.every((n) => n > 0)).toBe(true);
});
});
afterAll(async () => {
await js.terminate();
});
本机实测 npx vitest run src/core/job-system.test.ts 三条全过：200 个任务的计数之和恰好 200，连续两轮提交都能跑完（队列复用正确），且每个 worker 都分到了活。
实测：1000 个任务，串行 vs 4 worker
// file: src/core/job-bench.ts
// tier: B
// verify: node src/core/job-bench.ts
//
// 线程池实测：同一批 1000 个任务，先单线程跑一遍，再交给 4 个 worker 的
// JobSystem 跑一遍。任务体与 job-worker.mjs 的 runTask 一致。
import { performance } from 'node:perf_hooks';
import { JobSystem } from './job-system.ts';
const TASKS = 1000;
function runTask(id: number, iter: number): number {
let x = (id * 2654435761) >>> 0;
for (let k = 0; k < iter; k++) {
x ^= x << 13; x >>>= 0;
x ^= x >>> 17; x ^= x << 5; x >>>= 0;
}
return x;
}
async function phase(iter: number): Promise<void> {
let sink = 0;
const t0 = performance.now();
for (let i = 0; i < TASKS; i++) sink = (sink + runTask(i, iter)) >>> 0;
const serialMs = performance.now() - t0;
const js = new JobSystem(4, 4096, iter);
const r = await js.run(TASKS);
await js.terminate();
console.log(
`iter=${String(iter).padStart(6)} 单线程 ${serialMs.toFixed(1)} ms ` +
`线程池 ${r.ms.toFixed(1)} ms ${(serialMs / r.ms).toFixed(2)}x ` +
`分布 ${r.perWorker.join('/')} (sink=${sink})`,
);
}
await phase(200_000); // 每个任务够粗：并行有收益
await phase(2_000); // 每个任务太碎：调度开销盖过收益
本机实测（Node 26.5.0，10 逻辑核）：
iter=200000 单线程 405.5 ms 线程池 163.0 ms 2.49x 分布 257/248/251/244
iter= 2000 单线程 3.9 ms 线程池 46.9 ms 0.08x 分布 47/157/128/668
任务粒度：什么时候不该用线程池
上面第二行是这一节最重要的一组数字。每个任务缩到 iter=2000（约 4 微秒）之后，1000 个任务单线程只要 3.9 ms，丢进 4 worker 的线程池却要 46.9 ms——慢了 12 倍。原因不是并行本身有问题，而是任务太碎：每个任务分摊到的计算，还不如「抢一次锁 + 一次唤醒 + 一次计数」的固定开销多。
把这段数字和 14.1 的探针放一起看，规律很清楚：任务够粗时（每个约 0.4 ms、1000 个）线程池能到 2.5 倍左右，任务太碎时（每个约 4 微秒）反而掉到 0.08 倍。所以线程池的正确用法是批处理：不要「一个实体一个任务」，而是「一批实体一个任务」。这也是为什么第 13 章的查询分组、第 15 章的任务图都会把大量小工作合并成粗粒度的 job——并行的单位是 job，不是对象。
还有两条工程经验。其一，run() 里主线程用 setTimeout(1) 轮询 PENDING，有最多 1 ms 的检测延迟；对一帧十几毫秒的游戏可以接受，想更精确就让主线程也领一部分任务。其二，任务太碎时分布会变得极不均匀——本机观察到过「一个 worker 抢到全部 1000 个任务、其余三个空转」的极端情况，这正说明固定开销主导时，调度器已经不是在分计算，而是在分开销。
14.3 会把这条「所有 worker 抢同一条队列」的共享队列换掉：那把锁在任务多、核多的时候会成为新的瓶颈，而 work stealing 用「每个 worker 一条自己的双端队列」把抢锁降到接近零。
阅读导航：上一节：14.1 为什么 JS 也需要 Job System · 下一节：14.3 work stealing 队列的真实实现 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。