RxJS 与 Redux Observable 入门指南

RxJS 和 Redux Observable 是当前前端领域非常流行的函数式编程库，它们提供了很好的处理事件流和异步操作的能力，尤其在 Redux 状态管理中的应用，能够为我们带来更好的可维护性、可测试性和可扩展性，下面我们来详细介绍一下这两个库的基本使用和实践。

RxJS

RxJS 是 ReactiveX 在 JavaScript 中的实现，它的核心概念是 Observable 和 Observer，它们分别对应数据流的生产者和消费者，而 RxJS 则提供了一系列操作符（operators）可以对 Observable 进行转换、过滤、合并等操作。

基本使用

首先，我们需要使用 npm 安装 RxJS：

npm install rxjs

然后，在项目中引入 RxJS 依赖：

import { Observable } from 'rxjs';

接下来，我们可以创建一个 Observable，它可以发出三个数字：

const observable = new Observable(observer => {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();
});

其中，observer 是一个观察者对象，它有三个方法 next、error、complete 分别表示发送数据、错误和完成事件。在上述例子中，我们就依次发出了 1、2、3 三个数字，最后通过 complete 方法表示数据流结束。

现在，我们可以订阅这个 Observable 以获取它发出的数据：

observable.subscribe({
  next: value => console.log(value),
  complete: () => console.log('Complete'),
});

在上面的代码中，我们使用 subscribe 方法订阅了 Observable，当发出新的数据时，会触发 next 回调函数，在本例中会连续输出 1、2、3 三个数字。当 Observable 结束时，会触发 complete 回调函数，这里会输出 'Complete'。

操作符使用

除了上述基本用法，RxJS 还提供了大量丰富的操作符，下面列举一些常用的操作符及其用法：

• map：对 Observable 中的每个元素应用一个指定的变换函数。

  const source = from([1, 2, 3]);
  const result = source.pipe(
    map(value => value * 2),
  );
  result.subscribe(value => console.log(value)); // 2, 4, 6

• filter：只发出 Observable 中满足条件的元素。

  const source = from([1, 2, 3, 4]);
  const result = source.pipe(
    filter(value => value % 2 === 0),
  );
  result.subscribe(value => console.log(value)); // 2, 4

• debounceTime：防抖，只在 Observable 中间隔一定时间后才发出最后一个元素。

  const source = fromEvent(document, 'click');
  const result = source.pipe(
    debounceTime(1000),
  );
  result.subscribe(event => console.log(event)); // 点击事件，只在一秒内没有再次点击时输出

• merge：合并多个 Observables，按顺序将它们发出的元素合并成一个 Observable 。

  const source1 = of(1, 2);
  const source2 = of(10, 20);
  const result = merge(source1, source2);
  result.subscribe(value => console.log(value)); // 1, 2, 10, 20

更多操作符的使用方法可以参考 RxJS 官方文档。

Redux Observable

Redux Observable 则是基于 RxJS 创建的一个中间件，它能够让我们将“动作”和“副作用”分离开来，使得我们能够专注于处理我们的应用程序状态的更新，而不去担心异步减速的问题。

基本使用

首先，我们需要使用 npm 安装 Redux 和 Redux Observable：

npm install redux redux-observable rxjs

Redux Observable 的操作流程如下：

1. 创建一个 epic，它是一个能够处理异步操作转化为 action 的函数。
2. 将其作为 Redux 中间件传递给 createStore 中。
3. 响应这些 actions。

我们来看一下这个过程是如何实现的：

// javascriptcn.com 代码示例
import { applyMiddleware, createStore } from 'redux';
import { createEpicMiddleware } from 'redux-observable';
import { of } from 'rxjs';
import { mergeMap, map, delay } from 'rxjs/operators';

// 应用中间件
const epicMiddleware = createEpicMiddleware();
const store = createStore(rootReducer, applyMiddleware(epicMiddleware));

// 定义 epic
const pingEpic = action$ => action$.pipe(
  ofType('PING'),
  delay(1000),
  map(() => ({ type: 'PONG' })),
);

// 将 epic 作为中间件传递给 createStore
epicMiddleware.run(pingEpic);

在上述例子中，我们首先创建了一个名为 pingEpic 的 epic，它的作用是在接收到 PING action 后等待 1 秒钟，然后发出 PONG action。接下来，我们将这个 epic 作为 Redux 中间件传递给 createStore 方法，最后使用 epicMiddleware.run 方法来运行我们的 epic。

现在，我们我们 dispatch 一个 PING action，

store.dispatch({ type: 'PING' });

就可以在 1 秒后收到一个 PONG action。

同时处理多个 action

有时需要同时处理多个 action，这时可以使用 mergeMap 操作符，它可以将多个 action 压缩成一个 Observable，然后在异步完成后将它们解压。

// javascriptcn.com 代码示例
const fetchUserEpic = action$ => action$.pipe(
  ofType('FETCH_USER'),
  mergeMap(action =>
    ajax.getJSON(`/users/${action.payload}`).pipe(
      map(response => ({ type: 'FETCH_USER_SUCCESS', payload: response })),
      catchError(error => of({ type: 'FETCH_USER_FAILURE', payload: error })),
    ),
  ),
);

在上述例子中，需要处理两个 action：FETCH_USER 和 FETCH_USER_SUCCESS，我们使用 mergeMap 操作符来将这两个 action 压缩成一个 Observable，然后对异步请求结果进行处理，成功时发出 FETCH_USER_SUCCESS，失败时发出 FETCH_USER_FAILURE。

操作符使用

除了上述基本使用，Redux Observable 还提供了一些自己的操作符，下面列举一些常用的操作符及其用法：

• ofType：只处理指定类型的 action

  import { ofType } from 'redux-observable';
  
  const pingEpic = action$ => action$.pipe(
    ofType('PING'),
    delay(1000),
    map(() => ({ type: 'PONG' })),
  );

• delay：延迟操作

  // javascriptcn.com 代码示例
  const fetchUserEpic = action$ => action$.pipe(
    ofType('FETCH_USER'),
    delay(1000),
    mergeMap(action =>
      ajax.getJSON(`/users/${action.payload}`).pipe(
        map(response => ({ type: 'FETCH_USER_SUCCESS', payload: response })),
        catchError(error => of({ type: 'FETCH_USER_FAILURE', payload: error })),
      ),
    ),
  );

• debounceTime：防抖

  // javascriptcn.com 代码示例
  const fetchUserEpic = action$ => action$.pipe(
    ofType('FETCH_USER'),
    debounceTime(500),
    mergeMap(action =>
      ajax.getJSON(`/users/${action.payload}`).pipe(
        map(response => ({ type: 'FETCH_USER_SUCCESS', payload: response })),
        catchError(error => of({ type: 'FETCH_USER_FAILURE', payload: error })),
      ),
    ),
  );

更多操作符的使用方法可以参考 Redux Observable 官方文档。

总结

本文通过介绍 RxJS 和 Redux Observable 的基本概念和使用方法，让我们能够更好地掌握函数式编程在前端开发中的应用，这将为我们处理异步操作、状态管理和事件流等提供很大的帮助。同时，我们也了解了一些常用的操作符的用法，可以更加灵活地应对一些实际问题。在实际开发中，我们可以根据应用的情况来选择使用其中的一个或多个库，以便更好地满足需求。

来源：JavaScript中文网 ，转载请注明来源 本文地址：https://www.javascriptcn.com/post/653e5c917d4982a6eb7dfc7a