JavaScript 中的并发与多线程：高级进阶

JavaScript 中的并发与多线程：高级进阶

JavaScript 传统上被视为单线程语言，然而现代 Node.js 生态系统提供了强大的工具来处理并发和真正的多线程。在本文中，我们将深入探讨高级技术：从事件循环的细微之处到使用 Worker Threads、SharedArrayBuffer 和 Atomics。您将学习如何有效分配负载、避免竞态条件以及构建可扩展的应用程序。

1. 事件循环、微任务和宏任务：并发的基础

理解事件循环是管理无阻塞并发的关键。在 Node.js 中，事件循环由多个阶段组成：定时器、I/O 回调、空闲、轮询、检查（setImmediate）和关闭。区分微任务（Promise、process.nextTick）和宏任务（setTimeout、setInterval、I/O）非常重要。

微任务在当前阶段完成后、下一个宏任务之前立即执行。如果微任务被无限生成，这可能导致事件循环“饥饿”。

// 示例：微任务阻塞宏任务\nsetTimeout(() => console.log('宏任务'), 0);\n\nPromise.resolve().then(() => {\n  // 无限微任务\n  process.nextTick(() => {\n    console.log('微任务');\n    // 递归调用将阻塞事件循环\n    process.nextTick(() => process.nextTick(() => {}));\n  });\n});\n// 输出：'微任务'（无限），'宏任务'永远不会执行

对于高级控制，请使用 queueMicrotask() 和 setImmediate()。请记住：process.nextTick 的优先级高于 Promise 微任务。

2. Worker Threads：Node.js 中的真正多线程

worker_threads 模块允许在并行线程中执行 JavaScript。每个 Worker 都有自己的 V8 实例和隔离的内存。数据通过消息或 SharedArrayBuffer 进行交换。

创建简单的 Worker：

// main.js\nconst { Worker } = require('worker_threads');\n\nfunction runWorker(file, data) {\n  return new Promise((resolve, reject) => {\n    const worker = new Worker(file, { workerData: data });\n    worker.on('message', resolve);\n    worker.on('error', reject);\n    worker.on('exit', (code) => {\n      if (code !== 0) reject(new Error(`Worker 以代码 ${code} 停止`));\n    });\n  });\n}\n\n// 启动\nrunWorker('./worker.js', { iterations: 1e9 })\n  .then(result => console.log('结果:', result))\n  .catch(err => console.error(err));

// worker.js\nconst { parentPort, workerData } = require('worker_threads');\n\nfunction heavyComputation(n) {\n  let sum = 0;\n  for (let i = 0; i < n; i++) { sum += Math.sqrt(i); }\n  return sum;\n}\n\nparentPort.postMessage(heavyComputation(workerData.iterations));

重要提示： Worker 无法访问 DOM、进程全局对象（process.env）以及主线程中加载的模块。仅传递可序列化的数据。

Worker 池

为每个任务创建 Worker 开销很大。请使用池：

class WorkerPool {\n  constructor(file, size = 4) {\n    this.workers = [];\n    this.queue = [];\n    for (let i = 0; i < size; i++) {\n      const worker = new Worker(file);\n      worker.on('message', (result) => this._handleResult(worker, result));\n      worker.on('error', (err) => this._handleError(worker, err));\n      this.workers.push(worker);\n    }\n  }\n\n  runTask(data) {\n    return new Promise((resolve, reject) => {\n      this.queue.push({ data, resolve, reject });\n      this._processQueue();\n    });\n  }\n\n  _processQueue() {\n    if (this.queue.length === 0) return;\n    const worker = this.workers.find(w => !w.busy);\n    if (!worker) return;\n    worker.busy = true;\n    const task = this.queue.shift();