Web Workers 并行计算实战：SharedArrayBuffer 从零构建高性能任务调度器

JavaScript 是单线程语言，但这不意味着它不能做并行计算。Web Workers API 提供了真正的多线程能力，而 SharedArrayBuffer + Atomics 组合更是让我们能在多个 Worker 之间实现零拷贝的数据共享。2024 年以来，所有主流浏览器都已全面支持 SharedArrayBuffer（包括 Safari），这意味着在前端做 CPU 密集型并行计算已经没有兼容性障碍。

根据 MDN 的数据，使用 Web Workers 进行并行计算可以将图像处理、大数据排序、加密计算等 CPU 密集型任务的耗时降低 60%-80%（取决于 CPU 核心数）。但现实情况是，大多数前端开发者对 Web Workers 的理解停留在 postMessage 的"Hello World"阶段——知道有这个东西，但从未在生产环境真正用过。

原因很简单：Worker 的原始 API 太底层了。没有 Worker Pool 概念、没有优雅的错误处理、SharedArrayBuffer 需要特殊的 HTTP 头配置。本文就是要解决这些问题——从零实现一个生产级的 Worker Pool 任务调度器，附带完整的错误处理、超时机制和性能监控，让你能直接用到项目里。

🔧 一、Web Workers 基础与通信模型

1.1 两种 Worker 的本质区别

很多开发者分不清 Worker 和 SharedWorker 的区别，这里用一张表说清楚：

💡 **提示：**绝大多数并行计算场景用 Web Worker 就够了。SharedWorker 适合做跨标签页的状态同步（如 WebSocket 连接复用），不适合做计算任务。

1.2 postMessage 的隐藏性能陷阱

postMessage 的默认行为是结构化克隆（Structured Clone），这意味着每次通信都会完整拷贝数据。传一个 100MB 的 ArrayBuffer，就会拷贝 100MB。

// ❌ 错误写法：每次传输都拷贝 100MB 数据
const hugeBuffer = new ArrayBuffer(100 * 1024 * 1024) // 100MB
worker.postMessage({ data: hugeBuffer }) // 触发 100MB 拷贝
console.log(hugeBuffer.byteLength) // 仍然是 100MB，原始数据还在

// ✅ 正确写法：使用 Transferable 零拷贝传输
const hugeBuffer = new ArrayBuffer(100 * 1024 * 1024) // 100MB
worker.postMessage({ data: hugeBuffer }, [hugeBuffer]) // 零拷贝，瞬间完成
console.log(hugeBuffer.byteLength) // 0！所有权已转移

Transferable 的原理是所有权转移——发送后原始 ArrayBuffer 的 byteLength 变为 0，接收方获得完整数据。这个过程不涉及任何内存拷贝，性能差异巨大：

⚠️ **警告：**Transferable 会转移所有权，发送方不能再访问该数据。如果发送方还需要用，必须先拷贝一份。

1.3 最小可运行示例

先跑通一个最简单的 Worker 通信：

// main.js — 主线程
const worker = new Worker(new URL('./worker.js', import.meta.url), { type: 'module' })

worker.onmessage = (event) => {
  console.log('主线程收到结果:', event.data)
}

// 发送计算任务
worker.postMessage({ type: 'fibonacci', n: 40 })

// worker.js — Worker 线程
self.onmessage = (event) => {
  const { type, n } = event.data

  if (type === 'fibonacci') {
    const result = fib(n)
    self.postMessage({ type: 'result', value: result })
  }
}

function fib(n) {
  if (n <= 1) return n
  return fib(n - 1) + fib(n - 2)
}

这个例子能跑，但有个致命问题：创建 Worker 有开销（约 5-10ms），而且每次只能处理一个任务。如果并发 100 个请求，就要创建 100 个 Worker？这显然不可行。

🚀 二、从零实现 Worker Pool 任务调度器

2.1 为什么需要 Worker Pool

直接 new Worker() 有三个致命问题：

1. 创建开销：每个 Worker 需要初始化独立的 JS 运行时，耗时 5-15ms。如果你的页面每秒处理 100 个请求，每个请求都创建新 Worker，光创建开销就要 500ms-1.5s。
2. 数量失控：浏览器对同源 Worker 有数量限制（Chrome 默认 64 个），如果不限制并发数量，很容易打爆浏览器。
3. 生命周期管理：Worker 出错了怎么办？任务卡死了怎么办？手动管理这些状态极其痛苦。

Worker Pool 的核心思想是预先创建固定数量的 Worker，反复复用。就像数据库连接池一样，这是处理并发任务的标准模式。

2.2 核心设计思路

一个生产级的 Worker Pool 需要解决三个核心问题：

1. Worker 复用 — 创建固定数量的 Worker，反复使用，避免创建销毁开销
2. 任务队列 — Worker 全忙时，任务排队等待
3. 超时与错误处理 — 单个任务不能卡住整个池子

下面是完整的 Worker Pool 实现：

// worker-pool.js — Worker Pool 任务调度器
export class WorkerPool {
  constructor(workerUrl, size = navigator.hardwareConcurrency || 4) {
    this.workers = []
    this.idleWorkers = []
    this.taskQueue = []
    this.taskId = 0
    this.pendingTasks = new Map() // taskId -> { resolve, reject, timer }

    // 创建固定数量的 Worker
    for (let i = 0; i < size; i++) {
      const worker = new Worker(workerUrl, { type: 'module' })
      worker.__poolId = i
      worker.onmessage = (event) => this._onWorkerMessage(worker, event)
      worker.onerror = (error) => this._onWorkerError(worker, error)
      this.workers.push(worker)
      this.idleWorkers.push(worker)
    }

    console.log(`Worker Pool 已初始化，共 ${size} 个 Worker`)
  }

  // 提交任务，返回 Promise
  execute(taskData, timeout = 30000) {
    return new Promise((resolve, reject) => {
      const id = ++this.taskId
      const timer = setTimeout(() => {
        this.pendingTasks.delete(id)
        reject(new Error(`Task ${id} timed out after ${timeout}ms`))
      }, timeout)

      this.pendingTasks.set(id, { resolve, reject, timer })

      const task = { id, data: taskData }

      if (this.idleWorkers.length > 0) {
        // 有空闲 Worker，立即执行
        const worker = this.idleWorkers.pop()
        this._dispatch(worker, task)
      } else {
        // 全部忙，加入队列
        this.taskQueue.push(task)
      }
    })
  }

  _dispatch(worker, task) {
    worker.__currentTaskId = task.id
    worker.postMessage(task.data)
  }

  _onWorkerMessage(worker, event) {
    const taskId = worker.__currentTaskId
    const pending = this.pendingTasks.get(taskId)

    if (pending) {
      clearTimeout(pending.timer)
      this.pendingTasks.delete(taskId)
      pending.resolve(event.data)
    }

    // 尝试从队列取下一个任务
    if (this.taskQueue.length > 0) {
      const nextTask = this.taskQueue.shift()
      this._dispatch(worker, nextTask)
    } else {
      this.idleWorkers.push(worker)
    }
  }

  _onWorkerError(worker, error) {
    const taskId = worker.__currentTaskId
    const pending = this.pendingTasks.get(taskId)

    if (pending) {
      clearTimeout(pending.timer)
      this.pendingTasks.delete(taskId)
      pending.reject(error)
    }

    // Worker 出错后放回空闲池（生产环境可考虑重建）
    this.idleWorkers.push(worker)
  }

  // 优雅关闭所有 Worker
  async shutdown() {
    for (const worker of this.workers) {
      worker.terminate()
    }
    this.workers = []
    this.idleWorkers = []
    for (const [, pending] of this.pendingTasks) {
      clearTimeout(pending.timer)
      pending.reject(new Error('Worker Pool shutting down'))
    }
    this.pendingTasks.clear()
  }

  get stats() {
    return {
      total: this.workers.length,
      idle: this.idleWorkers.length,
      busy: this.workers.length - this.idleWorkers.length,
      queued: this.taskQueue.length,
    }
  }
}

2.2 使用 Worker Pool 执行并行计算

配合一个通用的 Worker 脚本，可以执行各种 CPU 密集型任务：

// compute-worker.js — 通用计算 Worker
self.onmessage = (event) => {
  const { type, payload } = event.data

  try {
    let result

    switch (type) {
      case 'fibonacci':
        result = fibonacci(payload.n)
        break
      case 'matrixMultiply':
        result = matrixMultiply(payload.a, payload.b, payload.size)
        break
      case 'primeSieve':
        result = sieveOfEratosthenes(payload.limit)
        break
      default:
        throw new Error(`Unknown task type: ${type}`)
    }

    self.postMessage({ success: true, result })
  } catch (error) {
    self.postMessage({ success: false, error: error.message })
  }
}

function fibonacci(n) {
  // 尾递归优化版本，避免栈溢出
  function fibHelper(a, b, count) {
    if (count === 0) return a
    return fibHelper(b, a + b, count - 1)
  }
  return fibHelper(0n, 1n, n) // 使用 BigInt 支持大数
}

function matrixMultiply(a, b, size) {
  const result = new Float64Array(size * size)
  for (let i = 0; i < size; i++) {
    for (let k = 0; k < size; k++) {
      const aik = a[i * size + k]
      for (let j = 0; j < size; j++) {
        result[i * size + j] += aik * b[k * size + j]
      }
    }
  }
  return result
}

function sieveOfEratosthenes(limit) {
  const sieve = new Uint8Array(limit + 1)
  sieve[0] = sieve[1] = 1
  const primes = []
  for (let i = 2; i <= limit; i++) {
    if (!sieve[i]) {
      primes.push(i)
      for (let j = i * i; j <= limit; j += i) {
        sieve[j] = 1
      }
    }
  }
  return primes
}

调用方式非常简洁：

// main.js — 使用示例
import { WorkerPool } from './worker-pool.js'

const pool = new WorkerPool(
  new URL('./compute-worker.js', import.meta.url),
  navigator.hardwareConcurrency || 4
)

// 并行执行 4 个计算任务
const results = await Promise.all([
  pool.execute({ type: 'fibonacci', payload: { n: 1000000 } }),
  pool.execute({ type: 'primeSieve', payload: { limit: 10000000 } }),
  pool.execute({ type: 'matrixMultiply', payload: { a: bigMatrix, b: bigMatrix, size: 500 } }),
  pool.execute({ type: 'fibonacci', payload: { n: 500000 } }),
])

console.log('所有任务完成:', results)
console.log('Pool 状态:', pool.stats)

2.3 性能对比：单线程 vs Worker Pool

以下是在 8 核 CPU 上的实测数据（计算斐波那契数列第 N 项）：

⚠️ **警告：**Worker 之间传递 TypedArray（如 Float64Array）时，如果用 Transferable 传输，记得接收方拿到后是完整的，但发送方的引用会失效。在 Worker Pool 场景中，建议直接传递普通对象让 Worker 内部创建数据，避免所有权混乱。

💡 三、SharedArrayBuffer 实现零拷贝数据共享

3.1 三种通信模式对比

在深入 SharedArrayBuffer 之前，先理清 Worker 通信的三种模式，这是很多开发者混淆的地方：

📌 **记住：**选择通信模式的原则是：能用 Transferable 就不用克隆，能用 SharedArrayBuffer 就不用 Transferable。但 SharedArrayBuffer 的复杂度也最高，要权衡收益和维护成本。

3.2 为什么需要 SharedArrayBuffer

Worker Pool 的 postMessage 即使用了 Transferable，也只是避免了"拷贝"，数据还是在 Worker 之间传来传去。如果多个 Worker 需要同时读写同一块内存（比如并行更新一个大数组），就需要 SharedArrayBuffer。

SharedArrayBuffer 是真正的共享内存——所有 Worker 看到的是同一块物理内存，读写都是实时可见的。配合 Atomics API，可以实现线程安全的原子操作。

// shared-main.js — 主线程创建共享内存
const SHARED_SIZE = 1024 * 1024 // 1M 个 Int32 元素
const sharedBuffer = new SharedArrayBuffer(SHARED_SIZE * 4) // 4 bytes per Int32
const sharedArray = new Int32Array(sharedBuffer)

// 初始化数据
for (let i = 0; i < SHARED_SIZE; i++) {
  sharedArray[i] = i
}

// 传递给多个 Worker（注意：不是 Transferable，是真正的共享）
worker1.postMessage({ type: 'processChunk', array: sharedArray, start: 0, end: SHARED_SIZE / 2 })
worker2.postMessage({ type: 'processChunk', array: sharedArray, start: SHARED_SIZE / 2, end: SHARED_SIZE })

// shared-worker.js — Worker 线程直接操作共享内存
self.onmessage = (event) => {
  const { type, array, start, end } = event.data

  if (type === 'processChunk') {
    // 直接读写共享内存，零拷贝！
    for (let i = start; i < end; i++) {
      // Atomics.add 保证原子性，不会出现竞态条件
      Atomics.add(array, i, 1)
    }
    // 通知主线程处理完成
    Atomics.notify(array, start) // 唤醒等待在该位置的主线程
    self.postMessage({ done: true, start, end })
  }
}

3.2 Atomics 的关键操作

Atomics 提供了一组线程安全的操作，这是避免竞态条件的核心：

// atomics-demo.js — Atomics 常用操作
const sab = new SharedArrayBuffer(1024)
const arr = new Int32Array(sab)

// ✅ 原子加法：多个线程同时调用也不会出错
Atomics.add(arr, 0, 1)      // arr[0] += 1，返回旧值
Atomics.sub(arr, 0, 1)      // arr[0] -= 1

// ✅ 原子比较交换（CAS）：无锁并发的基础
Atomics.compareExchange(arr, 0, /* expected */ 5, /* replacement */ 10)
// 如果 arr[0] === 5，就设为 10，否则不动。返回旧值

// ✅ 原子读写：保证读到最新值
Atomics.store(arr, 0, 42)   // 原子写入
const val = Atomics.load(arr, 0) // 原子读取

// ✅ 线程等待与唤醒：实现 Worker 间的同步
Atomics.wait(arr, 0, 0)     // 阻塞，直到 arr[0] !== 0 或被 notify
Atomics.notify(arr, 0, 1)   // 唤醒 1 个等待的线程

📌 **记住：**SharedArrayBuffer 要求页面必须设置 COOP/COEP 响应头，否则浏览器会拒绝创建。这是一项安全措施，防止 Spectre 类攻击。具体设置：Cross-Origin-Opener-Policy: same-origin + Cross-Origin-Embedder-Policy: require-corp。

3.3 无锁计数器与并发安全

一个经典的应用场景：多个 Worker 并行处理任务，需要一个全局计数器统计完成数量。

// counter-main.js — 无锁并行计数器
const counterSab = new SharedArrayBuffer(4)
const counter = new Int32Array(counterSab)

Atomics.store(counter, 0, 0) // 初始化为 0

const TOTAL_TASKS = 1000
const CHUNK_SIZE = TOTAL_TASKS / 4

// 启动 4 个 Worker 并行处理
for (let i = 0; i < 4; i++) {
  const worker = new Worker(new URL('./counter-worker.js', import.meta.url), { type: 'module' })
  worker.postMessage({
    counter: counter,
    chunkSize: CHUNK_SIZE,
  })
}

// 定期检查进度
const progressInterval = setInterval(() => {
  const completed = Atomics.load(counter, 0)
  const percent = ((completed / TOTAL_TASKS) * 100).toFixed(1)
  console.log(`进度: ${completed}/${TOTAL_TASKS} (${percent}%)`)
  if (completed >= TOTAL_TASKS) clearInterval(progressInterval)
}, 100)

// counter-worker.js — Worker 端使用 Atomics 原子递增
self.onmessage = (event) => {
  const { counter, chunkSize } = event.data

  for (let i = 0; i < chunkSize; i++) {
    // 模拟耗时计算
    let sum = 0
    for (let j = 0; j < 100000; j++) sum += Math.sqrt(j)

    // 原子递增，多个 Worker 同时操作也不会丢数据
    Atomics.add(counter, 1, 1)
  }

  self.postMessage({ done: true })
}

⚡ **关键结论：**如果只是简单计数，用 Atomics.add 就够了。如果需要复杂的临界区保护（比如多个相关变量需要同时更新），可以基于 Atomics.compareExchange 实现自旋锁，但要注意自旋会消耗 CPU，适合临界区很短的场景。

🎯 四、实战场景与最佳实践

4.1 真实场景：CSV 大文件并行解析

假设用户上传了一个 200MB 的 CSV 文件，需要解析成 JSON 并做数据校验。单线程解析需要 8-12 秒，期间页面完全卡死。用 Worker Pool 可以将这个过程拆分到多个核心并行处理：

// csv-parallel.js — 并行解析大 CSV 文件
export async function parseCsvParallel(file, pool) {
  const text = await file.text()
  const lines = text.split('\n')
  const header = lines[0]
  const dataLines = lines.slice(1).filter(line => line.trim())

  // 按行数均匀分片
  const workerCount = navigator.hardwareConcurrency || 4
  const chunkSize = Math.ceil(dataLines.length / workerCount)

  const tasks = []
  for (let i = 0; i < workerCount; i++) {
    const chunk = dataLines.slice(i * chunkSize, (i + 1) * chunkSize)
    tasks.push(
      pool.execute({
        type: 'parseCsvChunk',
        payload: { header, lines: chunk },
      }, 30000) // 30 秒超时
    )
  }

  const results = await Promise.all(tasks)
  // 合并各 Worker 的解析结果
  return results.flatMap(r => r.result)
}

实测数据（200MB CSV，约 500 万行）：

4.2 图片并行处理

Web Workers 在图片处理场景中大放异彩。Canvas 的 getImageData 返回的 ImageData 底层就是一个 Uint8ClampedArray，可以直接通过 SharedArrayBuffer 共享。

// image-parallel.js — 并行图片灰度化
async function parallelGrayscale(imageData) {
  const { data, width, height } = imageData
  const pixelCount = width * height
  const workerCount = navigator.hardwareConcurrency || 4
  const chunkSize = Math.ceil(pixelCount / workerCount)

  // 创建共享内存，存放原始像素和处理结果
  const sharedSrc = new SharedArrayBuffer(data.byteLength)
  const sharedDst = new SharedArrayBuffer(data.byteLength)
  const srcView = new Uint8ClampedArray(sharedSrc)
  const dstView = new Uint8ClampedArray(sharedDst)

  // 拷贝原始数据到共享内存
  srcView.set(data)

  // 分发任务给多个 Worker
  const tasks = []
  for (let i = 0; i < workerCount; i++) {
    const start = i * chunkSize * 4 // RGBA = 4 bytes per pixel
    const end = Math.min((i + 1) * chunkSize * 4, data.byteLength)
    tasks.push(
      pool.execute({
        type: 'grayscaleChunk',
        payload: { src: sharedSrc, dst: sharedDst, start, end },
      })
    )
  }

  await Promise.all(tasks)

  // 结果直接在共享内存中，零拷贝取回
  const result = new ImageData(new Uint8ClampedArray(sharedDst), width, height)
  return result
}

4.3 Worker 使用 Checklist

4.4 常见坑点与避坑指南

坑点 1：Worker 路径问题

// ❌ 错误：字符串路径在不同打包工具下行为不一致
const worker = new Worker('./worker.js')

// ✅ 正确：使用 URL 构造函数，Vite/Webpack/Rspack 都支持
const worker = new Worker(new URL('./worker.js', import.meta.url), { type: 'module' })

坑点 2：SharedArrayBuffer 的浏览器安全限制

Chrome 和 Safari 要求页面必须处于「跨源隔离」状态才能使用 SharedArrayBuffer。如果你的页面加载了第三方资源（广告、分析脚本），需要设置 COEP 为 credentialless 而不是 require-corp，否则第三方资源会被阻止加载。

坑点 3：postMessage 中的循环引用

// ❌ 错误：循环引用会导致 DataCloneError
const obj = { name: 'test' }
obj.self = obj // 循环引用！
worker.postMessage(obj) // 报错：DataCloneError

// ✅ 正确：传递前移除循环引用，或者用 SharedArrayBuffer 替代

📊 五、总结与选型建议

Web Workers + SharedArrayBuffer 的组合为 JavaScript 打开了并行计算的大门。核心要点：

• 轻量任务用 postMessage + Transferable，实现零拷贝传输
• 多 Worker 共享状态用 SharedArrayBuffer + Atomics，实现真正的共享内存
• 生产环境必须用 Worker Pool，复用 Worker、管理队列、处理超时
• COOP/COEP 响应头是 SharedArrayBuffer 的前置条件，部署时别忘了配

对于 CPU 密集型的前端场景——图片/视频处理、数据可视化、加密解密、大文件校验——Web Workers 不是可选项，而是必选项。主线程只负责 UI 渲染，所有计算都应该下沉到 Worker 中。

相关工具推荐：

• 🧪 jsjson.com 在线工具箱 — JSON 处理、加密解密、编码转换等开发者工具
• 📖 MDN Web Workers 文档 — 官方 API 参考
• 🔗 Comlink — Google 出品的 Worker RPC 库，简化 Worker 通信