RxJS 核心原理-操作符（源码实现）

news2024/10/6 16:15:59

本文将深入探讨RXJS操作符的核心原理，并介绍一些常见的操作符、应用场景以及相应的代码示例。通过理解RXJS的响应式编程思想，您将能够更好地应用它来处理异步数据流和事件流。

建议读者在阅读本文之前，先参考我的另一篇文章《深入浅出 RxJS 核心原理（响应式编程篇）》，这篇文章将为您提供关于RXJS的基本概念和原理的基础知识。通过结合这两篇文章，能够更好地理解本文的内容。

如果您觉得本文对您有帮助或有价值，请不吝点赞支持。同时，如果您有任何疑问、建议或意见，欢迎在评论区留言，我将根据反馈进行修改和完善。非常感谢您的阅读和支持！

RXJS操作符是做什么用的？？

RxJS 中的操作符（Operator）是用来对 Observable 进行转换、过滤、合并等操作的函数。操作符可以对 Observable 的数据流进行处理，生成新的 Observable。

操作符也可以按需组合使用，构建出复杂的数据处理流程。通过链式调用多个操作符，可以对数据流进行连续的转换和处理。每个操作符都接收上一个操作符处理后的结果，并返回一个新的Observable对象，从而实现了数据的连续处理和转换。

RXJS操作符的实现原理

操作符函数的定义：每个操作符都是一个函数，接受一个 Observable 作为输入，并返回一个新的 Observable 作为输出。
创建新的 Observable：在操作符函数内部，会创建一个新的 Observable 对象。这可以通过 Observable 的构造函数、静态创建方法（如Observable.create()或Observable.from()）来实现。
订阅源 Observable：在新创建的Observable上调用源 Observable 的subscribe方法，将观察者（Observer）订阅到源 Observable 上。这样可以监听源 Observable 的数据流，并在接收到数据时进行处理。
处理数据流：在订阅源 Observable 后，操作符函数会根据需要对数据进行转换、过滤、合并等操作。这通常涉及到对源 Observable 的数据进行处理，并将处理后的数据传递给观察者。这一步通常会使用观察者的next方法来发送处理后的数据。
处理错误和完成事件：操作符函数还需要考虑源 Observable 发送错误和完成事件的情况。在接收到错误事件时，操作符函数可以选择通过观察者的error方法发送错误通知。在接收到完成事件时，操作符函数可以选择通过观察者的complete方法发送完成通知。
返回新的 Observable：操作符函数最后会返回新创建的 Observable 对象，作为操作符的输出。这样，可以通过链式调用多个操作符来对 Observable 进行复杂的数据流处理。

画一个流程图

下面是一个简单的操作符map的实现原理示例，它用于对 Observable 中的每个数据进行映射转换：

function map(sourceObservable, transformFn) {
  return new Observable(observer => {
    const subscription = sourceObservable.subscribe({
      next: value => observer.next(transformFn(value)),
      error: error => observer.error(error),
      complete: () => observer.complete()
    });

    return {
      unsubscribe: () => subscription.unsubscribe()
    };
  });
}

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

const mappedObservable = map(numbersObservable, value => value * 2);

mappedObservable.subscribe({
  next: value => console.log(value)
});

在上面的示例中，map操作符接受一个源Observable sourceObservable和一个转换函数 transformFn作为参数。它返回一个新的 Observable，该 Observable 在订阅时会执行上面的实现原理步骤。

RXJS常见的操作符及应用场景

以下是一些常见的操作符、它们的应用场景以及相应的代码示例：

map: 用于对 Observable 中的每个数据进行映射转换。常用于对数据进行格式化、提取特定字段等操作。

const numbersObservable = of(1, 2, 3);

const mappedObservable = numbersObservable.pipe(
  map(value => value * 2)
);

mappedObservable.subscribe(value => console.log(value));
// 输出: 2, 4, 6

filter: 用于根据指定条件过滤 Observable 中的数据。常用于根据特定条件筛选数据流。

const numbersObservable = of(1, 2, 3, 4, 5);

const filteredObservable = numbersObservable.pipe(
  filter(value => value % 2 === 0)
);

filteredObservable.subscribe(value => console.log(value));
// 输出: 2, 4

merge: 用于将多个 Observable 合并成一个 Observable，按照时间顺序依次发出数据。常用于合并多个数据源或数据流。

const source1 = of('A', 'B', 'C');
const source2 = of('X', 'Y', 'Z');

const mergedObservable = merge(source1, source2);

mergedObservable.subscribe(value => console.log(value));
// 输出: A, B, C, X, Y, Z

debounceTime: 用于在指定时间间隔内，只发出 Observable 中最新的数据，忽略中间的数据。常用于处理输入框输入或搜索场景。

const inputObservable = fromEvent(inputElement, 'input');

const debouncedObservable = inputObservable.pipe(
  debounceTime(300)
);

debouncedObservable.subscribe(value => console.log(value));
// 在输入框停止输入300毫秒后，才会打印输入的值

switchMap: 用于在 Observable 发出数据时，取消前一个内部Observable 的订阅，并订阅新的内部 Observable。常用于处理基于先前请求结果的连续请求场景。

const searchObservable = fromEvent(searchButton, 'click').pipe(
  switchMap(() => {
    const query = inputElement.value;
    return searchApi(query);
  })
);

searchObservable.subscribe(results => console.log(results));
// 当点击搜索按钮时，取消前一个搜索请求，发送新的搜索请求，并打印结果