源完成时，RxJS取消订阅内部可观察对象

import { Observable, Subscription } from 'rxjs';

const observable = new Observable(/* 可观察对象的定义 */);
const subscription = observable.subscribe(/* 订阅处理函数 */);

// 取消订阅
subscription.unsubscribe();

import { Observable, Subject } from 'rxjs';
import { takeUntil } from 'rxjs/operators';

const observable = new Observable(/* 可观察对象的定义 */);
const stopSignal = new Subject();

observable.pipe(takeUntil(stopSignal)).subscribe(/* 订阅处理函数 */);

// 某个条件满足时，发送停止信号
stopSignal.next();

import { Observable, asyncScheduler } from 'rxjs';
import { unsubscribeOn } from 'rxjs/operators';

const observable = new Observable(/* 可观察对象的定义 */);

observable.pipe(unsubscribeOn(asyncScheduler)).subscribe(/* 订阅处理函数 */);

// 在指定的调度器上取消订阅
asyncScheduler.schedule(() => {
  subscription.unsubscribe();
});