什么是 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
