我每天都在使用的 10 个 RxJS 运算符

更新日期: 2023-05-12阅读: 1.1k标签: rxjs

作为一名 Angular 开发人员,您可能会发现以下 RxJS 运算符在您的日常开发中很有用:


1、map():

此运算符用于转换可观察对象发出的值。它以一个函数作为参数,它接收发出的值作为输入并返回转换后的输出。返回的输出作为可观察对象的新值发出。例如,您可以使用 map() 将用户对象流转换为用户名流:

import { map } from 'rxjs/operators';
import { UserService } from './user.service';

constructor(private userService: UserService) {}

this.userService.getUsers().pipe(
map(users => users.map(user => user.name))
).subscribe(names => console.log(names));


2、filter():

此运算符用于根据条件过滤掉 observable 发出的值。它以一个函数作为参数,它接收发出的值作为输入并返回一个布尔值。 

如果布尔值为 true,则发出该值,否则将其过滤掉。例如,您可以使用 filter() 只保留整数流中的偶数:

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

of(1, 2, 3, 4, 5, 6).pipe(
filter(num => num % 2 === 0)
).subscribe(evenNum => console.log(evenNum));


3、switchMap():

每当源可观察对象发出新值时,此运算符用于切换到新的可观察对象。它以一个函数作为参数,它接收发出的值作为输入并返回一个新的可观察值。 

最新的 observable 发出的值由返回的 observable 发出。 

例如,您可以使用 switchMap() 根据另一个可观察对象发出的 ID 获取用户数据:

import { switchMap } from 'rxjs/operators';
import { UserService } from './user.service';

constructor(private userService: UserService) {}

this.userId$.pipe(
switchMap(userId => this.userService.getUser(userId))
).subscribe(user => console.log(user));


4、catchError():

此运算符用于处理可观察对象抛出的错误。它以一个函数作为参数,它接收错误作为输入并返回一个新的可观察对象。 

返回的可观察对象可以发出新值、完成或抛出另一个错误。例如,您可以使用 catchError() 重试失败的 HTTP 请求:

import { catchError } from 'rxjs/operators';
import { HttpClient } from '@angular/common/http';

constructor(private http: HttpClient) {}

this.http.get('/api/data').pipe(
catchError(() => this.http.get('/api/fallbackData'))
).subscribe(data => console.log(data));


5、tap():

此运算符用于在不修改可观察对象的发射值的情况下执行副作用。它以一个函数作为参数,它接收发出的值作为输入并可以执行任何副作用。 

返回值被忽略,原始值不变。例如,您可以使用 tap() 来记录可观察对象发出的值:

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

of(1, 2, 3).pipe(
tap(value => console.log(`Emitting value: ${value}`))
).subscribe();


6、take():

此运算符用于限制可观察对象发出的值的数量。这是一个例子:

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

of(1, 2, 3, 4, 5).pipe(
take(3)
).subscribe(value => console.log(value)); // output: 1, 2, 3

在此示例中,take(3) 用于将 of(1, 2, 3, 4, 5) 发出的数字流限制为前三个值。


7、debounceTime():

此运算符用于按指定的时间量对值流进行去抖动。这是一个例子:

import { debounceTime } from 'rxjs/operators';
import { fromEvent } from 'rxjs';

const input = document.getElementById('input');

fromEvent(input, 'input').pipe(
debounceTime(500)
).subscribe(event => console.log(input.value));

在此示例中,debounceTime(500) 用于将来自输入元素的输入事件流去抖动 500 毫秒。这意味着如果用户快速连续键入,则只会记录 500 毫秒不活动后的输入值。


8、distinctUntilChanged():

此运算符用于仅从值流中发出不同的值。这是一个例子:

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

of(1, 2, 2, 3, 3, 3, 4).pipe(
distinctUntilChanged()
).subscribe(value => console.log(value)); // output: 1, 2, 3, 4

在此示例中,distinctUntilChanged() 用于仅从 of(1, 2, 2, 3, 3, 3, 4) 发出的数字流中发出不同的值。


9、merge():

此运算符用于将多个可观察对象合并为一个可观察对象。这是一个例子:

import { merge } from 'rxjs';
import { interval } from 'rxjs';

const obs1 = interval(1000);
const obs2 = interval(2000);

merge(obs1, obs2).subscribe(value => console.log(value));

在此示例中,merge(obs1, obs2) 用于将两个可观察对象 obs1 和 obs2 合并为一个可观察对象,该可观察对象从两个可观察对象发出值。


10、combineLatest():

该运算符将多个可观察对象发出的最新值组合到一个数组中。它仅在其中一个源可观察对象发出新值时发出新数组。

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

const observable1 = timer(0, 1000).pipe(map(x => x + ' seconds'));
const observable2 = timer(500, 500).pipe(map(x => x + ' half seconds'));

combineLatest([observable1, observable2]).subscribe(combinedValues => {
const [value1, value2] = combinedValues;
console.log(`observable1: ${value1}, observable2: ${value2}`);
});

在这个例子中,我们使用 timer() 函数创建了两个可观察对象。第一个 observable 每秒发出一个值,第二个 observable 每半秒发出一个值。我们使用 map() 运算符将发出的值转换为字符串。

然后我们将一组可观察对象传递给 combineLatest() 函数,该函数返回一个新的可观察对象,该可观察对象发出一个包含每个源可观察对象发出的最新值的数组。我们订阅这个新的 observable,并在发出新的组合值数组时记录每个源 observable 发出的值。

这段代码的输出看起来像这样:

observable1: 0 seconds, observable2: 0.5 half seconds
observable1: 1 seconds, observable2: 1 half seconds
observable1: 2 seconds, observable2: 1.5 half seconds
observable1: 3 seconds, observable2: 2 half seconds
observable1: 4 seconds, observable2: 2.5 half seconds

如您所见,只要其中一个源可观察对象发出新值, combineLatest() 运算符就会发出一个新的组合值数组。

在这种情况下,第一个 observable 每秒发出一个新值,第二个 observable 每半秒发出一个新值,因此组合的 observable 每半秒发出一个新值。

我希望,你发现这篇文章很有用。如果您有任何问题或建议,请在留言区给我留言。您的反馈将帮助我们变得更好。

英文 | https://medium.com/front-end-weekly/10-rxjs-operators-which-i-use-daily-as-an-angular-developer-c902edb6f05f
翻译:web前端开发

链接: https://www.fly63.com/article/detial/12480

rxjs入门指南

在复杂的,频繁的异步请求场景,使用rxjs。在依赖的多个异步数据,决定渲染的情景,使用rxjs。总之:在前台频繁的、大量的、和后台数据交互的复杂项目里面,使用rxjs(web端,iOS,android端等,客户端都可考虑使用)

Rxjs常用operators

本文使用的是angular6内置的rxjs,concat:通过顺序地发出多个 Observables 的值将它们连接起来,一个接一个的。count计算源的发送数量,并当源完成时发出该数值。

RxJS主题(Subject)

什么是主题?RxJS 主题就是一个特性类型的 Observable 对象,它允许值多路广播给观察者(Observers)。当一个简单的 Observable 是单播的(每个订阅的观察者它们自己都依赖 Observable 的执行)时候,主题(Subjects)就是多播的。

RxJS中Operators

RxJS 的操作符(operators)是最有用的,尽管 Observable 是最基本的。操作符最基本的部分(pieces)就是以申明的方式允许复杂的异步代码组合简化。操作符是函数。这里有两种操作符:

RxJS调度器(Scheduler)

调度是数据结构。它知道怎样在优先级或其他标准去存储和排队运行的任务,调度器是一个执行上下文。它表示任务在何时何地执行(例如,立即或是在回调机制中如 setTimeout 或 process.nextTick,又或是动画框架)

80 行代码实现简易 RxJS

RxJS 是一个响应式的库,它接收从事件源发出的一个个事件,经过处理管道的层层处理之后,传入最终的接收者,这个处理管道是由操作符组成的,开发者只需要选择和组合操作符就能完成各种异步逻辑,极大简化了异步编程

内容以共享、参考、研究为目的,不存在任何商业目的。其版权属原作者所有,如有侵权或违规,请与小编联系!情况属实本人将予以删除!