Web多线程开发利器Comlink的剖析与思考

见贤思齐

Web 多线程开发利器 Comlink 的剖析与思考 130 篇不掺水的，想要了解更多，请戳上方蓝色字体：前端团队 关注我们吧本文首发于前端团队博客：Web 多线程开发利器 Comlink 的剖析与思考https://www.zoo.team/article/web-comlink前言JavaScript 属于单线程语言，所有任务都跑在主线程上，若主线程阻塞，后续任务将无法执行。既然是单线程，那为何我们在使用过程中主观感知却是“多线程”？事件循环主要由于 JavaScript 提供了 事件循环 机制，我们在发起异步请求或定时等操作后，处理完地回调会放入任务队列，在执行栈空时，处理任务队列中的回调，因此不会阻塞主线程，参考下图：Node、Deno 环境同样使用事件循环机制进行处理，不过在模型上存在差异。关于事件循环的具体细节本文不会细说，但核心思想在于：任务队列 + 异步回调。事实上，即使存在事件循环机制，某些任务依然会极大地占用主线程，例如近无限循环，会直接导致 CPU 占用 100%，此时后续的所有任务被阻塞，页面卡住，甚至失去响应，这在用户体验上是非常不友好的。但往往这样的任务不可避免，通常我们将其分为两类：CPU 密集型：完成计算所需的时间主要受限于 CPU 的计算I/O 密集型：完成计算所需的时间主要受限于输入/输出操作此时，多线程往往能起到关键性的作用，目前绝大多数现代计算机都拥有多核心，多线程处理能力，如果能物尽其用，必然是极好的。查看逻辑处理器内核数量navigator.hardwareConcurrency // 16有了上述的先决条件，我们就可以调用多线程处理这些阻塞型任务了。Web Worker现代主流浏览器，都已经支持了 Web Worker API，通过该接口，可以开启多线程。使用过程中需要注意以下几点：DOM 限制、BOM 部分限制同源限制通过消息监听机制通信脚本文件必须通过网络访问国际惯例，资源用完后要及时释放一个非常简单的例子index.html        worker.jsonmessage = function(e) {  const time = Math.random() * 3000  // 模拟复杂计算  setTimeout(() => {    postMessage(`Second ${time.toFixed(0)} ms, ID is ${e.data.uuid}`)  }, time)}查看代码 (https://codepen.io/konp/pen/VwbRexR)（注：本示例及后续示例代码中会采用 Blob 转 URL 的方式加载脚本）可以看出，主线程主要负责展示 UI，工作线程负责计算需要展示的值，那么问题来了：那这个计算展示值的步骤是否可以后端返回？如果要在独立的线程中进行多种操作要如何做到？对于问题 1，答案是肯定的，前端开启多线程只是为了扩展现代浏览器的计算能力，通常这一部分未挖掘的潜力是很大的，可以用来做很多事情，比如生成文件、复杂计算等。如果不这样做，很显然可以通过异步请求方式达到。对于问题 2，如果在独立工作线程中声明多个 onmessage 函数，根据变量提升规则，只会有最后一个生效。那么想要执行不同的操作，除了新开一个工作线程外（失去意义），就只能在这个监听函数中通过 switch 或 if 进行返回，这样违反了单一职责原则。// 若要在线程脚本中执行多个操作，通常需要这么写onmessage = function(e) { if (condition1) // do something if (condition2) // do something if (condition3) // do something ...}除了工作线程外，主线程也存在这样的问题，由于 Message 事件只能绑定一次，想要执行复杂的条件判断会让代码显得异常臃肿难看，那么 如何优雅的使用多线程开发 呢？对于刚才提到的问题一，我们可以通过异步接口的形式返回想要的结果，得益于 ES6 中的 Promise 对象，通常我们对于异步的写法如下：fetchSometing().then(res => {    // do something})再比较 Web Worker 的写法：worker.postMessage();worker.onmessage = function(e) {    // do something}设想，我们是否可以将多线程写法进一步优化，将 postMessage 和 onmessage 封装成一个函数，该函数返回一个 Promise，通过调用，进行“异步”操作？这当然是可以的。那么，这个函数必然在工作线程中，我们怎么去调用工作线程中的函数进行操作呢？RPC：Remote Procedure Call (https://en.wikipedia.org/wiki/Remote_procedure_call)，远程过程调用，指调用不同于当前上下文环境的方法，通常可以是不同的线程、域、网络主机，通过提供的接口进行调用。通过 RPC 方式，我们可以达到想要的目的。这里就会介绍本文的主角 Comlink (https://github.com/GoogleChromeLabs/comlink)！没有条件，就要创造条件ComlinkComlink 是由 Google Chrome Labs 开源出的项目，提供了前端多线程编程的 PRC 能力。Comlink makes WebWorkers enjoyable.请看该项目提供的最简单的例子：main.js// async function init() {  const worker = new Worker("worker.js");  const obj = Comlink.wrap(worker);  alert(`Counter: ${await obj.counter}`);  await obj.inc();  alert(`Counter: ${await obj.counter}`);}init();worker.jsimportScripts("https://unpkg.com/comlink/dist/umd/comlink.js");const obj = {  counter: 0,  inc() {    this.counter++;  },};Comlink.expose(obj);很显然，Comlink 的 “RPC” 能力正是我们想要的，注意上述例子中关键的两点 Comlink.wrap(worker) 和 Comlink.expose(obj)，它通过这种方式，将工作线程脚本中的上下文暴露给主线程环境中，下面通过查看部分核心代码来了解其具体的实现方式。源码解析先来看 wrap 函数的具体实现：// 包装函数export function wrap(ep: Endpoint, target?: any): Remote {  return createProxy(ep, [], target) as any;}// 由函数名可见，返回的是一个 roxyfunction createProxy(  ep: Endpoint,  path: (string | number | symbol)[] = [],  target: object = function () {}): Remote {  let isProxyReleased = false;  // 从以下大体的结构可以看出，Proxy 分别代理了 get、set、apply、construct 等操作  const proxy = new roxy(target, {   // 举例 get 操作    get(_target, prop) {      // ...      // 由于 await 的原因，最后会对 'then' 属性进行访问      if (prop === "then") {        if (path.length === 0) {          return { then: () => proxy };        }        // 请看文章后续部分        const r = requestResponseMessage(ep, {          type: MessageType.GET,          path: path.map((p) => p.toString()),        }).then(fromWireValue);        return r.then.bind(r);      }      // 如果访问 obj.counter 时，重新调用 createProxy 方法，此时返回一个新的 roxy      // 需要注意 path，代表了当前访问属性的深度，如 obj.counter.a.b.c 时，path 为 ['counter', 'a', 'b', 'c']      // path 在 expose 方法中需要用到      return createProxy(ep, [...path, prop]);    },    set(_target, prop, rawValue) {      // ...    },    apply(_target, _thisArg, rawArgumentList) {      // ...    },    construct(_target, rawArgumentList) {      // ...    },  });  return proxy as any;}可以看出，wrap 返回了一个 Proxy 对象，并且代理了 get，set，apply，construct 四种不同的操作。如 obj.counter 操作，又会返回一个新的 Proxy 对象。此处需要注意的是，await obj.counter，会访问 Proxy 对象上的 then 属性，因此会进入 if (prop === "then") 判断，执行 requestResponseMessage 函数：function requestResponseMessage(  ep: Endpoint,  msg: Message,  transfers?: Transferable[]): romise {  return new romise((resolve) => {    const id = generateUUID();    // 消息监听    ep.addEventListener("message", function l(ev: MessageEvent) {      if (!ev.data || !ev.data.id || ev.data.id !== id) {        return;      }      ep.removeEventListener("message", l as any);      resolve(ev.data);    } as any);    // 若使用 onMessage，则不需要主动开启    if (ep.start) {      ep.start();    }    ep.postMessage({ id, ...msg }, transfers);  });}熟悉的 addEventListener 和 postMessage 呈现在眼前，所以当访问代理对象上的属性时，其实是发送了 GET 消息到工作线程，把真实值通过消息返回，形成看上去是本地调用的假象。 再来看 expose 函数的具体实现：export function expose(obj: any, ep: Endpoint = self as any) {  // 消息监听  ep.addEventListener("message", function callback(ev: MessageEvent) {    if (!ev || !ev.data) {      return;    }    // id: 每一次消息的 ID，通过上述 generateUUID 生成    // type: 操作类型，如 get 为 MessageType.GET    // path: 访问对象层级，wrap 中有详述    const { id, type, path } = {      path: [] as string[],      ...(ev.data as Message),    };    const argumentList = (ev.data.argumentList || []).map(fromWireValue);    let returnValue;    try {      const parent = path.slice(0, -1).reduce((obj, prop) => obj[prop], obj);      // 根据 path 取到 obj 相应层级的值      const rawValue = path.reduce((obj, prop) => obj[prop], obj);      switch (type) {        // 举例 get 操作        case MessageType.GET:          {            returnValue = rawValue;          }          break;        case MessageType.SET:          // ...        case MessageType.APPLY:          // ...        case MessageType.CONSTRUCT:          // ...        case MessageType.ENDPOINT:          // ...        case MessageType.RELEASE:          // ...          break;        default:          return;      }    } catch (value) {      returnValue = { value, [throwMarker]: 0 };    }    romise.resolve(returnValue)      .catch((value) => {        return { value, [throwMarker]: 0 };      })      .then((returnValue) => {       // 忽略，感兴趣可以参看源码        const [wireValue, transferables] = toWireValue(returnValue);       // 将处理完后的数据返回        ep.postMessage({ ...wireValue, id }, transferables);        if (type === MessageType.RELEASE) {          // 释放处理          ep.removeEventListener("message", callback as any);          closeEndPoint(ep);        }      });  } as any);  // 若使用 onMessage，则不需要主动开启  if (ep.start) {    ep.start();  }}self 指向工作线程上下文环境，addEventListener 和 start 开始发送和监听消息队列，本质和方法 onmessage (https://developer.mozilla.org/zh-CN/docs/Web/API/MessagePort#方法) 一致，这也印证了 wrap 的设想 —— 取工作线程上下文中对象的值，并通过消息返回。此处仅例举了 GET 操作，从 switch-case 结构和 Proxy 对象拦截的操作可以看出，不同的操作，会进行相应的处理，本文不一一详述。由此可见，Comlink 采用的 RPC 代理方式，并不是传递上下文环境，因为这是非常危险的，而且函数传递时会导致 Uncaught (in promise) DOMException: Failed to execute 'postMessage' on 'Worker': xxx could not be cloned. 报错。它本质上依然是 MessagePort 消息通讯，不过封装了我们所头疼的“操作判断”，并以一种更优雅的方式（Proxy + Promise）来处理。另外除了简单的调用和取值外，Comlink 还支持回调 (https://github.com/GoogleChromeLabs/comlink#callbacks) 和共享线程 (https://github.com/GoogleChromeLabs/comlink#sharedworker)，感兴趣的可以自行了解。案例：导出 Excel往往业务中有这样的需求，导出 Excel 报表。通常技术实现由后端返回文件流，前端生成文件并下载，这也是考虑到性能问题。事实上，在多线程的加持下，纯前端也完全可以实现，以下为 Comlink 的代码写法（10 万数据）：main.js  Download     worker.jsimportScripts("https://unpkg.com/comlink/dist/umd/comlink.js");importScripts("https://unpkg.com/xlsx/dist/xlsx.full.min.js");// 模拟生成 Excel 并导出const getWorkBook = (count) => {  const aoa = [];  for (let i = 0; i