发布于 由admin

JS 响应式编程：RxJS 操作符组合与数据流管理

各位靓仔靓女们，欢迎来到今天的RxJS操作符组合与数据流管理讲座！今天咱们不整虚的，直接上干货，用最接地气的方式，把RxJS这玩意儿给它盘明白了！

一、RxJS是啥玩意儿？为啥要学它？

简单来说，RxJS就是一个处理异步数据流的利器。想想咱们前端开发，各种异步操作：用户点击、网络请求、定时器……这些都像一条条流淌的数据，而RxJS就是帮你控制这些数据流的管道工！

为啥要学它？因为它可以让你的代码更简洁、更易维护、更具响应性。不再是回调地狱，不再是promise.then().then().then()…，而是优雅的数据流操作，想想都觉得舒服！

二、RxJS的核心概念：Observable、Observer、Subscription

这三个是RxJS的铁三角，理解了它们，RxJS就入门一半了！

  • Observable（可观察对象）： 它是数据流的源头，就像一个水龙头，源源不断地产生数据。
  • Observer（观察者）： 它是数据流的消费者，就像一个水桶，接收Observable发出的数据并进行处理。
  • Subscription（订阅）： 它建立了Observable和Observer之间的连接，就像水管，把水龙头和水桶连接起来。

用代码来解释一下：

// 创建一个Observable，每隔1秒发出一个数字
const observable = new Rx.Observable(subscriber => {
  let i = 0;
  const intervalId = setInterval(() => {
    subscriber.next(i++); // 发送数据
  }, 1000);

  // 返回一个函数，用于取消订阅
  return () => {
    clearInterval(intervalId); // 清除定时器
    console.log('Observable 被取消订阅了！');
  };
});

// 创建一个Observer，接收数据并打印
const observer = {
  next: (value) => { console.log('接收到数据：', value); },
  error: (err) => { console.error('发生错误：', err); },
  complete: () => { console.log('Observable 完成了！'); }
};

// 订阅Observable
const subscription = observable.subscribe(observer);

// 5秒后取消订阅
setTimeout(() => {
  subscription.unsubscribe(); // 取消订阅
}, 5000);

这段代码做了啥？

  1. 创建了一个Observable，它会每隔1秒发出一个递增的数字。
  2. 创建了一个Observer，它会接收Observable发出的数字，并在控制台打印出来。
  3. 使用subscribe()方法将Observable和Observer连接起来，建立了一个Subscription。
  4. 5秒后，使用unsubscribe()方法取消订阅，停止Observable发送数据。

三、RxJS操作符：管道工的工具箱

操作符是RxJS的灵魂，它们就像管道工的各种工具，可以对数据流进行各种各样的处理和转换。

RxJS的操作符有很多，咱们不可能一口气全部学会，所以今天挑一些最常用的、最实用的来讲。

1. 创建型操作符：创建数据流

  • of()： 将一系列值转换为Observable。

    const observable = Rx.of(1, 2, 3, 4, 5);
observable.subscribe(value => console.log(value)); // 输出：1, 2, 3, 4, 5

  • from()： 将数组、Promise、迭代器等转换为Observable。

    const array = [1, 2, 3, 4, 5];
const observable = Rx.from(array);
observable.subscribe(value => console.log(value)); // 输出：1, 2, 3, 4, 5

const promise = Promise.resolve('Hello, world!');
const observableFromPromise = Rx.from(promise);
observableFromPromise.subscribe(value => console.log(value)); // 输出：Hello, world!

  • interval()： 每隔指定的时间间隔发出一个数字。

    const observable = Rx.interval(1000); // 每隔1秒发出一个数字
observable.subscribe(value => console.log(value)); // 输出：0, 1, 2, 3, ...

  • timer()： 在指定的延迟后发出一个数字，或者每隔指定的时间间隔发出一个数字。

    const observable = Rx.timer(3000); // 3秒后发出一个数字 0
observable.subscribe(value => console.log(value));

const observableInterval = Rx.timer(1000, 2000); // 1秒后发出第一个数字 0，然后每隔2秒发出一个数字
observableInterval.subscribe(value => console.log(value)); // 输出：0, 1, 2, 3, ...

  • fromEvent()： 将DOM事件转换为Observable。

    const button = document.getElementById('myButton');
const observable = Rx.fromEvent(button, 'click');
observable.subscribe(event => console.log('按钮被点击了！', event));

2. 转换型操作符：改变数据流

  • map()： 对Observable发出的每个值应用一个函数，并将结果作为新的Observable发出。

    const observable = Rx.of(1, 2, 3, 4, 5);
const squaredObservable = observable.pipe(
  Rx.map(value => value * value) // 将每个值平方
);
squaredObservable.subscribe(value => console.log(value)); // 输出：1, 4, 9, 16, 25

  • filter()： 过滤掉Observable发出的不符合条件的值。

    const observable = Rx.of(1, 2, 3, 4, 5);
const evenObservable = observable.pipe(
  Rx.filter(value => value % 2 === 0) // 只保留偶数
);
evenObservable.subscribe(value => console.log(value)); // 输出：2, 4

  • scan()： 类似于数组的reduce()方法，对Observable发出的每个值进行累积计算。

    const observable = Rx.of(1, 2, 3, 4, 5);
const sumObservable = observable.pipe(
  Rx.scan((accumulator, value) => accumulator + value, 0) // 累加每个值，初始值为0
);
sumObservable.subscribe(value => console.log(value)); // 输出：1, 3, 6, 10, 15

  • pluck()： 从Observable发出的每个对象中提取指定的属性值。

    const observable = Rx.of(
  { name: 'Alice', age: 25 },
  { name: 'Bob', age: 30 },
  { name: 'Charlie', age: 35 }
);
const nameObservable = observable.pipe(
  Rx.pluck('name') // 提取每个对象的name属性
);
nameObservable.subscribe(value => console.log(value)); // 输出：Alice, Bob, Charlie

3. 过滤型操作符：控制数据流

  • take()： 只从Observable中获取指定数量的值。

    const observable = Rx.interval(1000); // 每隔1秒发出一个数字
const limitedObservable = observable.pipe(
  Rx.take(5) // 只获取前5个值
);
limitedObservable.subscribe(value => console.log(value)); // 输出：0, 1, 2, 3, 4

  • takeUntil()： 从Observable中获取值，直到另一个Observable发出值时停止。

    const observable = Rx.interval(1000); // 每隔1秒发出一个数字
const stopObservable = Rx.timer(5000); // 5秒后发出一个数字
const limitedObservable = observable.pipe(
  Rx.takeUntil(stopObservable) // 直到stopObservable发出值时停止
);
limitedObservable.subscribe(value => console.log(value)); // 输出：0, 1, 2, 3, 4

  • debounceTime()： 在指定的时间间隔内，只发出Observable发出的最后一个值。常用于处理搜索框输入，防止频繁请求。

    const input = document.getElementById('myInput');
const observable = Rx.fromEvent(input, 'keyup').pipe(
  Rx.map(event => event.target.value), // 获取输入框的值
  Rx.debounceTime(500) // 500毫秒内只发出最后一个值
);
observable.subscribe(value => console.log('搜索：', value));

  • distinctUntilChanged()： 只发出与上一个值不同的值。

    const observable = Rx.of(1, 1, 2, 2, 3, 3, 4, 4, 5, 5);
const distinctObservable = observable.pipe(
  Rx.distinctUntilChanged() // 只发出与上一个值不同的值
);
distinctObservable.subscribe(value => console.log(value)); // 输出：1, 2, 3, 4, 5

4. 组合型操作符：合并数据流

  • concat()： 将多个Observable按顺序连接起来。

    const observable1 = Rx.of(1, 2, 3);
const observable2 = Rx.of(4, 5, 6);
const combinedObservable = Rx.concat(observable1, observable2);
combinedObservable.subscribe(value => console.log(value)); // 输出：1, 2, 3, 4, 5, 6

  • merge()： 将多个Observable合并成一个Observable，并发地发出值。

    const observable1 = Rx.interval(1000).pipe(Rx.take(3)); // 每隔1秒发出一个数字，取前3个
const observable2 = Rx.interval(1500).pipe(Rx.take(3)); // 每隔1.5秒发出一个数字，取前3个
const combinedObservable = Rx.merge(observable1, observable2);
combinedObservable.subscribe(value => console.log(value)); // 并发输出：0, 0, 1, 1, 2, 2

  • zip()： 将多个Observable发出的值按顺序组合成一个数组，只有当所有Observable都发出值时才会发出。

    const observable1 = Rx.of(1, 2, 3);
const observable2 = Rx.of('a', 'b', 'c');
const combinedObservable = Rx.zip(observable1, observable2);
combinedObservable.subscribe(value => console.log(value)); // 输出：[1, "a"], [2, "b"], [3, "c"]

  • combineLatest()： 当任何一个Observable发出值时，将所有Observable的最新值组合成一个数组发出。

    const observable1 = Rx.interval(1000).pipe(Rx.take(3)); // 每隔1秒发出一个数字，取前3个
const observable2 = Rx.interval(1500).pipe(Rx.take(3)); // 每隔1.5秒发出一个数字，取前3个
const combinedObservable = Rx.combineLatest(observable1, observable2);
combinedObservable.subscribe(value => console.log(value));
// 输出：
// [0, 0] (1.5秒后)
// [1, 0] (2秒后)
// [1, 1] (3秒后)
// [2, 1] (3秒后)
// [2, 2] (4.5秒后)

四、操作符的组合：打造复杂的数据流

RxJS的强大之处在于可以将多个操作符组合起来，构建复杂的数据流处理逻辑。这就像搭积木一样，你可以根据自己的需求，选择不同的操作符，将它们组合在一起，实现各种各样的功能。

举个例子，假设我们需要实现一个搜索功能，用户在输入框中输入关键词，然后根据关键词进行搜索，并将搜索结果显示在页面上。我们可以使用以下操作符组合来实现：

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

const searchObservable = Rx.fromEvent(input, 'keyup').pipe(
  Rx.map(event => event.target.value), // 获取输入框的值
  Rx.debounceTime(300), // 300毫秒内只发出最后一个值
  Rx.distinctUntilChanged(), // 只发出与上一个值不同的值
  Rx.switchMap(searchTerm => { // 使用 switchMap 发起 HTTP 请求
    if (!searchTerm) {
      return Rx.of([]); // 如果搜索词为空，则返回空数组
    }
    return Rx.ajax.getJSON(`https://api.example.com/search?q=${searchTerm}`).pipe(
      Rx.catchError(error => {
        console.error('搜索出错：', error);
        return Rx.of([]); // 如果请求出错，则返回空数组
      })
    );
  })
);

searchObservable.subscribe(results => {
  searchResults.innerHTML = ''; // 清空搜索结果
  results.forEach(result => {
    const li = document.createElement('li');
    li.textContent = result.title;
    searchResults.appendChild(li);
  });
});

这段代码做了啥？

  1. 使用fromEvent()将输入框的keyup事件转换为Observable。
  2. 使用map()获取输入框的值。
  3. 使用debounceTime()防止频繁请求。
  4. 使用distinctUntilChanged()只发出与上一个值不同的值。
  5. 使用switchMap()发起HTTP请求，获取搜索结果。switchMap 会取消前一个未完成的请求，只保留最新的请求，避免请求堆积。
  6. 使用subscribe()将搜索结果显示在页面上。

五、错误处理：保证数据流的健壮性

在处理数据流时，难免会遇到错误。RxJS提供了一些操作符来处理错误，保证数据流的健壮性。

  • catchError()： 捕获Observable发出的错误，并返回一个新的Observable。

    const observable = Rx.of(1, 2, 3, 4, 5).pipe(
  Rx.map(value => {
    if (value === 3) {
      throw new Error('发生错误了！');
    }
    return value;
  }),
  Rx.catchError(error => {
    console.error('捕获到错误：', error);
    return Rx.of(6, 7, 8); // 返回一个新的Observable
  })
);
observable.subscribe(value => console.log(value)); // 输出：1, 2, 6, 7, 8

  • retry()： 在Observable发生错误时，自动重试指定次数。

    let retryCount = 0;
const observable = Rx.of(1, 2, 3, 4, 5).pipe(
  Rx.map(value => {
    retryCount++;
    if (retryCount < 3 && value === 3) {
      throw new Error('发生错误了！');
    }
    return value;
  }),
  Rx.retry(3) // 重试3次
);
observable.subscribe(value => console.log(value)); // 输出：1, 2, 1, 2, 1, 2, 4, 5

六、冷热Observable：理解数据流的共享

  • 冷Observable： 每次订阅都会创建一个新的Observable，每个订阅者都会收到完整的数据流。

    const observable = Rx.interval(1000); // 每隔1秒发出一个数字 (冷Observable)
const subscription1 = observable.subscribe(value => console.log('订阅者1：', value));
const subscription2 = observable.subscribe(value => console.log('订阅者2：', value));
// 订阅者1和订阅者2都会从0开始接收数据

  • 热Observable： 无论有多少个订阅者，Observable只会产生一个数据流，所有订阅者共享这个数据流。

    const observable = Rx.interval(1000).pipe(Rx.share()); // 每隔1秒发出一个数字 (热Observable)
const subscription1 = observable.subscribe(value => console.log('订阅者1：', value));
setTimeout(() => {
  const subscription2 = observable.subscribe(value => console.log('订阅者2：', value));
}, 3000);
// 订阅者1从0开始接收数据，订阅者2在3秒后加入，从当时Observable发出的值开始接收数据

RxJS常用操作符总结 (表格形式)

类别 操作符 描述 示例
创建型 of 将一系列值转换为 Observable Rx.of(1, 2, 3)
创建型 from 将数组、Promise、迭代器等转换为 Observable Rx.from([1, 2, 3])
创建型 interval 每隔指定的时间间隔发出一个数字 Rx.interval(1000)
创建型 timer 在指定的延迟后发出一个数字，或者每隔指定的时间间隔发出一个数字 Rx.timer(3000) 或 Rx.timer(1000, 2000)
创建型 fromEvent 将 DOM 事件转换为 Observable Rx.fromEvent(button, 'click')
转换型 map 对 Observable 发出的每个值应用一个函数，并将结果作为新的 Observable 发出 observable.pipe(Rx.map(value => value * 2))
转换型 filter 过滤掉 Observable 发出的不符合条件的值 observable.pipe(Rx.filter(value => value % 2 === 0))
转换型 scan 类似于数组的 reduce() 方法，对 Observable 发出的每个值进行累积计算 observable.pipe(Rx.scan((acc, val) => acc + val, 0))
转换型 pluck 从 Observable 发出的每个对象中提取指定的属性值 observable.pipe(Rx.pluck('name'))
过滤型 take 只从 Observable 中获取指定数量的值 observable.pipe(Rx.take(5))
过滤型 takeUntil 从 Observable 中获取值，直到另一个 Observable 发出值时停止 observable.pipe(Rx.takeUntil(stopObservable))
过滤型 debounceTime 在指定的时间间隔内，只发出 Observable 发出的最后一个值 observable.pipe(Rx.debounceTime(300))
过滤型 distinctUntilChanged 只发出与上一个值不同的值 observable.pipe(Rx.distinctUntilChanged())
组合型 concat 将多个 Observable 按顺序连接起来 Rx.concat(observable1, observable2)
组合型 merge 将多个 Observable 合并成一个 Observable，并发地发出值 Rx.merge(observable1, observable2)
组合型 zip 将多个 Observable 发出的值按顺序组合成一个数组，只有当所有 Observable 都发出值时才会发出 Rx.zip(observable1, observable2)
组合型 combineLatest 当任何一个 Observable 发出值时，将所有 Observable 的最新值组合成一个数组发出 Rx.combineLatest(observable1, observable2)
错误处理 catchError 捕获 Observable 发出的错误，并返回一个新的 Observable observable.pipe(Rx.catchError(error => Rx.of('Error!')))
错误处理 retry 在 Observable 发生错误时，自动重试指定次数 observable.pipe(Rx.retry(3))
多播 share 将冷 Observable 转换为热 Observable，允许多个订阅者共享同一个数据流 observable.pipe(Rx.share())
多播 shareReplay 类似于 share，但会缓存最近发出的指定数量的值，并在新的订阅者加入时立即发出这些值 observable.pipe(Rx.shareReplay(1)) // 缓存最近的一个值
多播 shareReplay({bufferSize: 1, refCount: true}) 缓存bufferSize个值，当最后一个订阅者退订时，会自动退订底层source的Observable，从而释放资源，refCount:true是关键，否则会造成内存泄露 observable.pipe(Rx.shareReplay({bufferSize: 1, refCount: true}))

七、实际应用场景：RxJS大显身手

  • 处理用户输入： 自动补全、搜索建议、表单验证。
  • 处理网络请求： 数据轮询、请求合并、错误重试。
  • 处理WebSocket消息： 实时数据更新、聊天室。
  • 处理动画： 创建流畅的动画效果。
  • 状态管理： Redux、NgRx等状态管理库都使用了RxJS。

八、写在最后：RxJS的学习之路

RxJS的学习曲线可能有点陡峭，但只要你坚持下去，掌握了它的核心概念和常用操作符，就能感受到它的强大和优雅。

记住，学习RxJS最好的方法就是实践！多写代码，多做项目，遇到问题多查资料、多交流。

好了，今天的讲座就到这里，希望对大家有所帮助！有问题随时提问，咱们一起进步！

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注