Published on

RxJS（三）：深入 RxJS 高级主题与实际应用

Authors
  • avatar
    Name
    青雲
    Twitter

在前面的两篇文章中，我们已经详细介绍了 RxJS 的基础知识、核心概念、操作符以及订阅管理。在这篇文章中，我们将进一步探讨 RxJS 的高级主题和实际应用，涵盖调试方法、内存泄漏防止、实践项目以及与其他库的集成。

调试RxJS

在开发和调试 RxJS 应用时，有一些有效的方法可以帮助我们更好地理解和调试数据流。本文将介绍两种主要的调试方法：marble diagrams 和 Chrome DevTools。此外，笔者在写这篇文章时搜索了一些 RxJS 相关的调试工具，发现 RxJS Devtools、RxJS Insights 等工具虽然存在，但很多已经不再维护。

Marble Diagrams

Marble diagrams 是一种视觉表示法，它用来描述 Observable 的行为。它们在文档、博客和 RxJS 课程中被大量使用，用来解释操作符和数据流的运行方式。Marble diagrams 用字符和符号来表示时间轴上的数据事件、错误和完成信号。

Marble Diagrams 基础

  • -：时间线的每一个刻度。
  • a、b、c：表示数据事件。
  • |：表示完成事件。
  • #：表示错误事件。
--a--b--c--|

上述示意表示一个 Observable 在时间上发出三次数据事件 a、b、c，然后完成。Marble diagrams 可以帮助我们直观地理解 RxJS 操作符的行为。

Marble Diagrams 示例

import { of } from 'rxjs'
import { map } from 'rxjs/operators'

// 创建一个 Observable 并使用 map 操作符
const numbers$ = of(1, 2, 3).pipe(map((x) => x * 10))

numbers$.subscribe((value) => console.log(value))
// 输出: 10, 20, 30

示意图：

Source Observable : --1--2--3--|
    map(x => x*10): --10--20--30--|

使用 Marble Diagrams 进行测试

RxJS 提供了 TestScheduler 类，使我们能够以编程方式创建和使用 Marble Diagrams 进行单元测试。

import { TestScheduler } from 'rxjs/testing';
import { map } from 'rxjs/operators';

const testScheduler = new TestScheduler((actual, expected) => {
  expect(actual).toEqual(expected);
});

testScheduler.run(({ cold, expectObservable }) => {
  const input = cold('--a--b--c--|', { a: 1, b: 2, c: 3 });
  const expected = '--y--z--w--|', { y: 10, z: 20, w: 30 };

  const result = input.pipe(map(x => x * 10));

  expectObservable(result).toBe(expected, { y: 10, z: 20, w: 30 });
});

Marble Diagrams 提供直观的视觉表示法，使我们能够轻松理解和测试 Observable 的行为，帮助我们调试和优化 RxJS 应用。

Chrome DevTools

尽管一些 RxJS 专用的调试插件（如 RxJS Devtools、RxJS Insights）已多年不维护，但标准的 Chrome DevTools 配合 RxJS 提供的一些操作符，依然可以进行有效的调试。

使用 tap 进行调试

tap 操作符允许你在 Observable 的每个值被发出时执行副作用操作，可以用来记录日志或调试。

import { of } from 'rxjs'
import { tap, map } from 'rxjs/operators'

const numbers$ = of(1, 2, 3).pipe(
  tap((x) => console.log('Before map:', x)),
  map((x) => x * 10),
  tap((x) => console.log('After map:', x))
)

numbers$.subscribe((value) => console.log('Final value:', value))
// 输出:
// Before map: 1
// After map: 10
// Final value: 10
// Before map: 2
// After map: 20
// Final value: 20
// Before map: 3
// After map: 30
// Final value: 30

通过这些日志输出，我们可以清晰地看到数据在流经每个操作符时的变化，帮助我们定位问题。

使用 console.time 和 console.timeEnd

利用 console.time 和 console.timeEnd 方法，可以测量和记录某段代码的执行时间。

import { range } from 'rxjs'
import { tap } from 'rxjs/operators'

console.time('range')
range(1, 10)
  .pipe(
    tap({
      next: (value) => console.log('Value:', value),
      complete: () => console.timeEnd('range'),
    })
  )
  .subscribe()
// 输出:
// Value: 1
// Value: 2
// ...
// Value: 10
// range: xx ms

RxJS Devtools 和 RxJS Insights 的调试原理

在 RxJS 调试工具的选择中，虽然一些工具已经不再维护，但了解其调试原理仍然可以帮助我们更好地理解如何进行有效的 RxJS 调试。以下是对 RxJS Devtools 和 RxJS Insights 调试原理的简单介绍：

RxJS Devtools

RxJS Devtools 是一个用于调试和监控 RxJS 代码的 Chrome 插件。它能够为开发者提供直观的界面来查看 Observable 的数据流和操作符的执行情况。RxJS Devtools提供对RxJS应用内部发送的数据流的深入洞察。这包括但不限于：

  • 查看应用内的Observable和Subject实例。
  • 观察数据流随时间的变化，包括订阅的添加和取消。
  • 监控数据流的性能，帮助识别可能的瓶颈或无效使用模式。

使用RxJS Devtools 需要集成 rxjs-spy 和 rxjs-spy-devtools-plugin。

npm install rxjs-spy rxjs-spy-devtools-plugin --save-dev

import { create } from 'rxjs-spy'
import { DevToolsPlugin } from 'rxjs-spy-devtools-plugin'

// 创建rxjs-spy实例
const spy = create()

// 添加rxjs-spy-devtools-plugin
spy.plug(new DevToolsPlugin())

RxJS Insights

RxJS Insights 是一个用于监视和调试RxJS代码的库。它能提供实时反馈，比如订阅的创建与销毁、数据流的传输路径以及潜在的性能问题。它可以：

  • 自动检测应用中的RxJS数据流。
  • 提供实时的性能分析报告，帮助开发者识别和优化性能瓶颈。
  • 可视化数据流，帮助开发者理解数据如何在应用中流动。
  • 提供丰富的调试信息，简化问题定位过程。

使用 RxJS Insights 需要引入相关依赖，并在应用入口处初始化，或者在构建工具里引入，以 webpack 为例：

npm install --save-dev @rxjs-insights/plugin-webpack5

// developement.webpack.config.js

const { RxjsInsightsPlugin } = require('@rxjs-insights/plugin-webpack5')

module.exports = {
  // (...)
  plugins: [new RxjsInsightsPlugin()],
  // (...)
}

防止内存泄露

内存泄漏是指程序在运行过程中无法释放已经不再使用的内存，导致内存消耗逐渐增加，从而影响程序性能和稳定性。在使用 RxJS 时，如果不正确地管理订阅关系，可能会导致内存泄漏。因此，采取适当的措施来防止内存泄漏是非常重要的。

适时取消订阅

确保在适当的时机取消订阅，可以防止内存泄漏和不必要的资源占用。我们可以通过显式取消订阅来管理内存。

import { interval, Subscription } from 'rxjs'

// 创建一个每秒发出值的 Observable
const source$ = interval(1000)

// 订阅 Observable
const subscription: Subscription = source$.subscribe((value) => console.log(value))

// 5秒后取消订阅
setTimeout(() => {
  subscription.unsubscribe()
  console.log('Unsubscribed')
}, 5000)
  1. 使用 interval 创建了一个每秒发出值的 Observable。
  2. 订阅 Observable，并在订阅关系上保存 Subscription 对象。
  3. 使用 setTimeout 方法，在 5 秒后调用 unsubscribe 方法取消订阅，停止接收值。
Source Observable:        --0--1--2--3--4--5--6--7--8--|
Unsubscribed at 5s:       --0--1--2--3--4--|

使用 takeUntil 操作符

takeUntil 操作符可以帮助我们管理订阅的生命周期，在特定条件下自动取消订阅。它会在另一个 Observable 发出时完成源 Observable，从而自动取消订阅。

import { fromEvent, interval } from 'rxjs'
import { takeUntil } from 'rxjs/operators'

// 创建一个点击事件 Observable
const clicks$ = fromEvent(document, 'click')

// 创建一个每秒发出值的 Observable
const source$ = interval(1000)

// 使用 takeUntil
const result$ = source$.pipe(takeUntil(clicks$))

result$.subscribe((value) => console.log(value))
  1. 使用 fromEvent 创建一个点击事件 Observable。
  2. 使用 interval 创建一个每秒发出值的 Observable。
  3. 使用 takeUntil 操作符，将点击事件 Observable 传递给 takeUntil。当点击事件发生时，源 Observable 将完成，并自动取消订阅，停止接收值。
clicks$:                --------c------|
interval$:              --0--1--2--3--4--5--6--7--|
takeUntil(clicks$):     --0--1--2--3|

实践项目

这一章通过实践项目，帮助我们深入理解和应用 RxJS 的概念和操作符，以处理实际开发中的各种任务。

应用实例

使用 RxJS 实现搜索建议

实时搜索建议功能在现代 Web 应用中非常常见。我们可以使用 RxJS 处理用户输入，并通过防抖动（debounce）和切换（switchMap）来实现流畅的搜索体验。

import { fromEvent, of } from 'rxjs'
import { debounceTime, map, switchMap } from 'rxjs/operators'

const searchInput = document.getElementById('search')

// 模拟 API 调用
function search(query) {
  return of(['Result 1', 'Result 2', 'Result 3']).pipe(
    map((results) => results.filter((result) => result.toLowerCase().includes(query.toLowerCase())))
  )
}

// 创建一个输入事件 Observable
fromEvent(searchInput, 'input')
  .pipe(
    debounceTime(300), // 防抖动
    map((event) => event.target.value),
    switchMap((query) => search(query)) // 切换到最新的搜索 Observable
  )
  .subscribe((results) => {
    // 渲染搜索结果
    const resultsContainer = document.getElementById('results')
    resultsContainer.innerHTML = results.map((result) => `<li>${result}</li>`).join('')
  })
  1. fromEvent 创建一个输入事件 Observable。
  2. debounceTime 防抖动操作符，使其在用户停止输入 300 毫秒后再发出事件。
  3. map 操作符将输入事件映射为输入值（query）。
  4. switchMap 操作符执行搜索并切换到最新的搜索结果 Observable，取消之前的搜索请求。
  5. 订阅并渲染搜索结果。

使用 RxJS 处理拖曳事件

拖曳（Drag and Drop）是常见的用户交互模式。我们可以使用 RxJS 处理鼠标事件来实现拖曳功能。

import { fromEvent } from 'rxjs'
import { switchMap, takeUntil, map } from 'rxjs/operators'

const box = document.getElementById('box')

const mouseDown$ = fromEvent(box, 'mousedown') // 鼠标按下事件
const mouseMove$ = fromEvent(document, 'mousemove') // 鼠标移动事件
const mouseUp$ = fromEvent(document, 'mouseup') // 鼠标放开事件

mouseDown$
  .pipe(
    switchMap((startEvent) =>
      mouseMove$.pipe(
        takeUntil(mouseUp$),
        map((moveEvent) => {
          // 计算元素的新位置
          return {
            left: moveEvent.clientX - startEvent.offsetX,
            top: moveEvent.clientY - startEvent.offsetY,
          }
        })
      )
    )
  )
  .subscribe((position) => {
    // 更新元素位置
    box.style.left = `${position.left}px`
    box.style.top = `${position.top}px`
  })
  1. fromEvent 创建鼠标事件 Observable。
  2. switchMap 将鼠标按下事件映射到新的鼠标移动事件 Observable。
  3. takeUntil 操作符在鼠标放开时取消订阅鼠标移动事件。
  4. map 操作符计算元素的新位置。
  5. 订阅并更新元素位置。

使用 RxJS 管理表单状态

表单状态管理是 Web 应用中的常见需求。通过 RxJS 可以高效管理多输入字段的联动状态。

import { fromEvent, combineLatest } from 'rxjs'
import { map } from 'rxjs/operators'

const nameInput = document.getElementById('name')
const ageInput = document.getElementById('age')

// 创建输入事件 Observable
const name$ = fromEvent(nameInput, 'input').pipe(map((event) => event.target.value))
const age$ = fromEvent(ageInput, 'input').pipe(map((event) => event.target.value))

// 使用 combineLatest 联合输入流
combineLatest([name$, age$]).subscribe(([name, age]) => {
  // 更新表单状态
  const status = document.getElementById('status')
  status.textContent = `Name: ${name}, Age: ${age}`
})
  1. fromEvent 为表单输入字段创建输入事件 Observable。
  2. map 操作符将输入事件映射为输入值。
  3. combineLatest 操作符将多个输入流合并为一个输出流，联动更新表单状态。

与其他库的集成

与 React 的集成

在 React 中，我们可以使用 RxJS 和 hooks 来处理数据流。

import React, { useEffect, useState } from 'react';
import { fromEvent } from 'rxjs';
import { map } from 'rxjs/operators';

const App = () => {
  const [text, setText] = useState('');

  useEffect(() => {
    const input = document.getElementById('input');
    const subscription = fromEvent(input, 'input').pipe(
      map(event => event.target.value)
    ).subscribe(value => setText(value));

    return () => subscription.unsubscribe();
  }, []);

  return (
    <div>
      <input id="input" type="text" />
      <p>{text}</p>
    </div>
  );
};

export default App;
  1. 使用 useEffect hook 处理 RxJS 数据流。
  2. 使用 RxJS 操作符处理输入事件。

与 Vue 的集成

在 Vue 中，可以使用 RxJS 和 VueRx 结合来实现响应式数据流管理。

import Vue from 'vue'
import VueRx from 'vue-rx'
import { Observable, of } from 'rxjs'
import { switchMap, map } from 'rxjs/operators'

Vue.use(VueRx)

new Vue({
  el: '#app',
  data: {
    query: '',
    results: [],
  },
  subscriptions() {
    return {
      results: this.$watchAsObservable('query').pipe(switchMap((query) => this.search(query))),
    }
  },
  methods: {
    search(query) {
      return of(['Result 1', 'Result 2', 'Result 3']).pipe(
        map((results) =>
          results.filter((result) => result.toLowerCase().includes(query.toLowerCase()))
        )
      )
    },
  },
  template: `
    <div>
      <input v-model="query" />
      <ul>
        <li v-for="result in results" :key="result">{{ result }}</li>
      </ul>
    </div>
  `,
})
  1. 使用 Vue 和 VueRx 结合处理响应式数据流。
  2. 使用 RxJS 操作符处理搜索建议逻辑。

复杂案例：聊天室应用

让我们考虑一个复杂的前端应用场景：一个实时聊天应用，其中涉及到：

  1. 用户在不同的聊天室之间切换。
  2. 对每个聊天室的消息进行实时监听，确保用户只接收当前所在聊天室的消息。
  3. 聊天室里的消息要支持实时更新和历史消息的拉取。
  4. 对网络请求失败进行重试处理。

普通方式实现

在不使用RxJS的情况下，实现这样的功能通常需要手动管理事件监听器、全局状态（例如，当前哪个聊天室是活跃的）、异步操作的取消与重启等。伪代码大概如下：

let currentRoomId = null
let messageListener = null

function switchRoom(newRoomId) {
  currentRoomId = newRoomId
  updateUIForRoomSwitch(newRoomId)
  if (messageListener) {
    // 假设removeMessageListener是取消监听的方法
    removeMessageListener(messageListener)
  }
  // 为新房间添加消息监听
  messageListener = addMessageListener(newRoomId, (messages) => {
    updateMessageUI(messages)
  })
}

function addMessageListener(roomId, callback) {
  // set up real-time message listener for the room
  // 假设这里我们通过WebSocket或其他方法来监听
  return {
    /* 返回一个监听器对象 */
  }
}

function removeMessageListener(listener) {
  // remove the listener
}

function updateUIForRoomSwitch(roomId) {
  // 更新UI到新房间
}

function updateMessageUI(messages) {
  // 实时更新聊天界面
}

这种实现方式有如下问题：

  1. 手动管理状态和监听器：你需要手动跟踪当前活跃的聊天室和对应的消息监听器。
  2. 复杂性和错误：在处理切换聊天室和清理过时的监听器时，代码很容易出错。
  3. 难以扩展和维护：随着功能的增加（如消息的发送、聊天室的搜索等），这种实现方式将变得越来越难以扩展和维护。

使用 RxJS 实现

实现步骤：

  1. 管理用户切换聊天室的事件流。
  2. 对当前聊天室的消息进行实时监听。
  3. 拉取历史消息并与实时消息流合并，确保用户看到的是最新的完整消息。
  4. 使用重试机制处理网络请求失败。
import { fromEvent, merge, of, EMPTY } from 'rxjs'
import { webSocket } from 'rxjs/webSocket'
import { switchMap, map, tap, retryWhen, takeUntil, delay, catchError } from 'rxjs/operators'

// 获取DOM元素
const chatContainer = document.getElementById('chat')
const roomSelect = document.getElementById('roomSelect')
const messageInput = document.getElementById('messageInput')
const sendButton = document.getElementById('sendButton')

// 处理聊天室切换事件
const roomChange$ = fromEvent(roomSelect, 'change').pipe(
  map((event) => (event.target as HTMLSelectElement).value)
)

// 创建 WebSocket 连接
const createWebSocket = (room: string) => {
  return webSocket(`ws://chat.example.com/rooms/${room}`).pipe(
    retryWhen((errors) =>
      errors.pipe(
        tap((err) => console.error('WebSocket error:', err)),
        delay(1000) // 重试延迟
      )
    )
  )
}

// 拉取历史消息
const fetchChatHistory = (room: string) => {
  return of([{ text: 'Historical message 1' }, { text: 'Historical message 2' }]) // 模拟API调用
}

// 管理聊天室消息流
const messages$ = roomChange$.pipe(
  switchMap((room) => {
    const chatWebSocket$ = createWebSocket(room)
    const chatHistory$ = fetchChatHistory(room)

    return merge(
      chatHistory$.pipe(map((message) => ({ ...message, historical: true }))),
      chatWebSocket$.pipe(
        catchError((error) => {
          console.error(`WebSocket error for room ${room}:`, error)
          return EMPTY
        })
      )
    )
  })
)

// 显示消息
const displayMessage = (message) => {
  const messageItem = document.createElement('li')
  messageItem.textContent = message.text + (message.historical ? ' (History)' : '')
  chatContainer.appendChild(messageItem)
}

messages$.subscribe((message) => {
  displayMessage(message)
})

// 处理发送消息
const sendMessage = (message) => {
  const room = roomSelect.value
  const chatWebSocket$ = createWebSocket(room)
  chatWebSocket$.next({ text: message })
}

fromEvent(sendButton, 'click')
  .pipe(
    tap(() => {
      const message = messageInput.value
      sendMessage(message)
      messageInput.value = ''
    })
  )
  .subscribe()
  1. 切换聊天室：
    1. 使用 fromEvent 创建一个聊天室切换事件 Observable。
    2. 使用 switchMap 在用户切换聊天室时取消上一个聊天室的订阅，切换到新的聊天室数据流。
  2. 实时监听消息：
    1. 使用 webSocket 操作符建立 WebSocket 连接，监听实时消息。
    2. 使用 retryWhen 处理 WebSocket 连接中断和重试机制。
  3. 拉取历史消息：
    1. 模拟 API 调用，通过 of 操作符创建一个拉取历史消息的 Observable。
    2. 使用 merge 将历史消息和实时消息流合并，确保用户看到最新的完整消息。
  4. 错误处理：
    1. 使用 catchError 捕获并处理 WebSocket 连接错误，确保应用的稳定性。

通过 RxJS，实现了一个多聊天室实时聊天应用，简化了代码逻辑，增强了可维护性和灵活性。具体的优势包括：

  • 代码简洁可读：通过使用 RxJS 操作符，减少了嵌套回调和复杂的状态管理，使代码更简洁、容易维护。
  • 灵活的流操作：使用 switchMap、merge 等操作符灵活地处理聊天室切换和数据流的合并。
  • 自动重试机制：通过 retryWhen 实现自动重试，增强了应用的可靠性和用户体验。
  • 资源管理：管理多聊天室的连接和消息流，确保资源的高效使用，防止内存泄漏。

性能优化与最佳实践

在使用 RxJS 开发复杂的应用程序时，性能优化和最佳实践至关重要。通过有效管理数据流和操作符，我们可以提升应用性能，确保代码的高效性和可维护性。

Lazy Loading 数据

Lazy Loading 数据意味着仅在需要时才加载和处理数据。这种方式可以节省资源，避免不必要的计算和数据传输。

使用 defer 操作符

defer 操作符延迟 Observable 的创建，直到有订阅者订阅时才开始执行。这在实现懒加载场景时特别有用。

import { defer, of } from 'rxjs'

const lazyObservable$ = defer(() => {
  return of('Lazy Loaded Data')
})

lazyObservable$.subscribe((value) => console.log(value)) // 输出: Lazy Loaded Data
  1. defer 操作符延迟了 Observable 的创建，直到有订阅者才开始执行。
  2. 这样确保只有在需要时才加载数据，节省了资源。

限制并发请求

在处理多个异步请求时，限制并发请求的数量可以避免系统过载，并提高应用的性能。

使用 mergeMap 和 concatMap

mergeMap 和 concatMap 操作符可以帮助我们控制并发请求的数量。

import { from, of } from 'rxjs'
import { mergeMap, concatMap, delay } from 'rxjs/operators'

const urls = ['url1', 'url2', 'url3']

// 模拟异步请求
const fetchData = (url) => of(`Data from ${url}`).pipe(delay(1000))

// 使用 mergeMap 并发处理多个请求
from(urls)
  .pipe(mergeMap((url) => fetchData(url)))
  .subscribe((data) => console.log('mergeMap:', data))

// 使用 concatMap 顺序处理多个请求
from(urls)
  .pipe(concatMap((url) => fetchData(url)))
  .subscribe((data) => console.log('concatMap:', data))
  1. mergeMap 可以并发地处理多个请求，适用于并行处理场景。
  2. concatMap 按顺序处理请求，适用于需要按顺序处理的场景。

避免嵌套订阅

嵌套订阅会导致代码复杂性增加，并且难以维护。使用高阶操作符可以避免嵌套订阅问题。

使用 switchMap

switchMap 可以在新的 Observable 发出值时取消之前的 Observable，简化嵌套处理逻辑。

import { fromEvent, interval } from 'rxjs'
import { switchMap } from 'rxjs/operators'

const button = document.getElementById('startButton')

// 使用 switchMap 处理嵌套订阅问题
fromEvent(button, 'click')
  .pipe(switchMap(() => interval(1000)))
  .subscribe((value) => console.log(value)) // 每次点击按钮时，开始新的计时器并取消上一个计时器
  1. switchMap 可以在新的数据源发出时取消之前的数据源，避免嵌套订阅。
  2. 它适用于处理频繁切换的数据流场景。

确保 Observable 完成

在某些场景中，确保 Observable 正确完成是必要的。通过使用 complete 方法和相应操作符，可以有效管理数据流的生命周期。

使用 complete() 方法

当创建自定义的 Observable 时，可以在适当的时刻调用 observer.complete() 来手动标记完成状态。这表示没有新的数据会被发射，并且数据流结束。

const { Observable } = require('rxjs')

const myObservable = new Observable((observer) => {
  // 发射一个值
  observer.next('Hello, RxJS!')

  // 标记完成
  observer.complete()
})

使用 finalize 操作符

finalize 操作符执行一些清理工作或副作用，无论Observable是成功完成还是因错误而终止。这对于释放资源特别有用。

import { interval } from 'rxjs'
import { take, finalize } from 'rxjs/operators'

interval(1000)
  .pipe(
    take(3),
    finalize(() => console.log('Cleaning up!'))
  )
  .subscribe({
    next: (value) => console.log(value),
    complete: () => console.log('Completed!'),
  })

使用 takeUntil 操作符

takeUntil 操作符可以帮助我们管理订阅的生命周期，确保适时取消订阅。

import { fromEvent, interval } from 'rxjs'
import { takeUntil } from 'rxjs/operators'

const stopButton = document.getElementById('stopButton')

// 创建点击事件和定时器 Observable
const stopClicks$ = fromEvent(stopButton, 'click')
const interval$ = interval(1000)

// 使用 takeUntil 确保在停止按钮点击时取消订阅
interval$.pipe(takeUntil(stopClicks$)).subscribe((value) => console.log(value)) // 点击停止按钮时，取消定时器的订阅
  1. 当停止按钮被点击时，stopClicks$ Observable 将发出事件并完成。
  2. 使用 takeUntil 操作符确保定时器 Observable 在按钮点击时取消订阅。

使用适当的调度器

RxJS 提供了多种调度器，可以控制 Observable 的执行上下文。选择适合的调度器有助于提升性能和资源利用率。

使用 asyncScheduler

asyncScheduler 可以在异步任务队列中调度任务，而不是自己实现setTimeout或setInterval，这对于时间控制和资源管理非常有用。

import { of, asyncScheduler } from 'rxjs'
import { observeOn } from 'rxjs/operators'

const observable$ = of('Hello', 'World').pipe(observeOn(asyncScheduler))

console.log('Before subscription')
observable$.subscribe((value) => console.log(value))
console.log('After subscription')

image.png

  1. observeOn 使用 asyncScheduler 将 Observable 的执行调度到异步队列。
  2. 这样可以确保执行顺序并提升性能。

操作符组合与流控制

在处理复杂的异步操作时，组合多个操作符和流控制技巧可以简化逻辑，提升性能。

使用 combineLatest 和 forkJoin

combineLatest 和 forkJoin 可以组合多个 Observable，分别适用于实时数据流和完成后的数据合并。 使用 combineLatest 进行实时数据流合并

import { combineLatest, interval } from 'rxjs'
import { map } from 'rxjs/operators'

const source1$ = interval(1000).pipe(map((val) => `Stream 1: ${val}`))
const source2$ = interval(1500).pipe(map((val) => `Stream 2: ${val}`))

combineLatest([source1$, source2$]).subscribe(([val1, val2]) => {
  console.log(val1, val2)
})

20240715205742_rec_.gif 使用 forkJoin 进行完成后的数据合并

import { forkJoin, of } from 'rxjs'
import { delay } from 'rxjs/operators'

const observable1$ = of('Hello').pipe(delay(1000))
const observable2$ = of('World').pipe(delay(2000))

forkJoin([observable1$, observable2$]).subscribe(([val1, val2]) => {
  console.log(`${val1} ${val2}`) // 输出: Hello World
})

20240715210018_rec_.gif

  1. combineLatest 在任意一个 Observable 发出值时，组合所有最新的值，适用于实时数据流联动。
  2. forkJoin 在所有 Observable 都完成时发出值，适用于需要等待所有异步操作完成后的场景。

谨慎使用 share 和 shareReplay

虽然 share 和 shareReplay 能够优化资源使用和性能，但在使用这些操作符时需要考虑正确的场景和上下文，避免潜在的问题如不当的缓存和内存泄漏。

import { of } from 'rxjs'
import { delay, shareReplay } from 'rxjs/operators'

const source$ = of('Cached Data').pipe(delay(1000), shareReplay(1))

source$.subscribe((value) => console.log('Subscriber 1:', value))
source$.subscribe((value) => console.log('Subscriber 2:', value))
// 输出:
// Subscriber 1: Cached Data
// Subscriber 2: Cached Data
  1. shareReplay 用于共享和重放Observable的值，适用于需要缓存数据的场景。
  2. 使用 shareReplay 时，确保配置合理的缓存策略和上下文，以避免内存泄漏。

结论

RxJS 是 JavaScript 中实现响应式编程的核心库，它提供了丰富的操作符和灵活的数据流控制能力。通过观察者模式和函数式编程范式，RxJS 让开发者能够高效地管理异步事件和数据流。

随着前端框架（如 Angular、React 和 Vue）的广泛采用，RxJS 在现代前端开发中的地位将更加巩固，成为高效管理应用状态和响应用户交互的工具。

通过系统学习 RxJS，我们能够在实际项目中应用其核心特性，提升代码的可读性、维护性和性能。在大型项目中，RxJS 可以用于高效处理复杂的异步操作，并结合其他库和框架，共同打造稳定和高性能的前端应用。

后记

在上面的章节中，我们详细介绍了 RxJS 的基础知识、操作符、订阅管理、高级主题以及其在实践项目中的应用。为了进一步提高代码的质量和开发效率，这里还包括一些额外的工具，帮助你更好地掌握和应用 RxJS。

observable-hooks

observable-hooks 是一个实用的库，用于在 React 应用中简单地使用 RxJS。它提供了几个自定义 hooks，使得在 React 组件中使用 RxJS 更加简洁和声明式。

npm install observable-hooks

useObservable：用于在组件中创建和订阅 Observable。

import React from 'react'
import { useObservable } from 'observable-hooks'
import { interval } from 'rxjs'
import { map } from 'rxjs/operators'

const TimerComponent = () => {
  const timer = useObservable(() => interval(1000).pipe(map((val) => `Seconds passed: ${val}`)))

  return <div>{timer}</div>
}

export default TimerComponent

useSubscription：用于在组件中订阅 Observable，并适时更新状态。

import React, { useState } from 'react'
import { useSubscription } from 'observable-hooks'
import { fromEvent } from 'rxjs'
import { map } from 'rxjs/operators'

const ClickTracker = () => {
  const [clickCount, setClickCount] = useState(0)

  useSubscription(fromEvent(document, 'click').pipe(map(() => setClickCount((count) => count + 1))))

  return <div>Document Clicks: {clickCount}</div>
}

export default ClickTracker

useEventCallback：用于创建事件回调，支持 RxJS 的操作符。

import React, { useRef } from 'react'
import { useEventCallback } from 'observable-hooks'
import { map } from 'rxjs/operators'

const InputEcho = () => {
  const inputRef = useRef()
  const [onInputChange, text] = useEventCallback(
    (event$) => event$.pipe(map((e) => e.target.value)),
    '' // 初始状态
  )

  return (
    <div>
      <input ref={inputRef} onChange={onInputChange} />
      <p>{text}</p>
    </div>
  )
}

export default InputEcho

rxjs-hooks

rxjs-hooks 是一个方便在 React 中使用 RxJS 的库。它提供了几个自定义 hook，使得在 React 组件中使用 RxJS 更加简洁和声明式。

npm install rxjs-hooks

import React from 'react'
import { useObservable } from 'rxjs-hooks'
import { interval } from 'rxjs'
import { map } from 'rxjs/operators'

const TimerComponent = () => {
  const timer = useObservable(() => interval(1000).pipe(map((val) => `Seconds passed: ${val}`)))

  return <div>{timer}</div>
}

export default TimerComponent

redux-observable

redux-observable 是一个中间件库，用于在 Redux 中使用 RxJS。它允许开发者使用 RxJS 创建复杂的异步数据流和副作用处理。

npm install redux-observable

import { combineEpics, createEpicMiddleware } from 'redux-observable'
import { filter, mapTo } from 'rxjs/operators'
import { createStore, applyMiddleware } from 'redux'
import { ofType } from 'redux-observable'

const pingEpic = (action$) => action$.pipe(ofType('PING'), mapTo({ type: 'PONG' }))

const epicMiddleware = createEpicMiddleware()
const rootEpic = combineEpics(pingEpic)

const rootReducer = (state = {}, action) => {
  switch (action.type) {
    case 'PING':
      return { ...state, isPinging: true }
    case 'PONG':
      return { ...state, isPinging: false }
    default:
      return state
  }
}

const store = createStore(rootReducer, applyMiddleware(epicMiddleware))
epicMiddleware.run(rootEpic)

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

附录

Discuss on Twitter • View on GitHub