rxjs如何观察未完成的可观察对象集合

import { of, from, merge } from 'rxjs';
import { delay, takeUntil } from 'rxjs/operators';

// 创建一些未完成的可观察对象
const observable1 = of(1, 2, 3).pipe(delay(1000));
const observable2 = of(4, 5, 6).pipe(delay(2000));
const observable3 = of(7, 8, 9).pipe(delay(3000));

// 合并这些可观察对象
const mergedObservable = merge(observable1, observable2, observable3);

// 订阅合并后的可观察对象
const subscription = mergedObservable.subscribe({
  next: (value) => console.log(value),
  complete: () => console.log('All observables completed'),
});

// 在5秒后取消订阅
setTimeout(() => {
  subscription.unsubscribe();
  console.log('Subscription cancelled');
}, 5000);

import { of, merge } from 'rxjs';
import { delay, takeUntil } from 'rxjs/operators';

const observable1 = of(1, 2, 3).pipe(delay(1000));
const observable2 = of(4, 5, 6).pipe(delay(2000));
const observable3 = of(7, 8, 9).pipe(delay(3000));

const stopSignal = new Subject();

const mergedObservable = merge(observable1, observable2, observable3).pipe(
  takeUntil(stopSignal)
);

const subscription = mergedObservable.subscribe({
  next: (value) => console.log(value),
  complete: () => console.log('All observables completed'),
});

setTimeout(() => {
  stopSignal.next();
  stopSignal.complete();
  console.log('Stop signal sent');
}, 5000);