使用RxJS / Angular组合管道运算符

RxJS 和 Angular 中的组合管道运算符是一种强大的功能，它们允许你通过组合多个操作符来创建复杂的流处理逻辑。以下是一些常用的组合管道运算符及其在 Angular 中的使用示例。

1. `mergeMap` / `flatMap`

mergeMap（在 RxJS 5 中称为 flatMap）用于将每个源值投射到一个 Observable 中，并将这些内部 Observable 合并到输出 Observable 中。

import { of } from 'rxjs';
import { mergeMap } from 'rxjs/operators';

of(1, 2, 3).pipe(
  mergeMap(id => http.get(`https://api.example.com/items/${id}`))
).subscribe(response => console.log(response));

2. `concatMap`

concatMap 类似于 mergeMap，但它会按顺序订阅每个内部 Observable，并将它们的值按顺序发出。

import { of } from 'rxjs';
import { concatMap } from 'rxjs/operators';

of(1, 2, 3).pipe(
  concatMap(id => http.get(`https://api.example.com/items/${id}`))
).subscribe(response => console.log(response));

3. `switchMap`

switchMap 用于取消之前的内部 Observable，并订阅新的内部 Observable。这对于处理搜索框输入等场景非常有用。

import { fromEvent } from 'rxjs';
import { switchMap, map } from 'rxjs/operators';

const input = document.getElementById('search-input');
fromEvent(input, 'input').pipe(
  map(event => event.target.value),
  switchMap(query => http.get(`https://api.example.com/search?q=${query}`))
).subscribe(response => console.log(response));

4. `exhaustMap`

exhaustMap 用于忽略新的源值，直到当前的内部 Observable 完成。

import { interval, fromEvent } from 'rxjs';
import { exhaustMap, take } from 'rxjs/operators';

const button = document.getElementById('fetch-button');
const interval$ = interval(1000).pipe(take(5));

fromEvent(button, 'click').pipe(
  exhaustMap(() => interval$)
).subscribe(value => console.log(value));

5. `combineLatest`

combineLatest 用于组合多个 Observable 的最新值，并在任何一个 Observable 发出新值时发出组合值。

import { interval } from 'rxjs';
import { combineLatest, map } from 'rxjs/operators';

const interval1$ = interval(1000);
const interval2$ = interval(1500);

combineLatest([interval1$, interval2$]).pipe(
  map(([value1, value2]) => `${value1} - ${value2}`)
).subscribe(console.log);

6. `forkJoin`

forkJoin 类似于 combineLatest，但它只在所有输入 Observable 完成时发出组合值。

import { forkJoin } from 'rxjs';
import { ajax } from 'rxjs/ajax';

const requests = [
  ajax.getJSON('https://api.example.com/items/1'),
  ajax.getJSON('https://api.example.com/items/2')
];

forkJoin(requests).subscribe(([response1, response2]) => {
  console.log(response1, response2);
});

在 Angular 中的使用示例

假设我们有一个简单的 Angular 组件，它使用 HttpClient 来获取数据：

import { Component, OnInit } from '@angular/core';
import { HttpClient } from '@angular/common/http';
import { Observable } from 'rxjs';
import { map, mergeMap } from 'rxjs/operators';

@Component({
  selector: 'app-example',
  template: `<div>{{ data$ | async | json }}</div>`
})
export class ExampleComponent implements OnInit {
  data$: Observable<any>;

  constructor(private http: HttpClient) {}

  ngOnInit() {
    this.data$ = this.http.get('https://api.example.com/items').pipe(
      mergeMap(items => items)
    );
  }
}

在这个示例中，我们使用 mergeMap 来处理从 API 获取的数组，并将其转换为单个值的流。

通过这些组合管道运算符，你可以灵活地处理复杂的异步数据流，从而提高代码的可读性和可维护性。