假设我们有一个Observable：
var observable = Rx.Observable
.fromEvent(document.getElementById('emitter'), 'click');
我怎样才能使它Complete（什么会为所有订阅的Observers触发onComplete事件）？
【问题讨论】：
标签： javascript stream observable rxjs reactivex
我为我的用例找到了一种更简单的方法，如果你想在 observable 完成时做某事，那么你可以使用这个：
const subscription$ = interval(1000).pipe(
finalize(() => console.log("Do Something")),
).subscribe();
finalize 在完成时触发，当所有订阅都被取消订阅等时。
【讨论】：
我认为您正在寻找的是dispose() 方法。
请注意，subscribe 方法返回一个 Disposable，以便您可以取消订阅序列并轻松地处理它。当你在 observable 序列上调用 dispose 方法时，观察者将停止监听 observable 的数据。通常，您不需要显式调用 dispose，除非您需要提前取消订阅，或者当源 observable 序列的生命周期比观察者更长时。 Rx 中的订阅是为不使用终结器的即发即弃场景而设计的。请注意，Observable 运算符的默认行为是尽快处理订阅（即，当发布 onCompleted 或 onError 消息时）。例如，代码将 x 订阅序列 a 和 b。如果 a 抛出错误，x 将立即从 b 退订。
【讨论】：
Calling unsubscribe() itself does not call complete method
对我有用的是使用take() 运算符。它将在 x 个事件后触发完整的回调。所以通过1，它会在第一个事件之后完成。
打字稿：
private preloadImage(url: string): Observable<Event> {
let img = new Image();
let imageSource = Observable.fromEvent(img, "load");
img.src = url;
return imageSource.take(1);
}
【讨论】：
first()可以代替take(1)
在目前的形式中，你不能。您的 observable 源自未完成的源，因此它本身无法完成。您可以做的是使用完成条件扩展此源。这将像：
var end$ = new Rx.Subject();
var observable = Rx.Observable
.fromEvent(document.getElementById('emitter'), 'click')
.takeUntil(end$);
当你想结束observable时，你做end$.onNext("anything you want here");。在这种情况下，结束事件是由您生成的。如果这是另一个产生该事件的源（按键等），那么您可以直接将一个从该源派生的可观察对象作为takeUntil 的参数。
文档：
【讨论】：
onCompleted 处理程序将为所有观察者触发，这就是您所要求的。关于内存泄漏，当最后一个观察者取消订阅 fromEvent 创建的 observable 时，将删除创建的事件监听器。
fromEvent 创建的可观察对象时，创建的事件监听器将被删除。哦-这就是使您的回答真正有用的原因。谢谢 ;)