在 `Web Workers` 中如何实现一个 `RPC` (远程过程调用) 机制，并处理复杂的数据传输 (如 `Transferable Objects`)？

各位听众，大家好！我是今天的主讲人，很高兴能和大家一起聊聊Web Workers中的RPC（Remote Procedure Call）机制，以及如何优雅地处理复杂数据的传输，尤其是那些让人兴奋的Transferable Objects。今天咱们就来一场代码与理论齐飞，幽默与干货并存的饕餮盛宴，让大家彻底搞懂Worker中的RPC！

第一节：Worker与主线程的爱恨情仇：为什么要RPC？

首先，我们得明白Web Worker是干嘛的。简单来说，它就像一个独立的“小弟”，在后台默默干活，不阻塞主线程（也就是用户看到的前端页面）。主线程呢，就是那个“大哥”，负责处理UI交互，响应用户操作。

但是，“小弟”干完活总要向“大哥”汇报，或者“大哥”有任务要分配给“小弟”。这就需要一种沟通机制，也就是我们今天的主角：RPC。

为什么不用直接共享内存呢？因为Web Workers运行在独立的线程中，它们之间不能直接访问彼此的内存空间。这就好比你在北京，我在上海，我们不能直接把东西从你的冰箱搬到我的冰箱里，只能通过快递（也就是消息传递）来实现。

第二节：基础版RPC：postMessage的简单爱情故事

最基础的RPC实现，就是利用postMessage。主线程和Worker之间通过发送消息来进行通信。

主线程 (main.js):

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

worker.addEventListener('message', (event) => {
  const { id, result } = event.data;
  console.log(`收到Worker #${id} 的结果:`, result);
});

function callWorker(method, params) {
  const id = Math.random().toString(36).substring(2, 15); // 生成唯一ID
  worker.postMessage({ id, method, params });
  return id; // 返回消息ID，方便后续追踪结果
}

const taskId = callWorker('add', [5, 3]);
console.log(`已发送任务 #${taskId} 给Worker`);

Worker 线程 (worker.js):

self.addEventListener('message', (event) => {
  const { id, method, params } = event.data;

  switch (method) {
    case 'add':
      const result = params[0] + params[1];
      self.postMessage({ id, result });
      break;
    default:
      self.postMessage({ id, error: 'Unknown method' });
  }
});

这个例子中，callWorker函数负责向Worker发送消息，消息包含一个唯一的id，调用的method，以及参数params。Worker接收到消息后，根据method执行相应的操作，然后将结果通过postMessage发送回主线程。

优点： 简单易懂，容易上手。

缺点：

手动管理消息ID： 需要自己生成和追踪消息ID，容易出错。
错误处理繁琐： 需要手动处理错误，代码不够优雅。
不支持Promise： 无法像Promise那样方便地处理异步操作。
类型安全： 没有类型检查，容易传递错误的数据类型。

第三节：Promise加持：让RPC拥有现代爱情

为了解决上述问题，我们可以使用Promise来改进我们的RPC机制。

主线程 (main.js):

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

const pendingCalls = new Map(); // 存储待处理的Promise

worker.addEventListener('message', (event) => {
  const { id, result, error } = event.data;
  const promise = pendingCalls.get(id);

  if (promise) {
    pendingCalls.delete(id);
    if (error) {
      promise.reject(error);
    } else {
      promise.resolve(result);
    }
  }
});

function callWorker(method, params) {
  return new Promise((resolve, reject) => {
    const id = Math.random().toString(36).substring(2, 15);
    pendingCalls.set(id, { resolve, reject });
    worker.postMessage({ id, method, params });
  });
}

async function run() {
  try {
    const sum = await callWorker('add', [5, 3]);
    console.log('Sum:', sum);

    const product = await callWorker('multiply', [4, 6]);
    console.log('Product:', product);

    const errorResult = await callWorker('divide', [10, 0]);
    console.log('Division result:', errorResult); // 这行不会执行
  } catch (error) {
    console.error('Error:', error);
  }
}

run();

Worker 线程 (worker.js):

self.addEventListener('message', (event) => {
  const { id, method, params } = event.data;

  switch (method) {
    case 'add':
      const result = params[0] + params[1];
      self.postMessage({ id, result });
      break;
    case 'multiply':
      const result = params[0] * params[1];
      self.postMessage({ id, result });
      break;
    case 'divide':
      if (params[1] === 0) {
        self.postMessage({ id, error: 'Division by zero!' });
      } else {
        const result = params[0] / params[1];
        self.postMessage({ id, result });
      }
      break;
    default:
      self.postMessage({ id, error: 'Unknown method' });
  }
});

这个版本的RPC使用了Promise，callWorker函数返回一个Promise，当Worker返回结果时，Promise会被resolve或reject。这样，我们可以使用async/await来更方便地处理异步操作，并且错误处理也更加清晰。

优点：

Promise支持： 使用Promise处理异步操作，代码更清晰。
错误处理更优雅： 使用Promise的reject机制，错误处理更方便。

缺点：

仍然需要手动管理消息ID： 虽然使用了Promise，但仍然需要手动生成和追踪消息ID。
类型安全： 仍然没有类型检查。
代码冗余： 主线程和Worker线程都需要编写大量的样板代码。

第四节： Transferable Objects：让数据飞起来

到目前为止，我们传递的数据都是通过复制来实现的。这意味着主线程和Worker线程各自拥有一份数据的副本。如果数据量很大，复制的开销就会变得非常可观。

Transferable Objects允许我们将数据的所有权从一个线程转移到另一个线程，而无需进行复制。这就像把房子的钥匙直接交给别人，而不是复印一把钥匙。

支持Transferable Objects的数据类型包括：

ArrayBuffer
MessagePort
ImageBitmap
OffscreenCanvas

下面是一个使用ArrayBuffer的例子：

主线程 (main.js):

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

worker.addEventListener('message', (event) => {
  const { id, result } = event.data;
  console.log(`收到Worker #${id} 的结果:`, result);

  // 如果结果是ArrayBuffer，则进行处理
  if (result instanceof ArrayBuffer) {
    const intArray = new Int32Array(result);
    console.log('Received ArrayBuffer from worker:', intArray);
  }
});

function callWorker(method, params, transfer) {
    const id = Math.random().toString(36).substring(2, 15); // 生成唯一ID
    worker.postMessage({ id, method, params }, transfer); // 注意transfer参数
    return id; // 返回消息ID，方便后续追踪结果
}

const buffer = new ArrayBuffer(1024 * 1024 * 4); // 4MB
const intArray = new Int32Array(buffer);
for (let i = 0; i < intArray.length; i++) {
  intArray[i] = i;
}

const taskId = callWorker('processArray', [buffer], [buffer]); // 将buffer的所有权转移给worker
console.log(`已发送任务 #${taskId} 给Worker，并转移ArrayBuffer的所有权`);

// 此时，intArray在主线程中将无法访问
// console.log(intArray[0]);  // 会报错：intArray is detached

Worker 线程 (worker.js):

self.addEventListener('message', (event) => {
  const { id, method, params } = event.data;

  switch (method) {
    case 'processArray':
      const buffer = params[0];
      const intArray = new Int32Array(buffer);

      // Worker线程可以访问和修改ArrayBuffer
      for (let i = 0; i < intArray.length; i++) {
        intArray[i] *= 2;
      }

      self.postMessage({ id, result: buffer }, [buffer]); // 将buffer的所有权转移回主线程
      break;
    default:
      self.postMessage({ id, error: 'Unknown method' });
  }
});

在这个例子中，我们创建了一个ArrayBuffer，并将其作为参数传递给Worker。关键在于postMessage的第二个参数transfer，它是一个数组，包含了需要转移所有权的对象。

当postMessage执行后，ArrayBuffer的所有权就转移到了Worker线程。这意味着主线程将无法再访问ArrayBuffer，尝试访问会报错。Worker线程可以自由地访问和修改ArrayBuffer，并将修改后的ArrayBuffer的所有权转移回主线程。

优点：

性能提升： 避免了大量数据的复制，提高了性能。
节省内存： 减少了内存占用。

缺点：

所有权转移： 数据的所有权会发生转移，需要小心处理。
类型限制： 只能转移特定类型的数据。

第五节：高级RPC框架：让RPC更上一层楼

手动实现RPC虽然能帮助我们理解其原理，但在实际项目中，我们通常会使用现成的RPC框架，例如：

Comlink: 由Google Chrome团队维护，是一个轻量级的RPC库，可以简化Web Workers的通信。
Workerize: 一个用于将模块转换为Web Workers的工具，可以自动生成RPC接口。

这里我们简单介绍一下 Comlink的使用。

首先安装 Comlink:

npm install comlink

主线程 (main.js):

import * as Comlink from 'comlink';

async function run() {
  const worker = new Worker('worker.js');
  const api = Comlink.wrap(worker);

  try {
    const sum = await api.add(5, 3);
    console.log('Sum:', sum);

    const product = await api.multiply(4, 6);
    console.log('Product:', product);

    const buffer = new ArrayBuffer(1024);
    const resultBuffer = await api.processArray(buffer);
    console.log("Processed ArrayBuffer:", new Uint8Array(resultBuffer));

    worker.terminate(); // 终止worker
  } catch (error) {
    console.error('Error:', error);
  }
}

run();

Worker 线程 (worker.js):

import * as Comlink from 'comlink';

const api = {
  add(a, b) {
    return a + b;
  },
  multiply(a, b) {
    return a * b;
  },
  processArray(buffer) {
    const array = new Uint8Array(buffer);
    for (let i = 0; i < array.length; i++) {
      array[i] = i % 256; // Some processing
    }
    return Comlink.transfer(buffer, [buffer]); // Transferable Objects的使用
  },
};

Comlink.expose(api);

Comlink 简化了主线程和 Worker 线程之间的通信。Comlink.wrap(worker) 在主线程中创建了一个与 worker 中暴露的 api 对象对应的代理对象。然后，你可以像调用本地函数一样调用 worker 中的函数。Comlink.expose(api) 暴露了 worker 中的 api 对象，使其可以被主线程访问。Comlink还处理了数据的序列化和反序列化。

优点：

代码简洁： 大大简化了RPC的实现，减少了样板代码。
类型安全： Comlink支持TypeScript，可以进行类型检查。
Transferable Objects支持： 可以方便地使用Transferable Objects进行数据传输。

第六节：总结与展望

今天我们一起学习了Web Workers中的RPC机制，从最基础的postMessage，到使用Promise进行改进，再到使用Transferable Objects优化数据传输，最后还介绍了高级RPC框架Comlink。

特性	基础版RPC (`postMessage`)	Promise版RPC	Transferable Objects	Comlink
代码简洁	简单	较复杂	较复杂	非常简洁
错误处理	手动	Promise的reject机制	Promise的reject机制	Promise的reject机制
异步操作	回调	Promise	Promise	Promise
数据传输	复制	复制	移动所有权	复制/移动所有权（自动处理）
类型安全	无	无	无	TypeScript支持
适用场景	简单任务	中等复杂度的任务	大数据传输	各种复杂度的任务，推荐使用

未来，随着Web技术的不断发展，我们相信Web Workers和RPC机制会越来越重要，也会涌现出更多优秀的RPC框架，让我们拭目以待！

好了，今天的讲座就到这里，感谢大家的聆听！希望大家有所收获！

发表回复 取消回复

发表回复取消回复