使用 RxJS 库实现响应式编程

kongxx

发布于 2025-02-06 21:10:17

9400

代码可运行

文章被收录于专栏：有困难要上，没有困难创造困难也要上！

运行总次数：0

代码可运行

什么是 RxJS？

RxJS（Reactive Extensions for JavaScript）是一个用于响应式编程的库，它使得处理异步数据流变得更加简单和优雅。通过使用 Observables（可观察对象），你可以轻松地处理事件、HTTP 请求、定时器等异步数据源。

基本概念

在深入使用 RxJS 之前，我们需要了解几个基本概念：

Observable（可观察对象）：表示一个可以被观察的数据流。
Observer（观察者）：一个对象，它定义了如何在 Observable 发出新数据时做出反应。
Subscription（订阅）：当你订阅一个 Observable 时，你会得到一个 Subscription 对象，它可以用来取消订阅。
Operators（操作符）：用于处理 Observable 的函数，例如 map、filter、mergeMap 等。

安装 RxJS

npm install rxjs

一个简单例子

下面看一下怎么使用RxJS，首先我们可以使用 new Observable 来创建一个新的 Observable

import { Observable } from 'rxjs';

const observable = new Observable(subscriber => {
  subscriber.next('Hello');
  subscriber.next('World');
  subscriber.complete();
});

在这个例子中，我们创建了一个 Observable，它会依次发出 "Hello" 和 "World" 字符串，然后完成。

接下来我们就可以订阅 Observable ,响应事件了，这里只是打印事件消息。

observable.subscribe({
  next(x) { console.log('Received: ' + x); },
  error(err) { console.error('Error: ' + err); },
  complete() { console.log('Done'); }
});

执行上面的代码，我们会得到下面的输出结果：

Received: Hello
Received: World
Done

一个安装例子

这里看一个应用场景，我们需要执行一个安装操作，这个安装可能需要执行多个步骤，比如下载文件、解压文件、安装依赖、安装资源文件等，我们可以使用 RxJS 来处理这些步骤

这里我们假定要执行下面这些步骤

download(file: string)
uncompress(file: string, workDir: string)
installDependencies(workDir: string)
installResources(workDir: string)
clean(workDir: string)

import { Observable, of, from, interval } from 'rxjs';
import { map, take, concatMap, delay, tap, catchError } from 'rxjs/operators';

// 执行状态
interface InstallProcessing {
  success: boolean;
  message: string;
  progress: number;
}

class Installer {
  file = '/path/to/file';
  workDir = '/path/to/workdir';

  install(): Observable<InstallProcessing> {
    return new Observable<InstallProcessing>((observer) => {
      const steps = [
        {
          name: '下载安装包',
          method: () => this.download(this.file),
        },
        {
          name: '解压安装包',
          method:  () => this.uncompress(this.file, this.workDir),
        },
        {
          name: '安装依赖',
          method:  () => this.installDependencies(this.workDir),
        },
        {
          name: '安装资源文件',
          method:  () => this.installResources(this.workDir),
        },
        {
          name: '清理',
          method:  () => this.clean(this.workDir),
        },
      ];

      const totalSteps = steps.length;
      let completedSteps = 0;

      const executeStep = async (index: number) => {
        if (index < totalSteps) {
          observer.next({
            success: false,
            message: `开始执行: ${steps[index].name}`,
            progress: (completedSteps / totalSteps) * 100,
          });

          try {
            await steps[index].method();

            completedSteps++;
            observer.next({
              success: completedSteps === totalSteps,
              message: `${(completedSteps / totalSteps) * 100} 完成`,
              progress: (completedSteps / totalSteps) * 100,
            });

            executeStep(index + 1); // 执行下一步骤
          } catch (error) {
            observer.next({
              success: false,
              message: `步骤失败: ${steps[index].name} - ${error.message}`,
              progress: (completedSteps / totalSteps) * 100,
            });
            observer.complete();
          }
        } else {
          observer.complete();
        }
      };

      executeStep(0); // 从第一个步骤开始
    });
  }

  async download(file: string): Promise<void> {
    // 模拟异步下载安装包
    console.log(`Downloading ${file}...`);
    await new Promise((resolve) => setTimeout(resolve, 1000)); // 模拟下载延迟
  }

  async uncompress(file: string, workDir: string): Promise<void> {
    // 模拟解压安装包
    console.log(`Uncompressing ${file} in ${workDir}...`);
    await new Promise((resolve) => setTimeout(resolve, 1000)); // 模拟解压延迟
  }

  async installDependencies(workDir: string): Promise<void> {
    // 模拟安装脚本的逻辑
    console.log(`Installing dependencies in ${workDir}...`);
    await new Promise((resolve) => setTimeout(resolve, 1000)); // 模拟安装延迟
  }

  async installResources(workDir: string): Promise<void> {
    // 模拟安装定时任务的逻辑
    console.log(`Installing resources in ${workDir}...`);
    await new Promise((resolve) => setTimeout(resolve, 1000)); // 模拟安装延迟
  }

  clean(workDir: string): void {
    // 模拟清理的逻辑
    console.log(`Cleaning up ${workDir}...`);
  }
}

使用下面的代码执行上面的安装过程并处理安装事件

const installer = new Installer();
installer.install().subscribe({
  next: (result) => console.log(`进度: ${result.progress}% - ${result.message}`),
  error: (err) => console.error(err),
  complete: () => console.log('Installation process completed'),
});

执行结果如下:

进度: 0% - 开始执行: 下载安装包
Downloading /path/to/file...
进度: 20% - 20 完成
进度: 20% - 开始执行: 解压安装包
Uncompressing /path/to/file in /path/to/workdir...
进度: 40% - 40 完成
进度: 40% - 开始执行: 安装依赖
Installing dependencies in /path/to/workdir...
进度: 60% - 60 完成
进度: 60% - 开始执行: 安装资源文件
Installing resources in /path/to/workdir...
进度: 80% - 80 完成
进度: 80% - 开始执行: 清理
Cleaning up /path/to/workdir...
进度: 100% - 100 完成
Installation process completed

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2025-02-06，如有侵权请联系 cloudcommunity@tencent.com 删除

rxjs