javascript - 如何取消订阅或取消对 RxJS observable 大型数组的过滤？

Question

我的理解是，整个数组被推送给订阅者，这与可以取消订阅/取消的间隔观察者不同。

例如以下取消工作...

// emit a value every second for approx 10 seconds
let obs = Rx.Observable.interval(1000)
  .take(10)
let sub = obs.subscribe(console.log);

// but cancel after approx 4 seconds
setTimeout(() => {
  console.log('cancelling');
  sub.unsubscribe()
}, 4000);

<script src="https://unpkg.com/rxjs@5.5.10/bundles/Rx.min.js"></script>

但是，用数组替换间隔则不会。

// emit a range
let largeArray = [...Array(9999).keys()];
let obs = Rx.Observable.from(largeArray)
let sub = obs.subscribe(console.log);

// but cancel after approx 1ms
setTimeout(() => {
  console.log('cancelling');
  sub.unsubscribe()
}, 1);

// ... doesn't cancel

<script src="https://unpkg.com/rxjs@5.5.10/bundles/Rx.min.js"></script>

是否需要以某种方式使每个元素异步，例如将其包装在 setTimeout(..., 0) 中？也许我已经盯着这个问题太久了，我完全没有想到可以取消数组的处理？

Answer 1

当在数组上使用 from(...) 时，所有值都将同步发出，这不允许向 setTimeout 授予任何执行时间您用来取消订阅的。事实上，它甚至在到达 setTimeout 行之前就完成了发射。为了允许发出不占用线程，您可以使用异步调度程序(from(..., Rx.Scheduler.async))，它将使用 setInterval 调度工作。

以下是文档:https://github.com/ReactiveX/rxjs/blob/master/doc/scheduler.md#scheduler-types

这是一个运行示例。我不得不将超时时间提高到 100，以便有更多的喘息空间。这当然会减慢你的执行速度。我不知道你尝试这样做的原因。如果您可以分享确切的用例，我们可能会提供一些更好的建议。

// emit a range
let largeArray = [...Array(9999).keys()];
let obs = Rx.Observable.from(largeArray, Rx.Scheduler.async);
let sub = obs.subscribe(console.log);

// but cancel after approx 1ms
setTimeout(() => {
  console.log('cancelling');
  sub.unsubscribe()
}, 100);

// ... doesn't cancel

<script src="https://unpkg.com/rxjs@5.5.10/bundles/Rx.min.js"></script>