Angular使用Subject和Observable处理异步数据流

????????在Angular中常用RxJS库中的Subject和Observable处理异步数据流。

概念：

• Observable：Observable是一个表示异步数据流的对象，它可以被订阅以获取数据并且可以被取消订阅。Observable通常用于创建和处理异步数据流，可以进行各种操作符的链式调用来处理数据流。
• Subject：Subject是一个特殊的Observable（可转换成Observable），它允许在数据流中添加新的值，并且可以同时作为Observable和Observer。它可以用来创建可观察对象，并且可以手动推送新的值到数据流中。Subject通常用于多播数据流，可以被多个观察者订阅。

相同点：

1. 都用于处理数据流：Subject和Observable都用于表示数据流，可以被订阅以获取数据，并且可以进行各种操作符的链式调用来处理数据流。

2. 都支持订阅和取消订阅：无论是Subject还是Observable，都支持订阅以获取数据，并且可以通过取消订阅来停止获取数据。

不同点：

1. Subject是观察者模式的实现：Subject既是可观察对象，也是观察者。它可以手动推送新的值到数据流中，并且可以被多个观察者订阅。而Observable只是一个可观察对象，它通常用于创建和处理数据流，但不能手动推送新的值。

2. Subject是热数据流，Observable是冷数据流：当多个观察者订阅一个Subject时，它们会共享同一个数据流，无论何时订阅，都会收到相同的数据。而Observable是冷数据流，每个观察者订阅时都会创建一个独立的数据流，它们之间是相互独立的。

3. Subject没有延迟执行：当Subject推送新的值时，所有已经订阅它的观察者都会立即收到这个值。而Observable可以延迟执行，只有在有观察者订阅时才会开始执行。

使用场景：

????????在Angular中有很多处理异步操作，比如从服务器获取数据、处理用户输入等，这些都可以用Subject或者Observe来实现。同时 Angular内置的一些服务（比如HttpClient）返回的数据通常是Observable对象，因此我们也可以使用Observable来订阅和处理这些异步数据流。

Subject:

• 当只需要处理一个数据源时，可以使用Subject来处理，在下面的示例中，我们创建了一个Subject作为数据源，通过它的next()方法来向数据源推送数据，并通过subscribe()方法来订阅数据源的事件。

import { Subject } from 'rxjs';

const dataSource = new Subject<number>();

// 订阅数据源
dataSource.subscribe({
  next: value => console.log(`获得数据：${value}`)
});

// 向数据源推送数据
dataSource.next(1);
dataSource.next(2);
dataSource.next(3);

?Observer:

• 示例一：当需要处理多个数据源时，可以使用Observer来处理，在下面的示例中，我们创建了一个Observer用于订阅多个数据源的事件，并通过subscribe()方法来订阅数据源的事件。当数据源发生错误或者完成时，Observer也可以对其进行响应。

import { Observer } from 'rxjs';

const observer: Observer<number> = {
  next: value => console.log(`获得数据：${value}`),
  error: error => console.error(`发生异常：${error}`),
  complete: () => console.log('操作完成')
}

// 创建数据源
const dataSource1$ = of(1, 2, 3);
const dataSource2$ = new Promise(resolve => setTimeout(() => resolve(4), 3000));

// 订阅数据源
dataSource1$.subscribe(observer);
dataSource2$.then(value => observer.next(value));

• 示例二：Observable处理异步数据流完整流程，在下面的示例中，我们首先创建了一个Observable对象，它表示一个异步数据流，每隔1秒推送一个新的数字。我们使用subscribe()方法订阅这个Observable对象，以获取它推送的新值，并对这些值进行处理。最后，我们使用unsubscribe()方法取消了对Observable的订阅。

import { Observable } from 'rxjs';

// 创建Observable对象，表示一个异步数据流
const observable = new Observable<number>((observer) => {
  let count = 0;

  // 模拟异步操作，每隔1秒推送一个新的值
  const intervalId = setInterval(() => {
    observer.next(count++);
  }, 1000);

  // 当Observable被取消订阅时，清除定时器
  return () => {
    clearInterval(intervalId);
  };
});

// 订阅Observable对象，以获取数据
const subscription = observable.subscribe(
  (data) => {
    console.log('Received data: ' + data);
  },
  (error) => {
    console.error('Error: ' + error);
  },
  () => {
    console.log('Completed');
  }
);

// 取消订阅Observable对象
subscription.unsubscribe();

Subject和Observer结合使用：? ? ??

• 在下面的示例中，我们创建了一个Subject作为数据源，同时也创建了一个Observer作为订阅者，然后通过subscribe()方法来完成订阅器的注册。这时在通过Subject的next()方法推送数据时，订阅者就可以收到数据了。

import { Subject, Observer } from 'rxjs';

const subject = new Subject<number>();

const observer: Observer<number> = {
  next: value => console.log(`获得数据：${value}`),
  error: error => console.error(`发生异常：${error}`),
  complete: () => console.log('操作完成')
}

// 订阅数据源
subject.subscribe(observer);

// 向数据源推送数据
subject.next(1);
subject.next(2);
subject.next(3);

转换使用：

? ? ? ? 可使用asObservable()方法将Subject对象转换为一个Observable对象。

import { Subject, Observable } from 'rxjs';

// 创建一个Subject对象
const subject = new Subject<number>();

// 创建一个Observable对象，用于订阅Subject的数据流
const observable = subject.asObservable();

// 订阅Observable，获取数据并处理
const subscription = observable.subscribe(
? (data) => {
? ? console.log('Received data: ' + data);
? },
? (error) => {
? ? console.error('Error: ' + error);
? },
? () => {
? ? console.log('Completed');
? }
);

// 推送新的值到Subject的数据流中
subject.next(1);
subject.next(2);
subject.next(3);

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

????????在上面的示例中，我们首先创建了一个Subject对象，然后使用asObservable()方法将其转换为一个Observable对象。接下来，我们订阅了这个Observable对象，以获取Subject推送的新值，并对这些值进行处理。最后，我们通过next()方法向Subject推送了一些新的值，并最终取消了订阅。