如何优雅地结合Rxjs和WebWorker

[lsa2127291]

在项目中同时使用rxjs和webworker的场景并不多，所以相关的资料也较少，恰巧我在开发项目时遇到了很适合使用这两项技术的场景，所以尝试着将其结合起来使用，也收到了很不错的效果，所以分享一下自己的使用经验和心得。
同时也专门写了将两者相结合的插件rxjs-webworker，给有这方面需求的朋友提供便利的使用方式。

简单介绍下这两种技术

• weborker
  webworker为浏览器赋予了多进程工作的能力，当遇到计算量大、耗时长的工作时，为了避免阻塞ui进程的工作，我们可以将其放入worker中运行待其运行完成后再显示结果。
• rxjs
  rxjs则是一个很特殊的异步编程库，它是以响应式风格设计的，受限于篇幅和本文的主题，这里不详细介绍rxjs，想要了解可以去官网或看其他相关文章。只说一下我认为rxjs最大的两点特色，其一就是它可以推送多个值这让它非常适合如分段加载、进度条等持续性异步事件，其二就是它丰富的操作符带来的对异步事件极强的可控性，在复杂的异步编程场景中使用rxjs会让你惊讶于它的强大，以后我还会发文深入介绍rxjs神奇的地方。

如何结合这两种技术

两者结合最好的方式就是将webworker抽象为rxjs中的一条流，如同抽象event和promise一样。
所以我仿造fromEvent和fromPromise，设计了fromWorker。通过fromWorker，我们可以非常简单的将worker变为流，比如：

const worker = new Worker('./worker.js')
fromWorker(worker).subscribe(() => {
  console.log('worker executed')
})

同时，它也支持直接传递函数，如：

fromWorker(() => {
  let count = 0
  while (count < 1000000) {
    count++
  }
  self.postMessage(count)
}).subscribe(count => {
  console.log('worker executed', count)
})

另外，它还允许你给worker传值，比如：

fromWorker(e => {
  while (count < 1000000) {
    count++
  }
  self.postMessage(`${e.data} ${count}`)
}, 'hello').subscribe(value => {
  // hello 1000000
  console.log(value)
})

除了fromWorker以外，我还设计了一个算子mapWorker，它的作用和map相似，不同之处在与它内部是运行在worker中的，用法也很简单，比如：

of(0).pipe(
  mapWorker(val => {
    // 运行在worker中
    while(val < 1000000) {
      val++
    }
    return val
  })
).subscribe(val => {
  // 1000000
  console.log(val)
})

注意：mapWorker中不能与主进程共享全局变量
利用fromWorker和mapWorker我们可以很方便得在rxjs中使用webworker

两者结合能带来何种好处

webworker本身是难以管理的，我们只能使用postMessage和onMessage与worker进程交流，但是结合rxjs后，大幅加强了对worker的管理，举个例子：

const cancel$ = timer(4000)
let valueMap = {}
const fromWorker$ = interval(1000).pipe(
  mergeMap(val => {
    return fromWorker(e => {
      setTimeout(() => {
        self.postMessage(e.data + 10000)
      }, 2000)
    }, val)
  }),
  map(val => {
    valueMap[val] = val + 20000
    return val
  }),
  takeUntil(cancel$)
)
fromWorker$.subscribe(val => {
  // value 10001
  console.log('value', val)
})
setInterval(() => {
  // {10001: 30001}
  console.log(valueMap)
}, 2000)

上面的例子中，我们看到受益于rxjs提供takeUntil算子，webworker可以被及时取消，大幅简化了我们去控制worker何时停止postMessage。
类似的情况还有很多，总之两者结合最大好处就是能借助rxjs强大对异步操作的可控性对webworker进行全面管理。
在实际项目中，两者的结合成功的将运算繁重的解压任务完全置于"后台"完成，丝毫不会阻塞我的ui线程，得到了类似客户端中多线程运行的体验，这对于浏览器来说是非常难以做到的。

总结

当项目复杂到一定程度时，特别是有很多运算量极大的地方时，结合rxjs和webworker能发挥你浏览器最大的性能获得如同App般的体验，很推荐有这样需求的人来尝试这两种技术。
最后再次小小宣传一下我自己基于rxjs6制作的rxjs-webworker，希望能帮助大家更加简单优雅地在rxjs中加入webworker。

[langhuihui]

非常好的想法！