如何取消RxJS中大型数组Observable的订阅或过滤操作？

这是个很典型的RxJS同步Observable和异步Observable的差异问题，我来帮你理清楚：

首先，你遇到的核心问题是：用数组创建的Observable是同步执行的。当你调用obs.subscribe(console.log)的瞬间，RxJS会立刻遍历整个数组，把所有元素一次性推送给订阅者——这个过程在当前事件循环的同步阶段就完成了，根本等不到1ms后setTimeout触发unsubscribe的时候。而interval是异步Observable，它的元素是每隔1秒（通过定时器）推送到事件循环的宏任务队列里，所以在4秒时取消订阅，后续的元素还没被推送，自然能生效。

那怎么让数组的Observable变成可取消的呢？关键是把同步推送改成异步推送，让元素的分发过程被放到事件循环队列中，这样unsubscribe才有机会中断后续的推送。这里有几种常用的方法：

方法1：使用asyncScheduler调度器 #

RxJS的from方法支持第二个参数指定调度器，用asyncScheduler就能让数组的元素异步分发：

// 异步推送数组元素
const largeArray = [...Array(9999).keys()];
const obs = Rx.Observable.from(largeArray, Rx.Scheduler.async);
const sub = obs.subscribe(console.log);

setTimeout(() => {
  console.log('cancelling');
  sub.unsubscribe();
}, 1);

方法2：用concatMap+delay异步包装每个元素 #

如果你需要更灵活的控制（比如给每个元素加延迟），可以用concatMap把每个元素转换成一个异步的Observable：

const largeArray = [...Array(9999).keys()];
const obs = Rx.Observable.from(largeArray)
  .concatMap(item => Rx.Observable.of(item).delay(0)); // delay(0)会把元素放到宏任务队列
const sub = obs.subscribe(console.log);

setTimeout(() => {
  console.log('cancelling');
  sub.unsubscribe();
}, 1);

为什么这两种方法有效？ #

它们本质上都是把原本同步的元素推送，改成了异步分批执行。这样当你调用subscribe后，RxJS不会立刻把所有元素都推完，而是把每个元素的推送任务放到事件循环的队列中。这时候1ms后的setTimeout触发unsubscribe时，还有很多元素的推送任务没执行，就能被成功中断了。

另外补充一点：如果你只是需要异步化而不需要额外延迟，asyncScheduler是更高效的选择，因为delay(0)底层其实也是用了类似的调度机制。

内容的提问来源于stack exchange，提问作者aryeh