可观察的异步与同步

Question

我怎么知道Observable生产者何时是异步还是同步？

同步示例:

Observable.of([1, 2, 3])

另一个异步示例(ngrx Store,请看这里)

this.store.take(1);

而现在一个明显的异步示例:

this.http.get(restUrl)

我完全理解它是如何工作的,并且一些Observable可以同步和其他异步.我不明白的是我怎么能预先区分出来？例如,生产者有一个明显的界面告诉我它会是异步的吗？

TL;博士

这个问题出现的主要原因是由于上面链接的答案(这里).@Sasxa已经(正确地)回答我们可以使用同步调用从ngrx存储中获取最新值:

function getState(store: Store<State>): State {
    let state: State;

    store.take(1).subscribe(s => state = s);

    return state;
}

然而,知道可观察量通常是异步的,我看到了这一点并立即想到了RACE CONDITION!在subscribe调用函数之前,该方法可以返回一个未定义的值.

我通过Store和Observable的内部调试,这个例子确实是同步的,但我应该怎么知道呢？对于那些NGRX知道,select方法 store(用于获取最新的值)是异步的,毫无疑问的,因为这是给我们的反应鬼,这就是为什么我来到假设我做到了.

这意味着我无法重构上面的代码如下:

function getLatest(observable: Observable): any {
    let obj: any;

    observable.take(1).subscribe(latest => obj = latest);

    return obj;
}

我可以用任何Observable来调用它,它可以用于同步生成器 - 它甚至可以用于异步生成器的某些时间,但是如果传入异步observable,这毫无疑问是竞争条件.

Answer 1

可以确定observable是否异步,以确定是否使用异步调度程序(或)直接调度,它暴露为:Scheduler.asyncScheduler.asapfoo$.scheduler

let schedulers = [null, Scheduler.asap];
let randomScheduler = schedulers[~~(Math.random()*2)]
let foo$ = Observable.of('foo', randomScheduler);

当foo$进一步处理时,例如与其他运营商链接,该信息变得更不可用.

并且无法确定是否将同步(在相同的时间点)或异步生成值,因为这取决于可观察的内部:

let foo$ = new Observable(observer => {
  if (~~(Math.random()*2))
    setTimeout(() => observer.next('foo'));
  else
    observer.next('foo');
});

TL; DR:肯定不可能知道这一点.

Answer 2

很难确定您获得的 observable 是同步的还是异步的（想象一下您从第三方库返回的 observable）。这就是为什么您以执行顺序可控和可预测的方式编写它。

这也是为什么有像 concat、combineLatest、forkjoin、switchMap、race、merge 这样的操作符来帮助你获得适合不同场景的顺序和性能的原因