RxJS 是一个 JavaScript 库，用于响应式编程。它提供了一种方便的方式，使得在异步数据流中处理及组合多个事件成为可能，更加的灵活，高效和易于维护。作为前端开发工程师，在应对各种各样复杂且多变的应用场景时，强大的 RxJS 会成为你必不可少的一项技能。在本文中，我将由简单到复杂，详尽地讲解 RxJS，让你从初学者到高阶开发者。
RxJS 基础
创建可观察对象
RxJS 通过 Observable 类型实现了响应式编程模型。在它里面，最基础也是最重要的单位是可观察对象。从最简单的理解来看，可观察对象会将多个值转化为一个响应流。这个过程中，我们所得到的流可能是三种响应：
- next 响应： 当有新的值被推送进流中时响应该事件。
- error 响应： 在出现错误时，处理该错误并结束流中的事件。
- complete 响应： 当所有的事件处理完后响应该事件。
在 RxJS 中，有多种方式可以创建可观察对象，例如使用 from 和 of 等静态函数可以基于一个数组、一个 Promise 或者其他类型的对象创建 Observables。以下示例演示了从一个数组创建 Observable。
// javascriptcn.com 代码示例
import { from } from 'rxjs';
const observable = from([1, 2, 3, 4, 5]);
observable.subscribe({
next: value => console.log(`Value: ${value}`),
error: error => console.log(`Error: ${error}`),
complete: () => console.log('Complete')
});
// Output:
// Value: 1
// Value: 2
// Value: 3
// Value: 4
// Value: 5
// Complete转换可观察对象
在某些情况下，我们可能需要使用一个可观察对象去操作流中的值。在 RxJS 中，我们可以通过操作符来转换可观察对象。以下示例演示如何使用 map 操作符将流中的每项值进行平方运算。
// javascriptcn.com 代码示例
import { from } from 'rxjs';
import { map } from 'rxjs/operators';
const observable = from([1, 2, 3, 4, 5]);
observable.pipe(
map(value => value ** 2)
).subscribe({
next: value => console.log(`Value: ${value}`),
error: error => console.log(`Error: ${error}`),
complete: () => console.log('Complete')
});
// Output:
// Value: 1
// Value: 4
// Value: 9
// Value: 16
// Value: 25
// Complete过滤可观察对象
在某些情况下，我们可能需要通过可观察对象过滤流中的某些值。在 RxJS 中，我们可以使用 filter 操作符来过滤流中的值。以下示例演示如何使用 filter 操作符过滤流中能被 2 整除的值。
// javascriptcn.com 代码示例
import { from } from 'rxjs';
import { filter } from 'rxjs/operators';
const observable = from([1, 2, 3, 4, 5]);
observable.pipe(
filter(value => value % 2 === 0)
).subscribe({
next: value => console.log(`Value: ${value}`),
error: error => console.log(`Error: ${error}`),
complete: () => console.log('Complete')
});
// Output:
// Value: 2
// Value: 4
// CompleteRxJS 进阶
处理多流
有时候，我们需要根据两个或以上的流来计算最终的结果。在 RxJS 中，我们可以使用 combineLatest 操作符处理多流。以下示例演示如何使用 combineLatest 操作符来计算两个流的和。
// javascriptcn.com 代码示例
import { fromEvent, combineLatest } from 'rxjs';
import { map } from 'rxjs/operators';
const input1 = document.getElementById('input1')!;
const input2 = document.getElementById('input2')!;
const input1$ = fromEvent(input1, 'input').pipe(
map(event => (event.target as HTMLInputElement).value)
);
const input2$ = fromEvent(input2, 'input').pipe(
map(event => (event.target as HTMLInputElement).value)
);
combineLatest(input1$, input2$).pipe(
map(([value1, value2]) => parseInt(value1) + parseInt(value2))
).subscribe({
next: value => console.log(`Sum: ${value}`),
error: error => console.log(`Error: ${error}`),
complete: () => console.log('Complete')
});处理自定义的可观察对象
正常情况下，我们可以通过 RxJS 自带的操作符来处理一些常见的可观察对象。但是，有时候我们需要处理和计算一些自定义的可观察对象，并需要自定义操作符。在 RxJS 中，我们可以通过 Observable.create 函数来创建自定义的可观察对象。以下示例演示了创建一个简单的可观察对象，并添加一个操作符以获得数组平均值。
// javascriptcn.com 代码示例
import { Observable } from 'rxjs';
const source$ = Observable.create(obs => {
const array = [1, 2, 3, 4, 5];
const sum = array.reduce((a, b) => a + b);
obs.next(sum);
obs.complete();
});
const average$ = source$.pipe(
map(sum => sum / 5)
);
average$.subscribe({
next: value => console.log(`Average: ${value}`),
error: error => console.log(`Error: ${error}`),
complete: () => console.log('Complete')
});
// Output:
// Average: 3
// Complete处理异步操作
在许多情况下，我们需要处理具有不确定性和异步性的数据。例如，一些 Web API 调用以及浏览器中的一些事件，如定时器和用户输入等都是异步的。在 RxJS 中，我们可以使用 interval 操作符来处理定时器，使用 fromEvent 操作符处理浏览器事件，并使用 Http 拦截器处理 Web API。以下示例演示如何使用 RxJS 处理一个 Web API 请求。
// javascriptcn.com 代码示例
import { from, pipe } from 'rxjs';
import { map, catchError } from 'rxjs/operators';
import { ajax } from 'rxjs/ajax';
const url = 'https://api.github.com/users?per_page=5';
const result$ = ajax(url).pipe(
map(response => response.response),
catchError(error => {
console.log(`Error: ${error}`);
return from([]);
})
);
result$.subscribe({
next: value => console.log(`Result: ${JSON.stringify(value)}`),
error: error => console.log(`Error: ${error}`),
complete: () => console.log('Complete')
});RxJS 高级
处理更加复杂的场景
在实际的应用场景中，我们经常需要处理和编排更加复杂的场景并能够管理和优化整个数据流。在 RxJS 中，我们可以使用 Subject 和 Multicast 等操作符来实现数据的集成和管理，并使用 switchMap 和 mergeMap 等操作符来优化流程。以下示例演示如何使用 switchMap 操作符以及 Subject 操作符来处理多个异步请求。
// javascriptcn.com 代码示例
import { Subject, of, zip } from 'rxjs';
import { switchMap } from 'rxjs/operators';
import { ajax } from 'rxjs/ajax';
const urlPrefix = 'https://api.github.com/users/';
const userEvents$ = new Subject<string>();
const user$ = userEvents$.pipe(
switchMap(name => zip(
ajax(`${urlPrefix}${name}`),
ajax(`${urlPrefix}${name}/repos`)
))
);
user$.subscribe({
next: value => console.log(`Result: ${JSON.stringify(value)}`),
error: error => console.log(`Error: ${error}`),
complete: () => console.log('Complete')
});
userEvents$.next('mojombo');
// Output:
// Result: [{"login":"mojombo"...]
userEvents$.next('torvalds');
// Output:
// Result: [{"login":"torvalds"...]处理大量的数据集合
在一些实际的应用场景中，我们可能需要处理大量的数据集合，并需要进行各种处理从而得到准确的结果。在 RxJS 中，我们可以使用 RxDB 和 d3 等库来处理大量数据集合，并进行数据的可视化以及自定义操作符的编写。以下示例演示了如何使用 d3 库来可视化大量数据集合。
// javascriptcn.com 代码示例
import * as d3 from 'd3';
import { interval } from 'rxjs';
import { map } from 'rxjs/operators';
const container = document.getElementById('container')!;
const data$ = interval(1000).pipe(
map(() => d3.range(1000).map(d3.randomUniform(1, 100)))
);
data$.subscribe(values => {
const selection = d3.select(container);
const bars = selection.selectAll('rect').data(values);
bars.enter().append('rect')
.attr('x', 0)
.attr('y', (d, i) => i * 10)
.attr('height', 5)
.attr('width', d => d)
.style('fill', 'steelblue');
bars.transition()
.duration(500)
.attr('width', d => d);
bars.exit().remove();
});总结
本文从 RxJS 基础开始讲解可观察对象的创建和转换，随后通过一些例子加深了读者对 RxJS 的理解。进阶部分从多流处理、自定义可观察对象、异步操作等方面进行了讲解，通过更加复杂的场景来体现 RxJS 在应对各种应用场景中的优越性。在最后的高级部分，我们将 RxJS 应用到了更加高级的场景中，并提供了处理更加庞大的数据集合的方法，以及如何使用 d3 库来进行数据集合和可视化。通过本文的阅读和实践，相信大家对 RxJS 有了更加深入的理解，也能够使 RxJS 成为进一步完善前端技术架构的有利工具。
来源：JavaScript中文网 ，转载请注明来源 本文地址：https://www.javascriptcn.com/post/654337bb7d4982a6ebcdbfbe