使用 RxJS 掌握异步 JavaScript

**TL;DR:** RxJS 是一个功能强大的 JavaScript 库，用于管理异步数据流，简化事件处理和 API 交互等复杂操作。它使用可观察对象来表示数据流，使用运算符来转换和操作数据流，使用订阅来对发出的值做出反应。

在现代 JavaScript 开发的动态环境中，高效处理异步操作至关重要。RxJS（JavaScript 的反应式扩展）是一个功能强大的库，旨在解决这一挑战，使开发人员能够优雅而精确地管理异步数据流。

什么是 RxJS？

RxJS 是一个库，它允许开发人员使用可观察序列来处理异步和基于事件的程序。它的核心是反应式编程的概念，这是一种以数据流和变化传播为中心的范式。这种方法在处理用户界面时尤其有用，因为各种事件（如用户交互、数据获取和应用状态变化）都可以被视为流动的数据流。反应式编程鼓励开发人员声明当这些流中发生变化时应用应该如何表现，而不是直接响应每个事件。

核心概念

要掌握 RxJS 的强大功能，必须了解其基本构成要素：

可观察对象：可观察对象是 RxJS 的核心，代表随时间发出值的数据源。它们可以从各种来源创建，包括事件、承诺和现有数据。将可观察对象视为数据流经的管道。

观察者：观察者是订阅可观察对象并定义如何对发出的值做出反应的对象。它充当侦听器，指示新数据到达时要采取什么操作。

订阅：订阅代表观察者与可观察对象之间的连接。它就像一份合同，允许观察者从可观察对象接收值。当您订阅可观察对象时，您将开始接收数据，直到您明确取消订阅。

运算符：运算符是纯函数，可用于转换、过滤和组合可观察对象。它们充当修饰符，对流经可观察对象流的数据进行整形和细化。它们提供了一种声明式的方式来操纵数据流，而无需修改原始源。

冷可观察对象与热可观察对象

RxJS 中的可观察对象可分为冷观察对象和热观察对象：

冷可观察对象是按需创建的，并且仅在订阅时才开始发出值。每次新的订阅都会触发可观察对象的重新执行。例如，从 HTTP 请求创建的可观察对象被视为冷的，因为它仅在订阅者表示感兴趣时才会发出请求。

Hot observables exist independently of subscriptions and emit values regardless of whether anyone is listening. They represent an ongoing stream of data that is shared among all subscribers. Examples include mouse events or stock tickers, where the data stream continues regardless of the number of observers.

Let’s illustrate these concepts with simple examples.

Creating an Observable

import { Observable } from "rxjs";

const first5Numbers$ = new Observable((obs) => {
  console.log("hello!");
  for (let i = 0; i < 5; i++) {
    obs.next(i);
  }
  obs.complete();
});

// Logs nothing.
first5Numbers$.subscribe((n) => {
  console.log(n);
});

// Logs "hello!" followed by 0 1 2 3 4.

In this example, **first5Numbers$** is a cold observable that emits numbers 0 to 4. The **subscribe** method attaches an observer to the observable. The **next** method is used to emit values from the observable. The **complete** method signals the end of the stream.

Using an operator

import { interval } from "rxjs";
import { take } from "rxjs/operators";

const first5SpacedNumbers$ = interval(1000).pipe(take(5));

Here, we create an observable **first5SpacedNumbers$** that emits a value every second. The **take** operator is used to limit the stream to the first five emissions.

Why use RxJS?

RxJS shines in several scenarios:

Handling complex, asynchronous operations: RxJS provides a structured approach to manage intricate asynchronous flows, preventing callback hell and deeply nested promises. Its declarative nature allows you to express complex logic concisely, making your code more readable and maintainable.

Real-time applications: With its support for hot observables, RxJS excels in building real-time apps like chat apps, stock tickers, and collaborative editing tools.

Event handling: RxJS simplifies the handling of user interactions, DOM events, and other asynchronous events, providing a streamlined way to manage event propagation and responses.

RxJS vs. promises and async/await

While promises and async/await are valuable for handling single asynchronous operations, RxJS is geared toward managing streams of asynchronous events. Here’s a comparison:

Promises: Resolve with a single value and are primarily useful for one-time asynchronous tasks.

Async/await: Provide a more synchronous-looking syntax for working with promises but still focus on individual asynchronous operations.

RxJS: Handles multiple values over time, offering operators to transform, filter, and combine these values. It’s ideal for scenarios where data arrives continuously or in bursts.

Setting up RxJS

Installation

You can install RxJS in your project using npm or yarn:

npm install rxjs

yarn add rxjs

Alternatively, you can include RxJS via a CDN link in your HTML file.

Let’s create a simple observable and subscribe to it.

import { of } from "rxjs";

const myObservable$ = of(1, 2, 3);

myObservable$.subscribe((value) => {
  console.log(value); // Outputs: 1, 2, 3
});

In this example, we use the **of** operator to create an observable that emits the values 1, 2, and 3.

Operators in RxJS

Operators are the backbone of RxJS, providing a rich vocabulary to manipulate data streams. Here are some categories of operators:

Creation operators: Create observables from various sources, like of, from, interval, and fromEvent.

Transformation operators: Modify the emitted values, such as map, flatMap, switchMap, and scan.

Filtering operators: Selectively emit values based on criteria, like filter, distinctUntilChanged, and take.

Combination operators: Merge or combine multiple observables, such as merge, concat, zip, and combineLatest.

Real-world use cases

让我们探索一些现实世界中的关键操作符的例子：

map：转换可观察对象发出的值。例如，您可以使用 map 从 HTTP 响应中提取特定数据。

import { of } from "rxjs";
import { map } from "rxjs/operators";

const source$ = of({ name: "John", age: 30 });

source$.pipe(map((person) => person.name)).subscribe((name) => {
    console.log(name); // Outputs: "John"
});

filter：仅发出满足特定条件的值。例如，您可以过滤事件流以仅处理特定区域内的鼠标点击。

import { fromEvent } from "rxjs";
import { filter } from "rxjs/operators";

const clicks$ = fromEvent(document, "click").pipe(
  filter(event => event.clientX > 100)
);

合并：将多个可观察对象合并为一个流，并在所有来源到达时发出值。这对于处理来自不同来源的事件（如用户输入和服务器响应）非常有用。

import { merge } from "rxjs";

const clicks$ = fromEvent(document, "click");
const keypresses$ = fromEvent(document, "keypress");

merge(clicks$, keypresses$).subscribe((event) => {
    console.log(event); // Outputs both click and keypress events
});

switchMap：当源可观察对象发出一个值时，它会订阅一个新的内部可观察对象并取消前一个内部可观察对象。这对于由用户输入触发的 API 调用等场景非常有用，在这些场景中，您只关心最新的请求。

import { fromEvent } from "rxjs";
import { switchMap } from "rxjs/operators";

const searchInput = document.getElementById("search") as HTMLInputElement;
const searchTerm$ = fromEvent(searchInput, "input").pipe(
  map((event) => (event.target as HTMLInputElement).value),
  switchMap((term) => fetch(`/api/search?q=${term}`))
);

catchError：在可观察流中妥善处理错误。它允许您捕获错误、执行日志记录或重试等操作，并可选择返回新的可观察流以继续该流。

import { of } from 'rxjs';
import { catchError, map } from 'rxjs/operators';

const source$ = of(1, 2, 3, 4, 5).pipe(
  map((value) => {
    if (value === 4) {
      throw new Error('Error occurred!');
    }
    return value;
  }),
  catchError((error) => {
    console.error(error);
    return of(0); // Replace the error with 0
  })
);

RxJS 中的错误处理

RxJS 提供了强大的机制来管理可观察流中的错误。

重试：如果可观察对象发出错误，重试运算符将重新订阅源可观察对象，尝试从错误中恢复。您可以指定重试次数或根据错误类型应用重试逻辑。

catchError：如前所述，catchError 操作符允许您优雅地处理错误、记录错误、用默认值替换错误，甚至返回一个新的可观察对象以继续流。

finalize：无论可观察对象是否成功完成或发出错误，此运算符都会执行回调函数。它对于清理任务很有用，例如关闭资源或重置状态。

请参阅以下 RxJS 中的错误处理代码示例。

import { of } from "rxjs"
import { retry, catchError, finalize } from "rxjs/operators"

const source$ = of(1, 2, 3, 4, 5).pipe(
    map((value) => {
        if (value === 4) {
            throw new Error("Error occurred!")
        }
        return value
    }),
    retry(2), // Retry twice if an error occurs
    catchError((error) => {
        console.error(error)
        return of(0) // Replace the error with 0
    }),
    finalize(() => console.log("Observable finalized!"))
)

在此示例中，如果发生错误，可观察对象会尝试重试两次。如果所有重试都失败，则 **catchError** 运算符会处理错误。当可观察对象完成或出错时，**finalize** 运算符会记录一条消息。

实际应用

让我们看看 RxJS 如何应用于实际场景：

表单验证：RxJS 非常适合创建响应式表单，在用户输入时会实时进行验证。您可以使用可观察对象来监控输入变化、应用验证规则并提供即时反馈。

API 轮询：RxJS 简化了轮询机制的实现。您可以使用 interval 和 switchMap 等运算符定期从 API 获取数据，并妥善处理响应和错误。

实时聊天应用：RxJS 非常适合构建实时聊天应用。热门可观察对象可以表示消息流，而 map 和 filter 等运算符可用于处理和显示消息。

提示和最佳做法

为了在你的项目中有效地利用 RxJS：

分解：将复杂的逻辑分解为更小的、可管理的、可使用运算符进行组合的可观察对象。

错误处理：使用 catchError 和 retry 来妥善处理错误并增强应用程序的弹性。

取消订阅：当不再需要可观察对象时，取消订阅可观察对象可防止内存泄漏。考虑使用 takeUntil 或 Angular 中的异步管道等工具来简化订阅管理。

测试：利用 RxJS 测试实用程序（如 TestScheduler）来彻底测试您的可观察逻辑。

常见陷阱

过度使用 RxJS：RxJS 虽然功能强大，但如果使用不当，可能会增加复杂性。请坚持使用其优势真正有益的场景。

内存泄漏：忽略取消订阅可观察对象可能会导致内存泄漏。始终确保正确的订阅管理。

结论

感谢您阅读本博客！RxJS 提供了一种强大而优雅的方式来处理 JavaScript 应用中的异步数据流。其反应式编程模型与一组丰富的运算符相结合，使开发人员能够构建响应迅速、可扩展且可维护的应用程序。通过掌握可观察对象、观察者和运算符的概念，您可以充分发挥 RxJS 的潜力并提升您的 JavaScript 开发技能。它的学习曲线最初可能看起来很陡峭，但在代码清晰度、可维护性和效率方面的回报非常值得付出努力。

CLIS.CC