Node.js Worker Threads 进阶：SharedArrayBuffer 与 Atomics 打造高性能多线程应用

2025/3/10 10:59:37 248 0 0 0

你好，我是老码农。今天我们来聊聊 Node.js 中 Worker Threads 的进阶应用，特别是如何在多个 Worker 之间共享数据（SharedArrayBuffer），以及如何避免数据竞争（Atomics）。如果你对多线程编程已经有一定了解，并且希望在 Node.js 中应用多线程技术来构建更复杂、更高效的应用程序，那么这篇文章绝对适合你。

为什么要使用 Worker Threads？

Node.js 以其单线程、非阻塞的特性而闻名，这使得它非常擅长处理 I/O 密集型任务，例如网络请求和数据库查询。然而，对于 CPU 密集型任务，例如图像处理、数据分析、加密解密等，单线程的特性可能会成为性能瓶颈。因为在执行 CPU 密集型任务时，Node.js 的事件循环会被阻塞，导致其他请求无法及时处理。为了解决这个问题，Node.js 引入了 Worker Threads。

Worker Threads 允许我们在 Node.js 中创建多个独立的 JavaScript 线程，每个线程都有自己的 V8 实例。这意味着我们可以将 CPU 密集型任务分配给 Worker 线程，从而避免阻塞主线程，提高应用程序的整体性能和响应速度。

Worker Threads 的基本用法回顾

在深入探讨 SharedArrayBuffer 和 Atomics 之前，我们先简单回顾一下 Worker Threads 的基本用法。

1. 创建 Worker

const { Worker } = require('worker_threads');

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

// 监听 worker 的消息
worker.on('message', (msg) => {
  console.log('Received message from worker:', msg);
});

// 监听 worker 退出事件
worker.on('exit', (code) => {
  console.log(`Worker stopped with exit code ${code}`);
});

// 向 worker 发送消息
worker.postMessage('Hello from main thread!');

2. worker.js 文件

const { parentPort } = require('worker_threads');

// 监听主线程的消息
parentPort.on('message', (msg) => {
  console.log('Received message from main thread:', msg);
  parentPort.postMessage('Hello from worker!');
});

3. 运行

node index.js

这个简单的例子演示了如何在主线程和 Worker 线程之间进行消息传递。主线程创建了一个 Worker，并向它发送一条消息。Worker 收到消息后，打印到控制台，并向主线程发送一条回复消息。主线程收到回复消息后，也打印到控制台。

共享内存的挑战：SharedArrayBuffer

在多线程编程中，共享数据是一个非常常见的需求。在 Node.js 的 Worker Threads 中，我们可以使用 SharedArrayBuffer 来实现多个 Worker 线程之间的数据共享。SharedArrayBuffer 允许我们在不同的线程之间共享一块内存区域，从而避免了通过消息传递来共享数据的开销，提高了数据共享的效率。

1. 什么是 SharedArrayBuffer？

SharedArrayBuffer 是一种特殊的 ArrayBuffer，它允许在不同的 JavaScript 线程之间共享一块内存区域。与普通的 ArrayBuffer 不同，SharedArrayBuffer 的内容可以在多个线程中被读写，而不需要进行数据的拷贝。这使得我们可以直接在不同的 Worker 线程中访问和修改相同的数据，从而实现高效的数据共享。

2. SharedArrayBuffer 的创建和使用

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');

if (isMainThread) {
  const sharedBuffer = new SharedArrayBuffer(16);
  const worker = new Worker(__filename, {
    workerData: { sharedBuffer: sharedBuffer },
  });

  worker.on('exit', (code) => {
    console.log(`Worker stopped with exit code ${code}`);
  });

  // 主线程修改共享内存
  const int32Array = new Int32Array(sharedBuffer);
  int32Array[0] = 10;
  int32Array[1] = 20;

} else {
  const sharedBuffer = workerData.sharedBuffer;

  // worker 线程访问共享内存
  const int32Array = new Int32Array(sharedBuffer);
  console.log('Worker thread: int32Array[0] =', int32Array[0]); // 输出: 10
  console.log('Worker thread: int32Array[1] =', int32Array[1]); // 输出: 20

  // worker 线程修改共享内存
  int32Array[0] = 30;
  int32Array[1] = 40;

  // 在修改之后，主线程也可以看到修改后的结果
  parentPort.postMessage('done');
}

在这个例子中，主线程创建了一个 SharedArrayBuffer，并将其传递给 Worker 线程。主线程和 Worker 线程都可以通过创建 Int32Array 来访问和修改 SharedArrayBuffer 中的数据。需要注意的是，为了安全起见，SharedArrayBuffer 必须在创建时指定大小，并且大小是固定的，不能动态调整。

3. 共享内存的注意事项

使用 SharedArrayBuffer 时，需要特别注意数据竞争的问题。由于多个线程可以同时访问和修改共享内存，如果不采取任何措施，就可能导致数据不一致。例如，一个线程正在读取某个变量的值，而另一个线程同时修改了这个变量的值，那么第一个线程读取到的值可能是不正确的。

避免数据竞争：Atomics

为了解决数据竞争的问题，我们可以使用 Atomics 对象。Atomics 对象提供了一组原子操作，可以保证对共享内存的读写操作是原子性的，即不可分割的。原子操作可以确保在同一时刻只有一个线程可以访问和修改共享内存，从而避免了数据竞争。

1. 什么是 Atomics？

Atomics 对象提供了一组静态方法，用于对 SharedArrayBuffer 中的数据进行原子操作。这些原子操作包括：

读操作： Atomics.load()，从共享内存中读取一个值。
写操作： Atomics.store()，将一个值写入共享内存。
比较并交换操作： Atomics.compareExchange()，如果共享内存中的值等于预期值，则将其修改为新值，否则不进行任何操作。这个操作可以用来实现乐观锁。
加法操作： Atomics.add()，将共享内存中的值加上一个给定的值。
减法操作： Atomics.sub()，将共享内存中的值减去一个给定的值。
与操作： Atomics.and()，将共享内存中的值与一个给定的值进行与运算。
或操作： Atomics.or()，将共享内存中的值与一个给定的值进行或运算。
异或操作： Atomics.xor()，将共享内存中的值与一个给定的值进行异或运算。
交换操作： Atomics.exchange()，将共享内存中的值替换为一个新的值，并返回旧值。
等待和唤醒操作： Atomics.wait() 和 Atomics.notify()，用于实现线程间的同步。当一个线程调用 Atomics.wait() 时，它会阻塞当前线程的执行，直到另一个线程调用 Atomics.notify() 唤醒它。

2. Atomics 的使用示例

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');

if (isMainThread) {
  const sharedBuffer = new SharedArrayBuffer(4);
  const int32Array = new Int32Array(sharedBuffer);
  int32Array[0] = 0;

  const worker = new Worker(__filename, {
    workerData: { sharedBuffer: sharedBuffer },
  });

  worker.on('exit', (code) => {
    console.log(`Worker stopped with exit code ${code}`);
  });

  // 启动另一个 worker 线程
  const worker2 = new Worker(__filename, {
    workerData: { sharedBuffer: sharedBuffer },
  });

  worker2.on('exit', (code) => {
    console.log(`Worker2 stopped with exit code ${code}`);
  });


  // 主线程读取共享内存
  setTimeout(() => {
      console.log('Main thread: int32Array[0] =', int32Array[0]);
  }, 1000)

} else {
  const sharedBuffer = workerData.sharedBuffer;
  const int32Array = new Int32Array(sharedBuffer);

  // 使用 Atomics.add 进行原子加法操作
  for (let i = 0; i < 100000; i++) {
    Atomics.add(int32Array, 0, 1);
  }

  // 告诉主线程 worker 完成了任务
  parentPort.postMessage('done');
}

在这个例子中，主线程创建了一个 SharedArrayBuffer，并将其传递给两个 Worker 线程。每个 Worker 线程都会对 SharedArrayBuffer 中的一个整数进行 100000 次原子加法操作。由于使用了 Atomics.add，可以确保加法操作是原子性的，从而避免了数据竞争。

3. Atomics 的高级应用：线程同步

除了避免数据竞争之外，Atomics 还可以用于实现线程间的同步。Atomics.wait() 和 Atomics.notify() 是一对用于线程同步的原子操作。

Atomics.wait(typedArray, index, value, timeout)：当指定位置的值等于 value 时，当前线程进入休眠状态，直到被 Atomics.notify() 唤醒或超时。如果 timeout 为 0，则立即返回。如果指定位置的值不等于 value，则立即返回。
Atomics.notify(typedArray, index, count)：唤醒在指定位置等待的线程。count 参数指定要唤醒的线程数量。如果指定位置没有线程等待，则该操作无任何效果。

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');

if (isMainThread) {
  const sharedBuffer = new SharedArrayBuffer(4);
  const int32Array = new Int32Array(sharedBuffer);
  int32Array[0] = 0;

  const worker = new Worker(__filename, {
    workerData: { sharedBuffer: sharedBuffer },
  });

  worker.on('exit', (code) => {
    console.log(`Worker stopped with exit code ${code}`);
  });

  worker.on('message', (msg) => {
    if (msg === 'ready') {
      // 主线程等待 worker 线程将 int32Array[0] 设置为 1
      Atomics.wait(int32Array, 0, 0);
      console.log('Main thread: int32Array[0] =', int32Array[0]); // 输出: 1
    }
  });


} else {
  const sharedBuffer = workerData.sharedBuffer;
  const int32Array = new Int32Array(sharedBuffer);

  // 通知主线程 worker 线程已经准备好
  parentPort.postMessage('ready');

  // 模拟一些耗时操作
  setTimeout(() => {
    int32Array[0] = 1;
    Atomics.notify(int32Array, 0, 1);
  }, 2000);
}

在这个例子中，主线程创建了一个 Worker 线程，并使用 Atomics.wait() 等待 Worker 线程将 int32Array[0] 设置为 1。Worker 线程在执行完一些耗时操作后，将 int32Array[0] 设置为 1，并使用 Atomics.notify() 唤醒主线程。通过这种方式，我们可以实现线程间的同步，确保数据的正确性和一致性。

实际案例：并发计算斐波那契数列

为了更好地理解 SharedArrayBuffer 和 Atomics 的应用，我们来看一个实际的案例：并发计算斐波那契数列。

1. 问题描述

斐波那契数列是一个经典的数学问题，它的定义如下：

F(0) = 0
F(1) = 1
F(n) = F(n-1) + F(n-2) (n >= 2)

我们的目标是使用多线程并发计算斐波那契数列，并比较单线程和多线程的性能差异。

2. 实现思路

我们可以将计算斐波那契数列的任务分解成多个子任务，每个子任务计算一部分斐波那契数。然后，使用多个 Worker 线程并发执行这些子任务。为了共享计算结果，我们可以使用 SharedArrayBuffer 来存储斐波那契数列的计算结果。为了避免数据竞争，我们需要使用 Atomics 来确保对 SharedArrayBuffer 的读写操作是原子性的。

3. 代码实现

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');

// 计算斐波那契数列
function fibonacci(n) {
  if (n <= 1) {
    return n;
  }
  return fibonacci(n - 1) + fibonacci(n - 2);
}

if (isMainThread) {
  const numWorkers = 4; // 线程数量
  const n = 35; // 计算到第 n 个斐波那契数
  const sharedBuffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT * (n + 1));
  const fibArray = new Int32Array(sharedBuffer);

  // 初始化 SharedArrayBuffer
  fibArray[0] = 0;
  fibArray[1] = 1;

  const workers = [];
  const startTime = performance.now();

  // 创建 worker 线程
  for (let i = 0; i < numWorkers; i++) {
    const worker = new Worker(__filename, {
      workerData: {
        start: 2 + Math.floor((n - 2) / numWorkers) * i,
        end: 2 + Math.floor((n - 2) / numWorkers) * (i + 1) - 1,
        sharedBuffer: sharedBuffer,
      },
    });
    workers.push(worker);

    worker.on('exit', (code) => {
      console.log(`Worker ${worker.threadId} stopped with exit code ${code}`);
    });

  }

  // 监听 worker 线程完成事件
  let completedWorkers = 0;
  for (const worker of workers) {
    worker.on('message', (msg) => {
      if (msg === 'done') {
        completedWorkers++;
        if (completedWorkers === numWorkers) {
          const endTime = performance.now();
          console.log(`多线程计算斐波那契数列(${n})耗时: ${endTime - startTime}ms`);
          console.log('结果:', fibArray[n]);
        }
      }
    });
  }

} else {
  const { start, end, sharedBuffer } = workerData;
  const fibArray = new Int32Array(sharedBuffer);

  // worker 线程计算斐波那契数列
  for (let i = start; i <= end; i++) {
    fibArray[i] = fibonacci(i);
  }

  // 通知主线程 worker 完成了任务
  parentPort.postMessage('done');
}

4. 运行结果

运行上述代码，我们可以看到多线程计算斐波那契数列的耗时明显低于单线程的耗时。当然，斐波那契数列的计算本身就比较适合并行化，因为它具有递归的特性。对于其他类型的计算任务，多线程的性能提升可能不如斐波那契数列那么明显。

总结与最佳实践

通过本文，我希望你对 Node.js Worker Threads 的进阶应用，特别是 SharedArrayBuffer 和 Atomics，有了更深入的理解。以下是一些总结和最佳实践：

选择合适的场景： Worker Threads 适用于 CPU 密集型任务，例如图像处理、数据分析、加密解密等。对于 I/O 密集型任务，Node.js 的单线程、非阻塞的特性通常就足够了。
合理分配任务： 将 CPU 密集型任务分解成多个子任务，并分配给不同的 Worker 线程。尽量使每个 Worker 线程的任务量均衡，以避免出现性能瓶颈。
使用 SharedArrayBuffer 进行数据共享： 如果需要在多个 Worker 线程之间共享数据，可以使用 SharedArrayBuffer。这样可以避免通过消息传递来共享数据的开销，提高数据共享的效率。
使用 Atomics 避免数据竞争： 当多个 Worker 线程同时访问和修改共享内存时，一定要使用 Atomics 对象来保证操作的原子性，从而避免数据竞争。
考虑线程同步： 如果需要在不同的 Worker 线程之间进行同步，可以使用 Atomics.wait() 和 Atomics.notify() 等原子操作来实现线程同步。
监控和调试： 在使用 Worker Threads 时，需要注意监控应用程序的性能，并使用调试工具来诊断和解决可能出现的问题。
避免过度使用： 虽然 Worker Threads 可以提高应用程序的性能，但是过度使用可能会导致资源消耗过大，甚至降低应用程序的性能。因此，在使用 Worker Threads 时，需要权衡其带来的收益和成本。

结语

希望这篇文章能够帮助你更好地理解 Node.js Worker Threads 的进阶应用。通过 SharedArrayBuffer 和 Atomics，你可以构建更复杂、更高效的多线程 Node.js 应用程序。多线程编程是一个复杂而有趣的话题，如果你想深入学习，可以参考 Node.js 官方文档和其他相关资料。祝你在 Node.js 多线程编程的道路上越走越远！

老码农的后院 Node.js Worker Threads SharedArrayBuffer Atomics