利用 Atomics.wait 与 notify 实现跨 Worker 任务分发：构建极低开销的生产者消费者模型

各位同仁，下午好！

今天，我们将深入探讨一个在现代Web应用开发中至关重要的话题：如何利用JavaScript的Atomics.wait和Atomics.notify机制，实现一个跨Worker的、极致低开销的生产者消费者模型。在日益复杂的Web应用场景中，性能瓶颈往往出现在主线程的计算密集型任务或高频度的I/O操作上。Web Workers的出现，使我们将这些任务卸载到后台线程成为可能，从而保持主线程的响应性和用户界面的流畅。

然而，Worker之间的通信，尤其是高频、低延迟的任务分发，传统上是一个挑战。postMessage虽然方便，但在消息量大、结构复杂时，其序列化/反序列化的开销不容忽视。我们需要的，是一种更接近操作系统底层、基于共享内存的同步与通信原语——这正是SharedArrayBuffer结合Atomics所提供的能力。

我们将构建一个生产者-消费者模型，其中一个或多个Worker作为生产者，负责生成任务；一个或多个Worker作为消费者，负责执行任务。核心思想是利用SharedArrayBuffer作为共享任务队列，并通过Atomics.wait和Atomics.notify在生产者和消费者之间进行高效的信号传递，避免忙等待（busy-waiting），从而实现近乎零CPU占用的等待机制。

1. 跨Worker通信的挑战与Atomics的登场

1.1 传统Web Worker通信的局限

在深入Atomics之前，我们先快速回顾一下Web Worker的通信方式及其局限性。

• postMessage API:

  • 优点: 简单易用，支持结构化克隆算法，可以传递各种JavaScript对象（包括ArrayBuffer、MessagePort等）。
  • 缺点: 每次调用都会涉及数据的序列化（发送方）和反序列化（接收方）。对于大型数据或高频通信，这会带来显著的性能开销。数据是复制而非共享，意味着内存使用量增加。
  • 适用场景: 低频次、非性能敏感的数据交换，或传递复杂对象。

• WebSocket/IndexedDB等: 这些通常用于持久化存储或与服务器通信，不直接用于Worker间的高效任务分发。

当我们需要在多个Worker之间共享大量数据或进行高频、低延迟的同步时，postMessage的局限性就变得非常明显。例如，一个视频处理应用可能需要主线程将视频帧数据传递给Worker进行处理，处理完成后再将结果传回。如果每一帧都通过postMessage传递，性能会急剧下降。

1.2 SharedArrayBuffer与Atomics：共享内存的革命

为了解决上述问题，ECMAScript 2017引入了SharedArrayBuffer和Atomics对象，它们是Web Worker通信领域的一场革命。

• SharedArrayBuffer (SAB):

  • 定义: 一种特殊类型的ArrayBuffer，其内存可以在多个JavaScript执行上下文（如主线程和Web Workers）之间共享。这意味着，不同Worker可以直接读写同一块内存区域，而无需进行数据复制。
  • 优势: 极大地降低了数据传输开销，实现了真正的共享内存并发。
  • 安全性考量: 由于共享内存引入了新的安全风险（如Spectre和Meltdown旁道攻击），浏览器对SharedArrayBuffer的使用施加了严格的限制。它要求页面必须通过Cross-Origin-Opener-Policy (COOP) 设置为 same-origin 或 same-origin-allow-popups，以及 Cross-Origin-Embedder-Policy (COEP) 设置为 require-corp。这是保障Web应用安全的重要前提。

• Atomics 对象:

  • 定义: Atomics对象提供了一组静态方法，用于对SharedArrayBuffer视图（如Int32Array）执行原子操作。原子操作是指在执行过程中不会被其他线程中断的操作，保证了操作的完整性和数据的一致性。
  • 核心作用:
    1. 原子读写和修改: Atomics.load(), Atomics.store(), Atomics.add(), Atomics.sub(), Atomics.and(), Atomics.or(), Atomics.xor(), Atomics.exchange(), Atomics.compareExchange() 等。这些方法保证了在多线程环境下对共享内存中特定位置的读写或修改操作是原子的，避免了数据损坏。
    2. 线程同步 (wait 和 notify): 这是我们本次讲座的重点。Atomics.wait() 和 Atomics.notify() 提供了一种高效的线程同步机制，允许一个线程在某个共享内存位置等待，直到另一个线程在该位置上发出通知。这避免了传统的忙等待（不断轮询检查条件）造成的CPU资源浪费。

1.3 Atomics.wait 和 Atomics.notify 详解

这两个方法是构建低开销生产者消费者模型的基石。

• Atomics.wait(typedArray, index, value, [timeout])

  • typedArray: 必须是 Int32Array 或 BigInt64Array 类型的SharedArrayBuffer视图。这是进行等待操作的内存区域。
  • index: typedArray中要等待的元素的索引。
  • value: 期望该索引处的值。如果 typedArray[index] 的当前值与 value 不匹配，wait操作会立即返回 'not-equal'，表示条件不满足，无需等待。如果匹配，当前线程将进入休眠状态，直到被 notify 唤醒或超时。
  • timeout (可选): 以毫秒为单位的等待时间。如果设置为 Infinity 或省略，则无限期等待。如果超时，wait操作返回 'timed-out'。
  • 返回值: 'ok' (被唤醒), 'not-equal' (初始值不匹配), 'timed-out' (超时)。

  Atomics.wait 的核心思想是，它允许线程在共享内存的特定位置“停车”（park），释放CPU资源，而不是持续检查条件。当条件满足时（通过notify），它才会被“唤醒”（unpark）。

• Atomics.notify(typedArray, index, [count])

  • typedArray: 必须是 Int32Array 或 BigInt64Array 类型的SharedArrayBuffer视图。
  • index: typedArray中要发出通知的元素的索引。
  • count (可选): 要唤醒的等待线程的数量。默认为 Infinity，表示唤醒所有在该索引处等待的线程。
  • 返回值: 实际被唤醒的线程数量。

  Atomics.notify 用于唤醒一个或多个在指定索引处等待的线程。它通常在共享状态发生变化，使得等待条件可能满足时被调用。

通过 Atomics.wait 和 Atomics.notify，我们可以实现经典的条件变量（condition variable）模式，这是构建高效并发结构的关键。

2. 生产者消费者模型设计

我们的目标是构建一个基于共享内存的循环缓冲区（ring buffer）作为任务队列，并利用Atomics进行同步。

2.1 核心组件与共享内存布局

为了实现这个模型，我们需要两个SharedArrayBuffer：

1. 控制缓冲区 (Control Buffer): 存储队列的元数据，如读写指针、当前任务数量等。这个缓冲区将主要由Int32Array视图访问，因为Atomics方法要求Int32Array或BigInt64Array。
2. 任务数据缓冲区 (Task Data Buffer): 存储实际的任务数据。这个缓冲区通常由Uint8Array视图访问，因为它以字节形式存储任意数据。

共享内存布局（Control Buffer）:

我们将controlBuffer设计为Int32Array，并为每个同步点和元数据定义常量索引，以提高可读性和维护性。

索引常量	作用	数据类型 (对应Int32Array)	初始值
CONTROL_HEAD_INDEX	消费者读取任务的起始位置（环形队列的头部）	Int32	0
CONTROL_TAIL_INDEX	生产者写入任务的起始位置（环形队列的尾部）	Int32	0
CONTROL_COUNT_INDEX	当前队列中任务的数量	Int32	0
CONTROL_QUEUE_SIZE	队列的最大容量（以字节为单位，而非任务数量）	Int32	N
CONTROL_PRODUCER_LOCK	生产者锁（用于序列化生产者对TAIL和写入区域的访问，尽管我们的设计可以通过COUNT减少锁的需求）	Int32	0
CONTROL_CONSUMER_LOCK	消费者锁（用于序列化消费者对HEAD和读取区域的访问）	Int32	0
CONTROL_MSG_LEN_INDEX	存储当前待处理消息的长度，用于消费者读取	Int32	0

重要说明：

• CONTROL_QUEUE_SIZE 存储的是任务数据缓冲区的 总字节容量，而不是任务数量。因为任务大小不固定，我们需要按字节管理空间。
• CONTROL_HEAD_INDEX 和 CONTROL_TAIL_INDEX 同样是以字节为单位的偏移量。
• CONTROL_COUNT_INDEX 存储的是 当前队列中实际的任务数量。这个是生产者/消费者等待/通知的核心条件变量。
• CONTROL_PRODUCER_LOCK 和 CONTROL_CONSUMER_LOCK 可以用于更复杂的互斥场景。但在我们的环形队列实现中，生产者只修改TAIL，消费者只修改HEAD，这两个操作本身没有直接冲突。冲突主要发生在队列满/空时的COUNT更新及等待。因此，我们可以简化，主要依靠CONTROL_COUNT_INDEX进行同步。

2.2 循环缓冲区（Ring Buffer）机制

循环缓冲区是一种固定大小的数据结构，它将内存视为一个环。当读写指针到达缓冲区的末尾时，它们会“环绕”到缓冲区的开头。

• 写入 (Enqueue):
  • 生产者向TAIL位置写入数据。
  • 写入后，TAIL指针前进。如果TAIL超过了缓冲区末尾，它会回到开头（TAIL = (TAIL + length) % capacity）。
  • 每次写入任务，CONTROL_COUNT_INDEX原子递增。
• 读取 (Dequeue):
  • 消费者从HEAD位置读取数据。
  • 读取后，HEAD指针前进。如果HEAD超过了缓冲区末尾，它会回到开头（HEAD = (HEAD + length) % capacity）。
  • 每次读取任务，CONTROL_COUNT_INDEX原子递减。

2.3 Atomics同步策略

这是实现低开销的关键。

1. 生产者等待（队列满）:

   • 当生产者尝试添加任务时，如果 CONTROL_COUNT_INDEX 已经达到 CONTROL_QUEUE_SIZE（表示队列已满，无法再添加任务），生产者需要等待。
   • 策略: 生产者调用 Atomics.wait(controlArray, CONTROL_COUNT_INDEX, controlArray[CONTROL_COUNT_INDEX])。这会使其休眠，直到CONTROL_COUNT_INDEX的值发生变化（即消费者取走任务）。
   • 何时唤醒: 消费者在成功取走任务后，会递减 CONTROL_COUNT_INDEX 并调用 Atomics.notify(controlArray, CONTROL_COUNT_INDEX, 1) 唤醒一个等待的生产者。

2. 消费者等待（队列空）:

   • 当消费者尝试取走任务时，如果 CONTROL_COUNT_INDEX 为 0（表示队列为空），消费者需要等待。
   • 策略: 消费者调用 Atomics.wait(controlArray, CONTROL_COUNT_INDEX, 0)。这会使其休眠，直到CONTROL_COUNT_INDEX的值变为非零（即生产者添加了任务）。
   • 何时唤醒: 生产者在成功添加任务后，会递增 CONTROL_COUNT_INDEX 并调用 Atomics.notify(controlArray, CONTROL_COUNT_INDEX, 1) 唤醒一个等待的消费者。

3. 任务长度传递: 由于任务大小不固定，我们需要在每个任务数据前预留几个字节来存储该任务的长度。这样，消费者就知道要读取多少字节。

3. 实现细节与代码示例

我们将构建一个SharedRingBuffer类来封装所有共享内存操作，并在主线程和Worker中实例化和使用它。

3.1 跨域隔离设置 (COOP / COEP)

为了使用SharedArrayBuffer，你的Web服务器必须发送特定的HTTP头。

# Nginx 配置示例
add_header Cross-Origin-Opener-Policy "same-origin";
add_header Cross-Origin-Embedder-Policy "require-corp";

或者在开发时使用带有这些头的本地服务器（如http-server工具或Webpack Dev Server配置）。

3.2 共享内存结构定义

首先，定义控制缓冲区所需的常量和索引。

// shared_constants.js (或直接内联到需要的地方)

// Control Buffer 的大小 (以 Int32 单元为准)
const CONTROL_BUFFER_SIZE_INT32 = 7; // 足够的空间给所有索引

// 控制缓冲区中的索引定义
const CONTROL_HEAD_INDEX = 0;        // 消费者读取任务的起始位置 (字节偏移)
const CONTROL_TAIL_INDEX = 1;        // 生产者写入任务的起始位置 (字节偏移)
const CONTROL_COUNT_INDEX = 2;       // 队列中当前任务的数量
const CONTROL_CAPACITY_INDEX = 3;    // 任务数据缓冲区的总字节容量
const CONTROL_PRODUCER_LOCK = 4;     // 生产者互斥锁（备用，简化版可能不需要）
const CONTROL_CONSUMER_LOCK = 5;     // 消费者互斥锁（备用，简化版可能不需要）
const CONTROL_MSG_LEN_INDEX = 6;     // 临时存储当前消息长度，用于生产者写入和消费者读取

// 任务数据前缀：存储消息长度的字节数 (例如，一个 4 字节的 Int32)
const MESSAGE_LENGTH_PREFIX_BYTES = 4; 

// 导出这些常量，以便在主线程和 Worker 中共享
export {
    CONTROL_BUFFER_SIZE_INT32,
    CONTROL_HEAD_INDEX,
    CONTROL_TAIL_INDEX,
    CONTROL_COUNT_INDEX,
    CONTROL_CAPACITY_INDEX,
    CONTROL_PRODUCER_LOCK,
    CONTROL_CONSUMER_LOCK,
    CONTROL_MSG_LEN_INDEX,
    MESSAGE_LENGTH_PREFIX_BYTES
};

3.3 SharedRingBuffer 类

这个类将封装所有对SharedArrayBuffer的读写和Atomics同步逻辑。

// SharedRingBuffer.js
import {
    CONTROL_HEAD_INDEX,
    CONTROL_TAIL_INDEX,
    CONTROL_COUNT_INDEX,
    CONTROL_CAPACITY_INDEX,
    CONTROL_MSG_LEN_INDEX,
    MESSAGE_LENGTH_PREFIX_BYTES
} from './shared_constants.js';

class SharedRingBuffer {
    /**
     * @param {Int32Array} controlArray - SharedArrayBuffer 的 Int32Array 视图，用于控制信息。
     * @param {Uint8Array} dataBufferView - SharedArrayBuffer 的 Uint8Array 视图，用于存储任务数据。
     */
    constructor(controlArray, dataBufferView) {
        if (!(controlArray instanceof Int32Array && controlArray.buffer instanceof SharedArrayBuffer)) {
            throw new Error('controlArray must be an Int32Array view of a SharedArrayBuffer.');
        }
        if (!(dataBufferView instanceof Uint8Array && dataBufferView.buffer instanceof SharedArrayBuffer)) {
            throw new Error('dataBufferView must be a Uint8Array view of a SharedArrayBuffer.');
        }

        this.controlArray = controlArray;
        this.dataBufferView = dataBufferView;
        this.dataCapacityBytes = Atomics.load(this.controlArray, CONTROL_CAPACITY_INDEX);

        // TextEncoder/Decoder 用于将字符串转换为字节数组，反之亦然
        this.textEncoder = new TextEncoder();
        this.textDecoder = new TextDecoder();

        console.log(`[SharedRingBuffer] Initialized with capacity: ${this.dataCapacityBytes} bytes.`);
        console.log(`[SharedRingBuffer] Current head: ${Atomics.load(this.controlArray, CONTROL_HEAD_INDEX)}, ` +
                    `tail: ${Atomics.load(this.controlArray, CONTROL_TAIL_INDEX)}, ` +
                    `count: ${Atomics.load(this.controlArray, CONTROL_COUNT_INDEX)} tasks.`);
    }

    /**
     * 生产者：将任务数据写入队列。
     * @param {any} taskData - 要写入的任务数据（会被JSON序列化）。
     * @returns {boolean} - true if enqueued, false if an error occurred (e.g., serialization error).
     */
    enqueueTask(taskData) {
        let serializedTask;
        try {
            // 1. 序列化任务数据
            const jsonString = JSON.stringify(taskData);
            serializedTask = this.textEncoder.encode(jsonString); // 转换为 Uint8Array
        } catch (e) {
            console.error('Error serializing task data:', e);
            return false;
        }

        const taskSizeBytes = serializedTask.length;
        const totalSizeNeeded = MESSAGE_LENGTH_PREFIX_BYTES + taskSizeBytes;

        if (totalSizeNeeded > this.dataCapacityBytes) {
            console.error(`Task too large (${totalSizeNeeded} bytes) for buffer capacity (${this.dataCapacityBytes} bytes).`);
            return false;
        }

        let currentCount = Atomics.load(this.controlArray, CONTROL_COUNT_INDEX);
        let head = Atomics.load(this.controlArray, CONTROL_HEAD_INDEX);
        let tail = Atomics.load(this.controlArray, CONTROL_TAIL_INDEX);

        // --- 生产者等待逻辑：如果队列已满，则等待 ---
        // 队列满的判断：当前任务数量 *单个任务平均大小* 接近总容量，或者说，剩余空间不足以容纳当前任务。
        // 更准确的判断是计算实际的可用字节空间。
        // 简化判断：如果 (tail + totalSizeNeeded) % capacity == head 并且 count > 0, 则队列满。
        // 或者，我们可以用一个更简单的策略：如果队列中的任务数量达到某个阈值（例如，控制缓冲区中存储的最大任务数），则等待。
        // 鉴于任务大小可变，直接用 `CONTROL_COUNT_INDEX` 来表示“当前队列中有多少个 *逻辑任务*”是更常见且有效的策略。
        // 这里的 `CONTROL_COUNT_INDEX` 不代表字节数，而是代表任务个数。
        // 我们需要一个逻辑上的最大任务数来判断队列是否“满”。
        // 假设 max_tasks = dataCapacityBytes / MESSAGE_LENGTH_PREFIX_BYTES (一个粗略估计)
        const MAX_LOGICAL_TASKS = Math.floor(this.dataCapacityBytes / (MESSAGE_LENGTH_PREFIX_BYTES + 10)); // 假设平均任务大小10字节

        while (Atomics.load(this.controlArray, CONTROL_COUNT_INDEX) >= MAX_LOGICAL_TASKS) {
            console.log(`[Producer] Queue full (${Atomics.load(this.controlArray, CONTROL_COUNT_INDEX)} tasks), waiting...`);
            // 等待 CONTROL_COUNT_INDEX 发生变化 (即消费者取走了任务)
            // 这里 value 是当前值，如果值不变，则等待
            Atomics.wait(this.controlArray, CONTROL_COUNT_INDEX, Atomics.load(this.controlArray, CONTROL_COUNT_INDEX));
        }

        // --- 写入任务数据 ---
        // 1. 获取当前的写入位置 (tail)
        tail = Atomics.load(this.controlArray, CONTROL_TAIL_INDEX);

        // 2. 检查是否有足够的连续空间，如果不足，则需要环绕
        let availableSpaceToEnd = this.dataCapacityBytes - tail;
        let wrapAround = false;

        if (availableSpaceToEnd < totalSizeNeeded) {
            // 需要环绕。首先，如果头部在尾部之前，那么队列中还有数据，不能直接环绕覆盖
            // 这是一个复杂的判断，简化处理：如果需要环绕，且环绕后会覆盖head，则等待
            // 更稳健的做法是：如果队列满，已经在上面的 `while` 循环中等待了。
            // 这里我们假设 `MAX_LOGICAL_TASKS` 已经保证了有足够的空间来写入。
            // 实际的环形缓冲区需要更精细的逻辑来处理碎片化和空间分配。
            // 为了简化，我们假设任务数据是连续写入的，并在必要时从头开始。

            // 为了演示 `Atomics`，我们先采用一个简化模型：
            // 如果剩余空间不足以写入当前任务，并且从 `0` 开始写入能够容纳，
            // 并且 `head` 在 `tail` 之后（即队列未被环绕），那么我们可以环绕。
            // 否则，需要等待。
            if (totalSizeNeeded <= head) { // 检查从0开始到head的空间是否足够
                console.log(`[Producer] Wrapping around. Current tail: ${tail}, head: ${head}, capacity: ${this.dataCapacityBytes}`);
                tail = 0; // 重置 tail 到缓冲区开头
                Atomics.store(this.controlArray, CONTROL_TAIL_INDEX, tail); // 原子更新 tail
                wrapAround = true;
            } else {
                // 即使环绕，空间也不够（或者会覆盖消费者未读的数据），这不应该发生如果MAX_LOGICAL_TASKS设置合理。
                // 这是一个需要更复杂空间管理或更严格队列满判断的场景。
                console.error('[Producer] Not enough contiguous space for task, even with wrap-around consideration. This indicates potential logic error or insufficient queue size/logic.');
                return false;
            }
        }

        // 3. 写入消息长度前缀
        // 这里需要一个 DataView 来写入 Int32
        const dataView = new DataView(this.dataBufferView.buffer);
        dataView.setInt32(tail, taskSizeBytes, true); // true for little-endian
        tail += MESSAGE_LENGTH_PREFIX_BYTES;

        // 4. 写入实际任务数据
        this.dataBufferView.set(serializedTask, tail);
        tail += taskSizeBytes;

        // 5. 更新 tail 指针 (原子操作)
        Atomics.store(this.controlArray, CONTROL_TAIL_INDEX, tail % this.dataCapacityBytes);

        // 6. 原子递增任务计数器
        const newCount = Atomics.add(this.controlArray, CONTROL_COUNT_INDEX, 1);
        console.log(`[Producer] Enqueued task. New count: ${newCount}. Tail: ${Atomics.load(this.controlArray, CONTROL_TAIL_INDEX)}`);

        // 7. 通知消费者可能有新任务
        Atomics.notify(this.controlArray, CONTROL_COUNT_INDEX, 1); // 唤醒一个等待的消费者
        return true;
    }

    /**
     * 消费者：从队列中读取任务数据。
     * @returns {any | null} - 读取到的任务数据，或 null 如果队列为空。
     */
    dequeueTask() {
        // --- 消费者等待逻辑：如果队列为空，则等待 ---
        while (Atomics.load(this.controlArray, CONTROL_COUNT_INDEX) === 0) {
            console.log(`[Consumer] Queue empty, waiting...`);
            // 等待 CONTROL_COUNT_INDEX 发生变化 (即生产者添加了任务)
            // 这里 value 是 0，如果值仍为 0，则等待
            Atomics.wait(this.controlArray, CONTROL_COUNT_INDEX, 0);
        }

        // --- 读取任务数据 ---
        let head = Atomics.load(this.controlArray, CONTROL_HEAD_INDEX);
        const dataView = new DataView(this.dataBufferView.buffer);

        // 1. 读取消息长度前缀
        const taskSizeBytes = dataView.getInt32(head, true); // true for little-endian
        head += MESSAGE_LENGTH_PREFIX_BYTES;

        // 2. 检查是否有足够的连续空间来读取整个任务
        let availableSpaceToEnd = this.dataCapacityBytes - head;
        let serializedTaskBytes;

        if (availableSpaceToEnd < taskSizeBytes) {
            // 需要环绕读取
            // 这是一个简化的环绕读取逻辑，实际可能需要分两次拷贝
            // 例如：先拷贝到末尾，再从0开始拷贝剩余部分
            serializedTaskBytes = new Uint8Array(taskSizeBytes);
            const firstPartLength = availableSpaceToEnd;
            const secondPartLength = taskSizeBytes - firstPartLength;

            serializedTaskBytes.set(this.dataBufferView.subarray(head, head + firstPartLength));
            serializedTaskBytes.set(this.dataBufferView.subarray(0, secondPartLength), firstPartLength);

            head = secondPartLength; // 新的 head 指向环绕后的位置
        } else {
            // 连续读取
            serializedTaskBytes = this.dataBufferView.subarray(head, head + taskSizeBytes);
            head += taskSizeBytes;
        }

        // 3. 更新 head 指针 (原子操作)
        Atomics.store(this.controlArray, CONTROL_HEAD_INDEX, head % this.dataCapacityBytes);

        // 4. 原子递减任务计数器
        const newCount = Atomics.sub(this.controlArray, CONTROL_COUNT_INDEX, 1);
        console.log(`[Consumer] Dequeued task. New count: ${newCount}. Head: ${Atomics.load(this.controlArray, CONTROL_HEAD_INDEX)}`);

        // 5. 通知生产者可能有新空间
        Atomics.notify(this.controlArray, CONTROL_COUNT_INDEX, 1); // 唤醒一个等待的生产者

        // 6. 反序列化任务数据
        try {
            const jsonString = this.textDecoder.decode(serializedTaskBytes);
            return JSON.parse(jsonString);
        } catch (e) {
            console.error('Error deserializing task data:', e);
            return null;
        }
    }

    /**
     * 获取当前队列中的任务数量。
     * @returns {number}
     */
    getTaskCount() {
        return Atomics.load(this.controlArray, CONTROL_COUNT_INDEX);
    }
}

export default SharedRingBuffer;

关于 MAX_LOGICAL_TASKS 的说明：
在上述 enqueueTask 方法中，我使用了一个 MAX_LOGICAL_TASKS 来判断队列是否“满”。这是一种简化，因为环形缓冲区实际的“满”和“空”判断需要考虑 head 和 tail 指针的相对位置以及是否绕回。
一个更精确的环形缓冲区满/空判断通常是：

• 空: head === tail && count === 0
• 满: head === tail && count > 0 (这意味着 tail 绕了一圈追上了 head)
  或者更常见的是，留一个空位：((tail + 1) % capacity) === head。
  然而，由于我们处理的是变长消息，且CONTROL_COUNT_INDEX存储的是 任务个数 而非字节数，最直接且安全的判断就是设定一个最大任务数。如果需要精确到字节级别的空间管理，则需要更复杂的逻辑来跟踪可用连续空间。这里的实现选择了简化，通过 MAX_LOGICAL_TASKS 和 CONTROL_COUNT_INDEX 进行同步，它足够演示 Atomics.wait/notify 的核心机制。

3.4 主线程 (index.html 和 main.js)

主线程负责初始化SharedArrayBuffer，创建Worker，并将SABs传递给它们。

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Atomics Cross-Worker Task Distribution</title>
    <!-- 必须设置 COOP 和 COEP 头，否则 SharedArrayBuffer 不可用 -->
    <!-- 这些头通常由服务器发送，但在这里为了演示，我们假设它们已设置 -->
    <!-- <meta http-equiv="Cross-Origin-Opener-Policy" content="same-origin"> -->
    <!-- <meta http-equiv="Cross-Origin-Embedder-Policy" content="require-corp"> -->
</head>
<body>
    <h1>Atomics Cross-Worker Task Distribution (Producer-Consumer)</h1>
    <p>Check console for output.</p>
    <button id="startWorkers">Start Workers & Produce Tasks</button>
    <button id="stopWorkers">Stop Workers</button>

    <script type="module" src="main.js"></script>
</body>
</html>

// main.js
import {
    CONTROL_BUFFER_SIZE_INT32,
    CONTROL_CAPACITY_INDEX,
    MESSAGE_LENGTH_PREFIX_BYTES
} from './shared_constants.js';
import SharedRingBuffer from './SharedRingBuffer.js';

// --- 配置参数 ---
const NUM_CONSUMER_WORKERS = 2;
const DATA_BUFFER_CAPACITY_BYTES = 1024 * 1024; // 1MB for task data
const INITIAL_TASKS_TO_PRODUCE = 20;

let controlBuffer = null;
let dataBuffer = null;
let controlArray = null;
let dataBufferView = null;
let producerRingBuffer = null;
let consumerWorkers = [];
let workerIdCounter = 0;
let isProducing = false;

document.getElementById('startWorkers').addEventListener('click', startSystem);
document.getElementById('stopWorkers').addEventListener('click', stopSystem);

async function startSystem() {
    if (isProducing) {
        console.warn('System already running.');
        return;
    }
    isProducing = true;

    console.log('--- Initializing SharedArrayBuffer and Workers ---');

    // 1. 创建 SharedArrayBuffer
    controlBuffer = new SharedArrayBuffer(CONTROL_BUFFER_SIZE_INT32 * Int32Array.BYTES_PER_ELEMENT);
    dataBuffer = new SharedArrayBuffer(DATA_BUFFER_CAPACITY_BYTES);

    // 2. 创建 Int32Array 视图并初始化控制信息
    controlArray = new Int32Array(controlBuffer);
    dataBufferView = new Uint8Array(dataBuffer);

    // 初始化队列容量
    Atomics.store(controlArray, CONTROL_CAPACITY_INDEX, DATA_BUFFER_CAPACITY_BYTES);

    console.log(`Control Buffer Size: ${controlBuffer.byteLength} bytes`);
    console.log(`Data Buffer Capacity: ${dataBuffer.byteLength} bytes`);
    console.log(`Initial controlArray: `, controlArray);

    // 3. 实例化主线程的生产者环形缓冲区
    producerRingBuffer = new SharedRingBuffer(controlArray, dataBufferView);

    // 4. 创建并启动消费者 Workers
    for (let i = 0; i < NUM_CONSUMER_WORKERS; i++) {
        const worker = new Worker('worker.js', { type: 'module' });
        worker.id = workerIdCounter++;
        consumerWorkers.push(worker);

        worker.onmessage = (e) => {
            if (e.data.type === 'result') {
                console.log(`[Main] Worker ${worker.id} processed task:`, e.data.payload);
            } else if (e.data.type === 'log') {
                console.log(`[Worker ${worker.id} Log]: ${e.data.message}`);
            }
        };

        worker.onerror = (e) => {
            console.error(`[Main] Worker ${worker.id} error:`, e);
        };

        // 将 SharedArrayBuffer 传递给 Worker
        // SharedArrayBuffer 在 postMessage 时是按引用传递的，而不是复制
        worker.postMessage({
            type: 'init',
            controlBuffer: controlBuffer,
            dataBuffer: dataBuffer,
            workerId: worker.id
        });
        console.log(`[Main] Started Worker ${worker.id} and sent SABs.`);
    }

    // 5. 开始生产任务
    console.log(`[Main] Starting to produce ${INITIAL_TASKS_TO_PRODUCE} tasks...`);
    for (let i = 0; i < INITIAL_TASKS_TO_PRODUCE; i++) {
        const task = {
            id: `task-${i}`,
            data: `Hello from main thread, task number ${i}!`
        };
        const enqueued = producerRingBuffer.enqueueTask(task);
        if (!enqueued) {
            console.error(`[Main] Failed to enqueue task ${i}.`);
            // 考虑在此处添加一个小的延迟，以允许消费者处理一些任务
            await new Promise(resolve => setTimeout(resolve, 50)); 
            // 再次尝试或跳过，取决于具体策略
        }
        // 为了演示，我们快速生产。实际应用中，生产速度可能受限或有延迟。
        // await new Promise(resolve => setTimeout(resolve, 10)); // 模拟生产延迟
    }
    console.log(`[Main] Finished producing ${INITIAL_TASKS_TO_PRODUCE} tasks.`);
    console.log(`[Main] Final task count in queue: ${producerRingBuffer.getTaskCount()}`);
}

function stopSystem() {
    console.log('--- Stopping Workers ---');
    isProducing = false; // 停止生产标志，如果生产是持续的
    consumerWorkers.forEach(worker => {
        worker.postMessage({ type: 'terminate' }); // 发送终止信号
        worker.terminate(); // 强制终止（如果worker没有优雅处理terminate消息）
        console.log(`[Main] Terminated Worker ${worker.id}.`);
    });
    consumerWorkers = [];
    controlBuffer = null;
    dataBuffer = null;
    controlArray = null;
    dataBufferView = null;
    producerRingBuffer = null;
    workerIdCounter = 0;
    console.log('System stopped.');
}

3.5 消费者 Worker (worker.js)

Worker接收SharedArrayBuffers，实例化SharedRingBuffer，并进入一个循环来消费任务。

// worker.js
import {
    CONTROL_HEAD_INDEX,
    CONTROL_TAIL_INDEX,
    CONTROL_COUNT_INDEX,
    CONTROL_CAPACITY_INDEX
} from './shared_constants.js';
import SharedRingBuffer from './SharedRingBuffer.js';

let controlArray = null;
let dataBufferView = null;
let workerRingBuffer = null;
let workerId = -1;
let running = false;

// 辅助函数，用于向主线程发送日志
function postLog(message) {
    self.postMessage({ type: 'log', message: `Worker ${workerId}: ${message}` });
}

self.onmessage = async (e) => {
    if (e.data.type === 'init') {
        controlArray = new Int32Array(e.data.controlBuffer);
        dataBufferView = new Uint8Array(e.data.dataBuffer);
        workerId = e.data.workerId;
        workerRingBuffer = new SharedRingBuffer(controlArray, dataBufferView);
        running = true;
        postLog(`Initialized and ready to consume tasks.`);
        consumeTasks(); // 启动消费者循环
    } else if (e.data.type === 'terminate') {
        running = false;
        postLog('Termination signal received. Stopping consumer loop.');
    }
};

async function consumeTasks() {
    while (running) {
        const task = workerRingBuffer.dequeueTask(); // 尝试获取任务

        if (task) {
            // 模拟任务处理时间
            postLog(`Processing task ID: ${task.id}`);
            await new Promise(resolve => setTimeout(resolve, Math.random() * 50 + 20)); // 20-70ms
            self.postMessage({ type: 'result', payload: { taskId: task.id, processedBy: workerId, result: `Done: ${task.data}` } });
        } else {
            // 如果 dequeueTask 返回 null，说明队列暂时为空且 Atomics.wait 已经处理了等待逻辑
            // 理论上，如果 Atomics.wait 返回 'ok'，task 不会是 null，除非反序列化失败。
            // 这里的 else 块更多是作为防御性编程，或者处理超时/错误情况。
            // 实际上，如果队列为空，`dequeueTask` 会在 `Atomics.wait` 处阻塞。
            // 如果 `running` 标志变为 false，则 `Atomics.wait` 可能被中断或超时（如果设置了超时）。
            // 为了优雅退出，我们可以给 `Atomics.wait` 设置一个短超时，或者在 `terminate` 消息中强制唤醒。
            // 例如：`Atomics.notify(controlArray, CONTROL_COUNT_INDEX, -1)` 可以唤醒所有等待者。
            // 为了简单，当前实现依赖 `running` 标志来在下一次循环迭代时退出。
            // 每次循环迭代，如果队列为空，就会再次 `Atomics.wait`。
            // 当 `running` 变为 `false` 时，如果 Worker 此时正在 `Atomics.wait`，
            // 它会一直等到被 `notify` 或超时。一个更优雅的退出方式是主线程在发送 `terminate` 时也 `notify` 一次。
            // 或者，在 `dequeueTask` 中添加一个超时，以便定期检查 `running` 状态。
            // postLog('No task available, will wait for next notification.');
        }
        // 极小延迟以避免紧密循环，虽然 Atomics.wait 已经防止了忙等待
        // await new Promise(resolve => setTimeout(resolve, 1)); 
    }
    postLog('Consumer loop stopped.');
}

3.6 运行项目

1. 将上述所有代码文件（index.html, main.js, worker.js, shared_constants.js, SharedRingBuffer.js）放在同一个目录下。
2. 使用一个支持HTTP服务器并能设置COOP和COEP头的工具来运行它。例如，使用 http-server：
   • 安装: npm install -g http-server
   • 运行: http-server . --cors -H "Cross-Origin-Opener-Policy: same-origin" -H "Cross-Origin-Embedder-Policy: require-corp"
3. 打开浏览器访问 http://localhost:8080。
4. 打开开发者工具的控制台，点击 "Start Workers & Produce Tasks" 按钮。

你将看到主线程生产任务，然后Worker们轮流从共享队列中取出任务并处理，同时在控制台输出日志。

4. 健壮性、性能与高级考量

4.1 错误处理与超时

• 序列化/反序列化错误: 在 enqueueTask 和 dequeueTask 中，我们已经加入了 try-catch 来处理 JSON.stringify 和 JSON.parse 可能引发的错误。
• Atomics.wait 超时: Atomics.wait 可以接受一个 timeout 参数。在消费者长时间等待任务而无任务到来时，可以设置一个超时。如果返回 'timed-out'，消费者可以选择重试、记录日志或优雅地退出。这对于需要定期检查Worker状态或在特定时间后放弃任务的场景非常有用。
• 任务缓冲区溢出: 我们的 enqueueTask 中有一个简单的 if (totalSizeNeeded > this.dataCapacityBytes) 检查。如果任务太大，它将直接拒绝。对于生产速度过快导致队列持续满载的情况，生产者会通过 Atomics.wait 自动暂停。

4.2 优雅关闭与终止

当前的 stopSystem 简单地 terminate() Workers。一个更优雅的关闭流程是：

1. 主线程停止生产新任务。
2. 主线程向所有Worker发送一个“终止”信号（通过postMessage）。
3. Worker收到信号后，不再从队列中取新任务，但会完成所有已在处理中的任务。
4. 当Worker处理完所有任务或队列为空后，它会自行退出（self.close()）或向主线程发送一个“我已准备好关闭”的信号。
5. 主线程在收到所有Worker的确认后，再安全地释放SharedArrayBuffer。
   为了让 Atomics.wait 能够响应终止信号，Worker可以在 Atomics.wait 中设置一个短超时，以便定期检查 running 标志。或者，主线程在发送终止信号时，可以向 CONTROL_COUNT_INDEX 发送一个 Atomics.notify，即使没有实际数据变化，也可以唤醒所有等待的消费者，让他们检查 running 状态。

4.3 多生产者/多消费者

我们的模型天生支持多生产者和多消费者。

• 多生产者: 多个主线程或Worker都可以实例化 SharedRingBuffer 并调用 enqueueTask。Atomics.add(controlArray, CONTROL_COUNT_INDEX, 1) 和 Atomics.notify(controlArray, CONTROL_COUNT_INDEX, 1) 操作是原子的，可以正确处理并发生产。
• 多消费者: 多个Worker同时调用 dequeueTask。Atomics.sub(controlArray, CONTROL_COUNT_INDEX, 1) 和 Atomics.notify(controlArray, CONTROL_COUNT_INDEX, 1) 同样是原子的。当 Atomics.notify 唤醒多个消费者时，只有一个能成功获取任务并递减 CONTROL_COUNT_INDEX，其他被唤醒的消费者会再次检查 CONTROL_COUNT_INDEX，如果发现已经变为0，则会再次进入 Atomics.wait。

4.4 性能考量

• 序列化/反序列化: 尽管我们使用了共享内存，任务数据本身仍然需要序列化成字节流并反序列化回JavaScript对象。JSON.stringify和JSON.parse的开销依然存在。对于非常高性能的场景，可能需要自定义二进制协议或直接在SharedArrayBuffer中维护结构化数据（例如，固定大小的结构体）。
• Atomics.wait/notify 开销: 相较于忙等待或postMessage，Atomics.wait/notify 的开销极低。它涉及操作系统级别的线程停车和唤醒，CPU周期消耗微乎其微。
• 缓存一致性: Atomics操作在硬件层面保证了缓存一致性，确保所有Worker看到的是最新的共享内存状态。这避免了开发者手动处理内存屏障的复杂性。

4.5 潜在的复杂性

• 变长消息的环形队列: 变长消息在环形队列中管理空间比固定长度消息复杂得多。我们的简化实现没有考虑碎片化和最优化空间利用。在实际生产环境中，你可能需要更复杂的内存分配器来管理dataBufferView中的空间。例如，可以维护一个空闲块列表，或者使用更先进的无锁队列算法。
• 僵尸进程/死锁: 如果notify没有被正确调用，或者wait的value参数设置不当，可能会导致线程永久休眠，形成死锁或僵尸进程。严谨的逻辑是关键。

5. 展望与未来方向

我们所构建的生产者消费者模型是许多高级并发模式的基础。在此基础上，可以进一步探索：

• Atomics.waitAsync: 这是一个正在提案中的API，它允许在主线程中异步等待Atomics.wait，而不会阻塞主线程。这将极大地简化主线程与Worker之间的异步协调。
• 工作窃取（Work Stealing）: 在多消费者模型中，一个空闲的消费者可以从另一个繁忙消费者的任务队列中“窃取”任务，以更好地平衡负载。
• 优先队列: 实现一个支持任务优先级的队列，高优先级的任务可以插队处理。这需要更复杂的共享内存数据结构和同步逻辑。
• 更高效的序列化: 对于特定类型的数据，可以考虑使用Protocol Buffers、FlatBuffers或其他二进制序列化库，直接在Uint8Array中操作，避免JSON.stringify/parse的开销。

SharedArrayBuffer和Atomics为Web平台带来了真正的共享内存并发能力，为构建高性能、响应迅速的Web应用打开了新的大门。理解并掌握它们，是迈向Web高性能编程的关键一步。

结语

本次讲座详细探讨了如何利用Atomics.wait和Atomics.notify构建一个高效的跨Worker生产者消费者模型。通过共享内存和原子操作，我们实现了极低开销的任务分发，显著提升了Web应用的并发处理能力与响应性。希望这些理论与实践能为你在高性能Web开发中提供有益的指导。