RxJS 是一种基于事件流和数据流的编程模式，它可以让我们更容易的处理异步数据，比如网络请求、用户输入、定时器等等。在 Angular6 中，RxJS 是一个非常重要的工具，几乎每个 Angular 应用都需要用到它。本文将介绍完整的 RxJS 学习笔记，包括基础概念、操作符、实践案例等等。
基础概念
Observable
Observable 是 RxJS 中最重要的概念之一，它代表一个可以被订阅的事件流。我们可以通过 Observable.create() 或 new Observable() 来创建一个 Observable，如下所示：
import { Observable } from 'rxjs';
const observable = new Observable(subscriber => {
subscriber.next(1);
subscriber.next(2);
subscriber.next(3);
subscriber.complete();
});在上面的代码中，我们创建了一个 Observable，它会依次发出数字 1、2、3，然后完成。我们可以通过 subscribe() 方法来订阅这个 Observable，如下所示：
observable.subscribe({
next: value => console.log(value),
complete: () => console.log('complete')
});在上面的代码中，我们订阅了这个 Observable，当它发出事件时，我们会依次输出数字 1、2、3，最后输出 'complete'。
Operator
Operator 是 RxJS 中用来操作 Observable 的方法，它可以对 Observable 发出的每个事件进行处理，比如过滤、映射、合并等等。我们可以通过 pipe() 方法来使用 Operator，如下所示：
// www.javascriptcn.com code example
import { of } from 'rxjs';
import { filter, map } from 'rxjs/operators';
const source = of(1, 2, 3, 4, 5);
const result = source.pipe(
filter(value => value % 2 === 0),
map(value => value * value)
);
result.subscribe(value => console.log(value));在上面的代码中，我们创建了一个 Observable，它会依次发出数字 1、2、3、4、5。然后我们使用 filter() 操作符过滤出偶数，再使用 map() 操作符将偶数平方。最后输出数字 4、16。
Subject
Subject 是 RxJS 中另一个重要的概念，它既是 Observable，又是 Observer。我们可以通过 new Subject() 来创建一个 Subject，如下所示：
// www.javascriptcn.com code example
import { Subject } from 'rxjs';
const subject = new Subject();
subject.subscribe({
next: value => console.log('subscriber 1:', value)
});
subject.next(1);
subject.subscribe({
next: value => console.log('subscriber 2:', value)
});
subject.next(2);在上面的代码中，我们创建了一个 Subject，它可以被多个订阅者订阅。当我们调用 next() 方法时，所有订阅者都会收到这个事件。在上面的代码中，首先订阅者 1 收到数字 1，然后订阅者 2 也收到数字 1。接着订阅者 1 和订阅者 2 都收到数字 2。
常用操作符
map
map() 操作符用于将 Observable 中的每个事件都映射成新的事件。它的参数是一个函数，这个函数接收 Observable 发出的事件作为参数，返回一个新的事件。比如下面的例子：
// www.javascriptcn.com code example
import { from } from 'rxjs';
import { map } from 'rxjs/operators';
const source = from([1, 2, 3, 4, 5]);
const result = source.pipe(
map(value => value * value)
);
result.subscribe(value => console.log(value));在上面的代码中，我们创建了一个 Observable，它会依次发出数字 1、2、3、4、5。然后我们使用 map() 操作符将每个数字平方。最后输出数字 1、4、9、16、25。
filter
filter() 操作符用于过滤 Observable 中的事件。它的参数是一个函数，这个函数接收 Observable 发出的事件作为参数，返回一个布尔值，表示是否保留这个事件。比如下面的例子：
// www.javascriptcn.com code example
import { from } from 'rxjs';
import { filter } from 'rxjs/operators';
const source = from([1, 2, 3, 4, 5]);
const result = source.pipe(
filter(value => value % 2 === 0)
);
result.subscribe(value => console.log(value));在上面的代码中，我们创建了一个 Observable，它会依次发出数字 1、2、3、4、5。然后我们使用 filter() 操作符过滤出偶数。最后输出数字 2、4。
take
take() 操作符用于从 Observable 中取出指定数量的事件。它的参数是一个数字，表示要取出的事件数量。比如下面的例子：
// www.javascriptcn.com code example
import { interval } from 'rxjs';
import { take } from 'rxjs/operators';
const source = interval(1000);
const result = source.pipe(
take(5)
);
result.subscribe(value => console.log(value));在上面的代码中，我们创建了一个 Observable，它会每隔 1 秒发出一个数字。然后我们使用 take() 操作符取出前 5 个数字。最后输出数字 0、1、2、3、4。
merge
merge() 操作符用于将多个 Observable 合并成一个 Observable。它的参数是多个 Observable，可以是数组或可变参数。比如下面的例子：
import { interval, merge } from 'rxjs';
const source1 = interval(1000);
const source2 = interval(2000);
const result = merge(source1, source2);
result.subscribe(value => console.log(value));在上面的代码中，我们创建了两个 Observable，它们分别每隔 1 秒和 2 秒发出一个数字。然后我们使用 merge() 操作符将它们合并成一个 Observable。最后输出数字 0、0、1、1、2、2、...。
combineLatest
combineLatest() 操作符用于将多个 Observable 中最新的事件合并成一个事件。它的参数是多个 Observable，可以是数组或可变参数。比如下面的例子：
import { interval, combineLatest } from 'rxjs';
const source1 = interval(1000);
const source2 = interval(2000);
const result = combineLatest([source1, source2]);
result.subscribe(value => console.log(value));在上面的代码中，我们创建了两个 Observable，它们分别每隔 1 秒和 2 秒发出一个数字。然后我们使用 combineLatest() 操作符将它们合并成一个 Observable。最后输出数组 [0, 0]、[1, 0]、[1, 1]、[2, 1]、[2, 2]、...。
实践案例
聊天室
下面是一个简单的聊天室应用，它使用了 Angular6 和 RxJS。用户可以输入用户名和消息，然后发送消息到服务器，服务器会广播这条消息给所有用户。每个用户都可以看到所有其他用户发送的消息。
// www.javascriptcn.com code example
import { Component } from '@angular/core';
import { WebSocketSubject } from 'rxjs/webSocket';
@Component({
selector: 'app-chat-room',
template: `
<div *ngFor="let message of messages">{{message}}</div>
<input [(ngModel)]="username" placeholder="username">
<input [(ngModel)]="message" placeholder="message">
<button (click)="send()">send</button>
`
})
export class ChatRoomComponent {
private socket: WebSocketSubject<any>;
private messages: string[] = [];
public username: string;
public message: string;
constructor() {
this.socket = new WebSocketSubject('ws://localhost:8080');
this.socket.subscribe(message => this.messages.push(message));
}
public send() {
this.socket.next(`${this.username}: ${this.message}`);
this.message = '';
}
}在上面的代码中，我们使用了 WebSocketSubject 来创建一个 WebSocket 连接，然后订阅这个连接的消息。当用户点击发送按钮时，我们使用 next() 方法发送一条消息到服务器。
自动完成
下面是一个简单的自动完成组件，它使用了 Angular6 和 RxJS。用户可以输入关键字，然后从服务器获取匹配的结果，最后显示在下拉菜单中。当用户选择一个结果时，会触发一个事件。
// www.javascriptcn.com code example
import { Component } from '@angular/core';
import { FormControl } from '@angular/forms';
import { debounceTime, distinctUntilChanged, switchMap } from 'rxjs/operators';
import { HttpClient } from '@angular/common/http';
@Component({
selector: 'app-auto-complete',
template: `
<input [formControl]="keyword">
<ul>
<li *ngFor="let result of results" (click)="select(result)">{{result}}</li>
</ul>
`
})
export class AutoCompleteComponent {
private url = 'http://localhost:8080/search';
public keyword = new FormControl();
public results: string[] = [];
constructor(private http: HttpClient) {
this.keyword.valueChanges.pipe(
debounceTime(300),
distinctUntilChanged(),
switchMap(keyword => this.http.get<string[]>(`${this.url}?q=${keyword}`))
).subscribe(results => this.results = results);
}
public select(result: string) {
// emit event
}
}在上面的代码中，我们使用了 FormControl 来监听输入框的变化，然后使用 RxJS 操作符来处理输入事件。当用户输入一个关键字时，我们使用 debounceTime() 操作符来延迟发送请求，避免频繁请求。然后使用 distinctUntilChanged() 操作符来过滤掉重复的关键字。最后使用 switchMap() 操作符来将关键字转换成一个 Observable，然后订阅这个 Observable，获取服务器返回的结果。当用户选择一个结果时，我们可以触发一个事件，让父组件处理。
Source: FunTeaLearn,Please indicate the source for reprints https://funteas.com/post/67d9c9f5a941bf7134186357