Web Workers中的消息传递模式

Web Workers中的消息传递模式

Web Workers为JavaScript提供了多线程能力，允许在主线程之外运行脚本。由于Worker线程与主线程相互隔离，消息传递成为两者通信的唯一方式。这种机制催生了几种典型的消息模式，每种模式适用于不同的场景。

简单请求-响应模式

最基本的消息模式是单向请求-响应。主线程发送消息给Worker，Worker处理完成后返回结果。这种模式适合离散的、无状态的计算任务。

// 主线程代码
const worker = new Worker('worker.js');
worker.postMessage({ type: 'calculate', data: 42 });

worker.onmessage = (event) => {
  console.log('Received:', event.data.result);
};

// worker.js
self.onmessage = (event) => {
  if (event.data.type === 'calculate') {
    const result = performHeavyCalculation(event.data.data);
    self.postMessage({ result });
  }
};

这种模式的特点是每次通信都是独立的，Worker不保留任何状态。当需要处理图像滤镜、数学计算等场景时特别有效。

长连接对话模式

当需要维持长时间通信时，可以建立对话式消息传递。这种模式下，双方通过消息ID或序列号来关联请求和响应。

// 主线程
const worker = new Worker('worker.js');
let messageId = 0;

function sendWithId(payload) {
  const id = messageId++;
  worker.postMessage({ ...payload, _id: id });
  return new Promise((resolve) => {
    worker.addEventListener('message', function callback(event) {
      if (event.data._id === id) {
        worker.removeEventListener('message', callback);
        resolve(event.data);
      }
    });
  });
}

// 使用示例
sendWithId({ type: 'startSession' })
  .then(response => console.log('Session started:', response));

Worker端需要维护一个消息处理器映射表：

// worker.js
const handlers = new Map();

self.onmessage = (event) => {
  const { _id, type, ...data } = event.data;
  
  if (type === 'startSession') {
    const sessionId = createSession();
    handlers.set(_id, () => ({ sessionId }));
  }
  
  // 其他处理逻辑...

  const handler = handlers.get(_id);
  if (handler) {
    self.postMessage({ ...handler(), _id });
    handlers.delete(_id);
  }
};

这种模式适合需要多次交互的场景，如数据库操作、复杂工作流等。

发布-订阅模式

当多个组件需要监听Worker消息时，可以采用发布-订阅模式。Worker作为消息中心，主线程的不同模块可以订阅特定类型的消息。

// 主线程
const worker = new Worker('worker.js');
const subscribers = {
  progress: [],
  result: []
};

worker.onmessage = (event) => {
  const { type, data } = event.data;
  subscribers[type]?.forEach(callback => callback(data));
};

function subscribe(type, callback) {
  subscribers[type].push(callback);
  return () => {
    subscribers[type] = subscribers[type].filter(cb => cb !== callback);
  };
}

// 使用示例
const unsubscribe = subscribe('progress', (percent) => {
  console.log(`Progress: ${percent}%`);
});

// 稍后取消订阅
unsubscribe();

Worker端可以广播不同类型的消息：

// worker.js
function processTask() {
  for (let i = 0; i <= 100; i += 10) {
    self.postMessage({ type: 'progress', data: i });
    simulateWork();
  }
  self.postMessage({ type: 'result', data: finalResult });
}

这种模式在需要实时更新UI进度、多组件协同等场景下特别有用。

双向流模式

对于需要持续双向通信的场景，如实时数据处理或聊天应用，可以建立双向流。这种模式下，双方都可以随时发送消息，形成持续的对话。

// 主线程
const worker = new Worker('worker.js');
const messageQueue = [];
let isWorkerReady = false;

worker.onmessage = (event) => {
  if (event.data === 'ready') {
    isWorkerReady = true;
    messageQueue.forEach(msg => worker.postMessage(msg));
    messageQueue.length = 0;
  } else {
    handleWorkerMessage(event.data);
  }
};

function sendToWorker(message) {
  if (isWorkerReady) {
    worker.postMessage(message);
  } else {
    messageQueue.push(message);
  }
}

// worker.js
self.postMessage('ready');

self.onmessage = (event) => {
  const processed = process(event.data);
  self.postMessage(processed);
};

这种模式需要注意消息顺序和流量控制。可以通过添加序列号或使用Transferable对象来优化性能。

共享内存模式

当需要传输大量数据时，可以使用SharedArrayBuffer实现共享内存。这种模式避免了消息拷贝的开销，但需要手动同步。

// 主线程
const sharedBuffer = new SharedArrayBuffer(1024);
const sharedArray = new Int32Array(sharedBuffer);

const worker = new Worker('worker.js');
worker.postMessage({ buffer: sharedBuffer });

// 修改共享数据
Atomics.store(sharedArray, 0, 42);

// worker.js
let sharedArray;

self.onmessage = (event) => {
  if (event.data.buffer) {
    sharedArray = new Int32Array(event.data.buffer);
    startProcessing();
  }
};

function startProcessing() {
  // 读取共享数据
  const value = Atomics.load(sharedArray, 0);
  // 处理数据...
}

这种模式适合高频更新的数据，如音频处理、物理模拟等。但需要注意竞态条件，使用Atomics API进行同步。

错误处理模式

健壮的Worker通信需要包含错误处理机制。可以通过专用消息类型或错误优先回调来处理异常。

// 主线程
worker.postMessage({ type: 'process', data: input });

worker.onmessage = (event) => {
  if (event.data.type === 'error') {
    console.error('Worker error:', event.data.message);
    return;
  }
  // 正常处理...
};

// worker.js
self.onmessage = async (event) => {
  try {
    const result = await dangerousOperation(event.data);
    self.postMessage(result);
  } catch (error) {
    self.postMessage({
      type: 'error',
      message: error.message,
      stack: error.stack
    });
  }
};

对于未捕获的异常，还需要监听error事件：

worker.onerror = (event) => {
  console.error('Worker runtime error:', event);
};

性能优化技巧

消息传递会带来序列化和反序列化的开销。对于大型数据，可以使用Transferable对象来转移所有权：

// 主线程
const largeBuffer = new ArrayBuffer(10000000);
worker.postMessage({ buffer: largeBuffer }, [largeBuffer]);
// 此后largeBuffer不能再被主线程访问

// worker.js
self.onmessage = (event) => {
  const buffer = event.data.buffer;
  // 直接使用buffer...
};

另一种优化是批量处理消息，减少通信次数：

// worker.js
let messageQueue = [];
let isProcessing = false;

self.onmessage = (event) => {
  messageQueue.push(event.data);
  if (!isProcessing) {
    processBatch();
  }
};

async function processBatch() {
  isProcessing = true;
  while (messageQueue.length) {
    const batch = messageQueue.splice(0, 10); // 每次处理10条
    const results = await processInBatch(batch);
    self.postMessage(results);
  }
  isProcessing = false;
}

复杂状态管理

当Worker需要维护复杂状态时，可以采用状态机模式。通过定义明确的状态转换规则来管理消息处理逻辑。

// worker.js
const State = {
  IDLE: 'idle',
  PROCESSING: 'processing',
  PAUSED: 'paused'
};

let currentState = State.IDLE;
let pendingData = null;

self.onmessage = (event) => {
  switch (currentState) {
    case State.IDLE:
      if (event.data.type === 'start') {
        currentState = State.PROCESSING;
        processData(event.data);
      }
      break;
    case State.PROCESSING:
      if (event.data.type === 'pause') {
        currentState = State.PAUSED;
      } else if (event.data.type === 'data') {
        pendingData = event.data;
      }
      break;
    case State.PAUSED:
      if (event.data.type === 'resume') {
        currentState = State.PROCESSING;
        if (pendingData) {
          processData(pendingData);
          pendingData = null;
        }
      }
      break;
  }
};

这种模式特别适合需要精确控制工作流程的场景，如分步数据处理、任务队列等。