Web Workers与多线程并发—SharedArrayBuffer、Atomics锁与线程池设计|新宇宙博客sort
(a, b) =>
const
new
Worker
'sort.worker.js'
postMessage
onmessage
(e) =>
renderResult
data
Web Workers 的本质：独立的 JS 执行上下文（独立 EventLoop、独立堆），通过消息传递与主线程通信。
三种 Worker 的职责划分
| 类型 | 作用域 | 生命周期 | 主要用途 |
|---|
| Dedicated Worker | 单一页面 | 页面关闭时销毁 | CPU 密集计算 |
| Shared Worker | 同源多个页面 | 无连接时销毁 | 跨 Tab 共享状态 |
| Service Worker | 同源所有页面 | 独立于页面 | 离线缓存、网络代理 |
Worker 线程可用 API
self.postMessage()
importScripts('lib.js')
fetch / XMLHttpRequest
IndexedDB
WebSockets
Crypto API
FileReader
TextEncoder/TextDecoder
Performance API
setTimeout / setInterval
Promise / async-await
console.log
document / window
localStorage
alert / confirm
canvas（部分浏览器支持 OffscreenCanvas）
Dedicated Worker 深度解析
基础使用
const worker = new Worker('./worker.js', {
type: 'module',
name: 'myWorker',
});
worker.postMessage({ cmd: 'sort', data: [3, 1, 4, 1, 5, 9, 2, 6] });
worker.onmessage = (e) => {
console.log('Worker 返回：', e.data);
};
worker.onerror = (e) => {
console.error('Worker 错误：', e.message, e.filename, e.lineno);
e.preventDefault();
};
worker.onmessageerror = (e) => {
console.error('消息反序列化失败：', e);
};
worker.terminate();
self.onmessage = (e) => {
const { cmd, data } = e.data;
if (cmd === 'sort') {
const sorted = [...data].sort((a, b) => a - b);
self.postMessage({ cmd: 'sortResult', data: sorted });
}
};
self.addEventListener('message', (e) => { });
Inline Worker（无需单独文件）
function createInlineWorker(fn) {
const code = `(${fn.toString()})()`;
const blob = new Blob([code], { type: 'application/javascript' });
const url = URL.createObjectURL(blob);
const worker = new Worker(url);
URL.revokeObjectURL(url);
return worker;
}
const worker = createInlineWorker(function() {
self.onmessage = (e) => {
const result = e.data * e.data;
self.postMessage(result);
};
});
worker.postMessage(10);
worker.onmessage = (e) => console.log(e.data);
Module Worker
const worker = new Worker('./worker.mjs', { type: 'module' });
import { heavyCompute } from './utils.mjs';
self.onmessage = async (e) => {
const result = await heavyCompute(e.data);
self.postMessage(result);
};
postMessage 通信机制
结构化克隆算法（Structured Clone）
postMessage 使用结构化克隆序列化数据，支持的类型远多于 JSON.stringify：
worker.postMessage({
num: 42,
str: 'hello',
bool: true,
nil: null,
date: new Date(),
regexp: /abc/gi,
map: new Map([['key', 'value']]),
set: new Set([1, 2, 3]),
error: new Error('test'),
buf: new ArrayBuffer(8),
u8: new Uint8Array([1, 2, 3]),
});
worker.postMessage(function() {});
worker.postMessage(document.body);
worker.postMessage(Symbol('id'));
worker.postMessage(new Proxy({}, {}));
双向通信封装（Promise 化）
class WorkerBridge {
#worker;
#pending = new Map();
#nextId = 0;
constructor(src) {
this.#worker = new Worker(src, { type: 'module' });
this.#worker.onmessage = (e) => this.#handleMessage(e);
this.#worker.onerror = (e) => this.#handleError(e);
}
call(method, ...args) {
const id = this.#nextId++;
return new Promise((resolve, reject) => {
this.#pending.set(id, { resolve, reject });
this.#worker.postMessage({ id, method, args });
});
}
#handleMessage({ data: { id, result, error } }) {
const pending = this.#pending.get(id);
if (!pending) return;
this.#pending.delete(id);
error ? pending.reject(new Error(error)) : pending.resolve(result);
}
#handleError(e) {
for (const [id, { reject }] of this.#pending) {
reject(new Error(e.message));
}
this.#pending.clear();
}
terminate() { this.#worker.terminate(); }
}
const handlers = {
add: ([a, b]) => a + b,
sort: ([arr]) => [...arr].sort((a, b) => a - b),
fib: ([n]) => fibonacci(n),
};
self.onmessage = ({ data: { id, method, args } }) => {
try {
const result = handlers[method]?.(args);
self.postMessage({ id, result });
} catch (e) {
self.postMessage({ id, error: e.message });
}
};
const bridge = new WorkerBridge('./worker.js');
const sum = await bridge.call('add', 3, 4);
console.log(sum);
Transferable Objects 零拷贝
原理
普通 postMessage 拷贝数据（O(n) 时间复杂度）。对 Transferable 对象，只转移所有权指针（O(1)），原持有方的对象被 detach。
支持 Transferable 的类型
const buf = new ArrayBuffer(1024 * 1024 * 100);
worker.postMessage(buf, [buf]);
console.log(buf.byteLength);
const { port1, port2 } = new MessageChannel();
worker.postMessage({ port: port2 }, [port2]);
port1.onmessage = (e) => console.log('主线程收到：', e.data);
port1.postMessage('ping');
const canvas = document.getElementById('myCanvas');
const offscreen = canvas.transferControlToOffscreen();
worker.postMessage({ canvas: offscreen }, [offscreen]);
const [readable, writable] = new TransformStream().readable.tee();
性能对比
const SIZE = 50 * 1024 * 1024;
const buf = new ArrayBuffer(SIZE);
console.time('clone');
worker1.postMessage(buf);
console.timeEnd('clone');
const buf2 = new ArrayBuffer(SIZE);
console.time('transfer');
worker2.postMessage(buf2, [buf2]);
console.timeEnd('transfer');
OffscreenCanvas 实战
const canvas = document.getElementById('canvas');
const offscreen = canvas.transferControlToOffscreen();
const worker = new Worker('./canvas-worker.js');
worker.postMessage(
{ type: 'init', canvas: offscreen, width: 800, height: 600 },
[offscreen]
);
worker.postMessage({ type: 'draw', data: complexSceneData });
let ctx;
self.onmessage = ({ data }) => {
if (data.type === 'init') {
ctx = data.canvas.getContext('2d');
ctx.canvas.width = data.width;
ctx.canvas.height = data.height;
} else if (data.type === 'draw') {
renderComplexScene(ctx, data.data);
}
};
SharedArrayBuffer + Atomics 共享内存
基础同步原语
const sab = new SharedArrayBuffer(4);
const arr = new Int32Array(sab);
Atomics.store(arr, 0, 0);
const worker = new Worker('./atomic-worker.js');
worker.postMessage({ sab });
const { async, value } = Atomics.waitAsync(arr, 0, 0);
if (async) {
value.then(() => console.log('Worker 已完成，值：', Atomics.load(arr, 0)));
}
self.onmessage = ({ data: { sab } }) => {
const arr = new Int32Array(sab);
for (let i = 0; i < 1_000_000; i++) {
Atomics.add(arr, 0, 1);
}
Atomics.notify(arr, 0, Infinity);
};
实现 Mutex（互斥锁）
class Mutex {
#lock;
static UNLOCKED = 0;
static LOCKED = 1;
constructor(sab, byteOffset = 0) {
this.#lock = new Int32Array(sab, byteOffset, 1);
}
lock() {
while (true) {
const old = Atomics.compareExchange(
this.#lock, 0,
Mutex.UNLOCKED,
Mutex.LOCKED
);
if (old === Mutex.UNLOCKED) return;
Atomics.wait(this.#lock, 0, Mutex.LOCKED);
}
}
unlock() {
Atomics.store(this.#lock, 0, Mutex.UNLOCKED);
Atomics.notify(this.#lock, 0, 1);
}
withLock(fn) {
this.lock();
try { return fn(); }
finally { this.unlock(); }
}
}
RingBuffer（无锁生产者-消费者）
class RingBuffer {
#buf;
#head;
#tail;
#data;
constructor(capacity) {
this.#buf = new SharedArrayBuffer(8 + capacity * 4);
this.#head = new Int32Array(this.#buf, 0, 1);
this.#tail = new Int32Array(this.#buf, 4, 1);
this.#data = new Int32Array(this.#buf, 8, capacity);
this.capacity = capacity;
}
get sab() { return this.#buf; }
write(value) {
const tail = Atomics.load(this.#tail, 0);
const head = Atomics.load(this.#head, 0);
const next = (tail + 1) % this.capacity;
if (next === head) return false;
Atomics.store(this.#data, tail, value);
Atomics.store(this.#tail, 0, next);
Atomics.notify(this.#head, 0, 1);
return true;
}
read() {
const head = Atomics.load(this.#head, 0);
const tail = Atomics.load(this.#tail, 0);
if (head === tail) return null;
const value = Atomics.load(this.#data, head);
Atomics.store(this.#head, 0, (head + 1) % this.capacity);
return value;
}
}
手写验证：线程池实现
完整线程池
class ThreadPool {
#workers = [];
#taskQueue = [];
#freeWorkers = [];
#pending = new Map();
#taskId = 0;
constructor(workerSrc, size = navigator.hardwareConcurrency ?? 4) {
for (let i = 0; i < size; i++) {
this.#createWorker(workerSrc);
}
}
#createWorker(src) {
const worker = new Worker(src, { type: 'module' });
worker.onmessage = ({ data }) => this.#onWorkerMessage(worker, data);
worker.onerror = (e) => this.#onWorkerError(worker, e);
this.#workers.push(worker);
this.#freeWorkers.push(worker);
}
#onWorkerMessage(worker, { id, result, error }) {
const { resolve, reject } = this.#pending.get(id) ?? {};
this.#pending.delete(id);
this.#freeWorkers.push(worker);
error ? reject?.(new Error(error)) : resolve?.(result);
this.#dispatch();
}
#onWorkerError(worker, e) {
this.#workers = this.#workers.filter(w => w !== worker);
this.#freeWorkers = this.#freeWorkers.filter(w => w !== worker);
worker.terminate();
this.#createWorker(worker._src);
}
#dispatch() {
if (this.#taskQueue.length === 0) return;
if (this.#freeWorkers.length === 0) return;
const worker = this.#freeWorkers.pop();
const { id, task, transfer } = this.#taskQueue.shift();
worker.postMessage({ id, ...task }, transfer ?? []);
}
exec(task, transfer) {
const id = this.#taskId++;
return new Promise((resolve, reject) => {
this.#pending.set(id, { resolve, reject });
this.#taskQueue.push({ id, task, transfer });
this.#dispatch();
});
}
async execBatch(tasks) {
return Promise.all(tasks.map(t => this.exec(t.task, t.transfer)));
}
get size() { return this.#workers.length; }
get freeCount() { return this.#freeWorkers.length; }
get queueLen() { return this.#taskQueue.length; }
terminate() {
this.#workers.forEach(w => w.terminate());
this.#workers = [];
this.#freeWorkers = [];
for (const { reject } of this.#pending.values()) {
reject(new Error('ThreadPool terminated'));
}
this.#pending.clear();
}
}
const handlers = {
sort(data) {
const arr = new Int32Array(data);
arr.sort();
return arr.buffer;
},
fib(n) {
function fib(n) { return n <= 1 ? n : fib(n - 1) + fib(n - 2); }
return fib(n);
},
compress(data) {
return data.reduce((acc, v) => acc + v, 0);
},
};
self.onmessage = ({ data: { id, method, args, buffer } }) => {
try {
const result = handlers[method]?.(buffer ?? args);
const isBuffer = result instanceof ArrayBuffer;
self.postMessage(
{ id, result },
isBuffer ? [result] : []
);
} catch (e) {
self.postMessage({ id, error: e.message });
}
};
const pool = new ThreadPool('./pool-worker.js', 4);
const results = await Promise.all(
Array.from({ length: 100 }, (_, i) =>
pool.exec({ method: 'fib', args: 35 + (i % 5) })
)
);
const buf = new ArrayBuffer(4 * 1_000_000);
new Int32Array(buf).forEach((_, i, a) => (a[i] = Math.random() * 1e9 | 0));
const sortedBuf = await pool.exec(
{ method: 'sort', buffer: buf },
[buf]
);
console.log('排序完成，前 10 个：', new Int32Array(sortedBuf).slice(0, 10));
pool.terminate();
Worker 进度上报
self.onmessage = ({ data: { id, items } }) => {
const total = items.length;
let processed = 0;
for (const item of items) {
heavyProcess(item);
processed++;
if (processed % 1000 === 0) {
self.postMessage({ type: 'progress', id, progress: processed / total });
}
}
self.postMessage({ type: 'result', id, result: 'done' });
};
worker.onmessage = ({ data }) => {
if (data.type === 'progress') {
progressBar.style.width = (data.progress * 100) + '%';
} else if (data.type === 'result') {
console.log('完成！', data.result);
}
};
深度追问
Q1：Shared Worker 和 Dedicated Worker 的应用场景区别？
Dedicated Worker 为单个页面服务，生命周期与创建页面绑定，适合 CPU 密集型计算（图像处理、加密运算、大数据排序）。
Shared Worker 可被同源多个页面/iframe 共享，通过 MessagePort 连接，适合跨 Tab 共享状态（如实时数据同步、WebSocket 连接复用）。注意：Shared Worker 的错误调试需在 chrome://inspect/#workers 中进行。
Q2：为什么 SharedArrayBuffer 需要 COOP/COEP 头？
2018 年 Spectre 漏洞披露后，攻击者可利用高精度计时器（performance.now）通过侧信道推断跨域内存内容。SharedArrayBuffer 可用于构建精度更高的计时器（共享内存自旋等待）。为此浏览器需要：
Cross-Origin-Opener-Policy: same-origin：隔离浏览上下文组，防止跨域窗口引用Cross-Origin-Embedder-Policy: require-corp：要求所有嵌入资源明确授权
满足这两个条件后，crossOriginIsolated 为 true，SharedArrayBuffer 才可用。
理论上等于 CPU 核心数（navigator.hardwareConcurrency），避免线程切换开销。但需考虑：
- IO 密集型任务（fetch、IndexedDB）可适当增加（线程常处于等待状态）
- CPU 密集型任务等于核心数即可
- 留 1 个核心给主线程，防止主线程饥饿
Q4：Worker 内能使用 DOM 吗？如何操作 Canvas？
Worker 内无法访问 DOM（因为 DOM API 不是线程安全的）。但通过 OffscreenCanvas 可以在 Worker 中进行 Canvas 渲染：主线程调用 canvas.transferControlToOffscreen() 将控制权转移给 Worker，Worker 获得完整的 2D/WebGL 上下文，实现渲染不占用主线程。
