RxJS：如何订阅仅在订阅之后才发出的 Observable 数据？答案

【问题标题】：RxJS: How do I subscribe to Observable data that is emitted only after the moment of subscription?RxJS：如何订阅仅在订阅之后才发出的 Observable 数据？
【发布时间】：2016-08-14 07:34:21
【问题描述】：

上下文：我有很多ConnectableObservables，几乎所有的重播计数都为 1 或更多。在任何给定时间，都有许多观察者订阅和取消订阅。

我想要什么：在许多情况下，当观察者订阅这些可观察对象之一时，我并不关心我可能收到的预先存在的发射数据，因为可观察对象的数据重放机制。最近订阅的观察者唯一感兴趣的数据是订阅之后发出的数据。

const observable = Rx.Observable
    .interval(100)
    .take(4)
    .publishReplay(3);

observable.connect();

问题：据我所知，当观察者订阅 observable 时，它无法知道它观察到的数据是在订阅之前还是之后发出的。

observable.subscribe(x => console.log('observed', x));

setTimeout(() => 
    observable.subscribe(y => console.log('delayed observed', y)), 
    400
);

上面的代码会输出：

// => observed 0
// => observed 1
// => observed 2
// => delayed observed 0 **don't care**
// => delayed observed 1 **don't care**
// => delayed observed 2 **don't care**
// => observed 3
// => delayed observed 3

在这种假设情况下，延迟观察者只对订阅时刻之后发出的数据感兴趣；在这种情况下，3。

我已经搜索了RxJS 5 reference docs，但似乎找不到一个银弹操作员来完成我所追求的。有什么想法吗？

【问题讨论】：

您要解决的实际问题是什么？你能提供一个不那么抽象的例子吗？
那么谁需要回放呢？延迟的观察者不能订阅未重播的流版本吗？
你想要的是调用 Hot Observables。查看此文档github.com/Reactive-Extensions/RxJS/blob/master/doc/…

标签： reactive-programming rxjs rxjs5

【解决方案1】：

您可以使用.skipUntil(Rx.Observable.timer(0))，因为重放的元素将被同步重放，而skipUntil 将在接收重放值的那一刻停止同步执行Observable 的其余部分。

这段代码会产生你想要的结果：

const observable = Rx.Observable
    .interval(100)
    .take(4)
    .publishReplay(3);

observable.subscribe(x => console.log('observed', x));

setTimeout(() => 
    observable
        .skipUntil(Rx.Observable.timer(0))
        .subscribe(y => console.log('delayed observed', y)), 
    400
);

observable.connect();

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.3/Rx.min.js"&gt;&lt;/script&gt;

【讨论】：

【解决方案2】：

你能做这样的事情吗？

const observableNoReplay = Rx.Observable
    .interval(100)
    .take(4);

const observable = observableNoReplay
    .publishReplay(3);

observable.connect();

您可以订阅当时需要的任何 observable，而不必担心任何形式的灵丹妙药。

【讨论】：