使用 Web Worker 实现简单的非阻塞异步

摘要：只会把一个函数延后执行，但还是在主线程中执行，执行函数的时候会阻塞线程。规范并没有定义多线程，至今也没有原生的多线程实现。然而在中却定义了用于实现浏览器中的多线程。使用也非常简单，只需要预先在中注册事件，在主线程中给处理就好了。

之前的文章提到了 JavaScript 中的异步编程，然而无论早就存在的 setTimeout 还是 ES6 中的 Promise，它们都是 阻塞 异步，执行函数的时候，会阻塞线程。setTimeout 只会把一个函数延后执行，但还是在主线程中执行，执行函数的时候会阻塞线程。换句话说，setTimeout 只实现了过程间并发（concurrent）而未实现并行（parallel）。

ES 规范并没有定义多线程，Node.js 至今也没有原生的多线程实现。然而在 HTML5 中却定义了 Web Worker 用于实现浏览器中的多线程。

引用 MDN 原文：

Web Workers 使得一个Web应用程序可以在与主执行线程分离的后台线程中运行一个脚本操作。这样做的好处是可以在一个多带带的线程中执行费时的处理任务，从而允许主（通常是UI）线程运行而不被阻塞/放慢。

与朴素（原始）的多线程编程方式不同，Web Worker 通常不允许线程间共享数据，所以没有线程同步、数据竞争等问题，更没有没有锁（Mutex）和条件变量（Condition variable）等概念（注 1）。它们使用 postMessage 相互通信，可以认为是 JS 中的参与者模式实现。各个 Worker 间数据独立，不共享内存：postMessage 始终通过结构化克隆的方式深拷贝传值。

使用 Web Worker 也非常简单，只需要预先在 Worker 中注册 message 事件，在主线程中 postMessage 给 Worker 处理就好了。处理完后可以再通过 postMessage 传结果给主线程。

需要注意的是，Web Worker 中不可以操作 DOM，一切与 DOM 操作相关的函数、类都不能使用（创建一个 DOM 元素发回给主线程 appendChild 也不行），所以可以使用的方法非常有限，只适用于处理数据（注 2）。

前面提到 Promise 是阻塞异步，那是否可以把要处理的数据转发给某个 Worker 处理并返回一个 Promise，在处理完后将其 resolve 掉呢？

答案当然是可以的，而且实现并不复杂。

首先当然是 new 一个 Worker 出来。需要注意的是 Worker 的构造函数 接受的是一个 JavaScript 脚本的 URL，可否接受 data-uri 看浏览器，实测 Chrome、Firefox 可以，Safari、Edge 不行（会抛 SECURITY_ERR 异常）。

简单起见，这里还是采取 data-uri 的形式。考虑可移植性的话可以先指定一个静态文件，然后使用 postMessage 把函数体传过去。

this._worker = new Worker("data:text/javascript," + encodeURIComponent(`"use strict";
const __fn = ${fn};
onmessage = e => postMessage(__fn(...e.data));`));

Worker 中做了两件事：

定义一个函数变量 __fn，其值 fn 是需要执行的函数。如果 fn 本身是一个函数对象，这里将其转换为字符串，相当于把函数的源代码拼到了字符串里。

绑定 message 事件。将传入的值作为参数列表调用 __fn，然后将 __fn 的返回值通过 postMessage 传给主函数。

function dispatch(...args) {
  return new Promise((resolve, reject) => {
    this._queue.push({ resolve, reject });
    this._worker.postMessage(args);
  });
}

返回一个 Promise。注意这里不能只是简单的 postMessage。因为如果使用者多次调用 dispatch 函数一次创建了多个 Promise，之后很难确定是哪个 Promise 完成了。这里通过一个队列记忆创建的 Promise 顺序，然后依次 resolve（单个 Worker 处理 message 事件还是顺序执行的）。当然你也可以多传一个标记值给 Worker 用于标记被 resolve 的 Promise。

JavaScript 里的队列就是数组：

this._queue = [];

this._worker.onmessage = e => this._queue.shift().resolve(e.data);
this._worker.onerror = e => this._queue.shift().reject(e.error);

onmessage 表示正常返回；onerror 表示出现了异常。对应的 Promise 的 resolve 和 reject 直接从队列里取出来。

class Dispatcher {
  constructor(fn) {
    this._queue = [];
    this._worker = new Worker("data:text/javascript," + encodeURIComponent(`"use strict";
const __fn = ${fn};
onmessage = e => postMessage(__fn(...e.data));`));
    this._worker.onmessage = e => this._queue.shift().resolve(e.data);
    this._worker.onerror = e => this._queue.shift().reject(e.error);
  }

  dispatch(...args) {
    return new Promise((resolve, reject) => {
      this._queue.push({ resolve, reject });
      this._worker.postMessage(args);
    });
  }
}

这就是完整代码了，总共不到 20 行。使用的话也很简单：

const dispatcher = new Dispatcher(arr => { // 创建对象，把入口函数传入
  for (let i=0; i<1000; ++i) arr.sort(); // 耗费些时间
  return arr;  // 返回处理后的结果
});

const arr = Array.from({ length: 8192 }, () => Math.random() * 10000); // 需要处理的数据
dispatcher.dispatch(arr)  // 派发给 Worker
  .then(res => console.log(res));  // 处理完毕后输出

在浏览器中测试，会生成这样一段代码：

排序大数组 1000 次的同时 UI 响应仍然不受影响。

这里还有一个线程池的版本，可以创建多个 Worker 同时并行执行多个任务：https://github.com/CarterLi/T...

因为要区分究竟是哪个 Worker 完成运行，处理 Worker 返回值的逻辑复杂了一些，有什么建议欢迎提出。

注 1：ES2017 中加入 SharedArrayBuffer 后已经可以在主线程和各 Web Worker 间共享数据，使用 Atomics.wait() 和 Atomics.wake() 还可以实现传统意义上的锁和条件变量。但由于其出现较晚且并非使用 Web Worker 的主流方式，这里不展开讨论。

注 2：还有一个可能是在 Worker 中画图，见 OffscreenCanvas。一旦实现，对游戏编程是个不小的帮助。