当前位置：首页 > news >正文

Nestjs框架: RxJS 核心方法实践与错误处理详解

news 2025/7/20 7:47:21

RxJS 核心方法实践与错误处理详解

围绕 RxJS 的核心方法进行实操
涵盖 observable 的构建、定时器的使用、错误处理机制（如 retry 和 catchError）、以及 defer 和 lastValueFrom 的使用场景

1 ）RxJS 基础实践：Timer 的使用

我们首先在项目中使用 RxJS 的 Observable 和 timer 方法来进行实际操作。由于项目中已通过 netDS 依赖引入了 RxJS，因此无需额外安装。

我们新建一个名为 indexMGS 的文件，并在其中导入以下内容：

import { Observable } from 'rxjs';
import { timer } from 'rxjs';

随后，我们定义一个定时器：

const exampleTimer = timer(2000); // 2秒后触发

然后订阅该定时器：

exampleTimer.subscribe({next: (value) => console.log('timeremitted', value),complete: () => console.log('timercomplete')
});

这段代码会在两秒后输出 timeremitted 0，并在完成后打印 timercomplete。

2 ）错误处理：Retry 与 CatchError 的配合使用

接下来我们演示如何使用 RxJS 提供的错误处理 API，包括 throwError、retry 和 catchError。

我们定义一个 observable，主动抛出错误：

const errorObservable = new Observable((subscriber) => {subscriber.error(new Error('this is an error'));
});

通过 pipe 方法结合 retry(3) 和 catchError 实现重试三次并捕获最终错误：

import { retry, catchError } from 'rxjs/operators';errorObservable.pipe(retry(3),catchError((err) => {console.log('caught error:', err.message);return throwError(() => new Error('after retries'));})
).subscribe({error: (error) => console.log('subscribe error final:', error.message)
});

在测试过程中，“retry”被执行了4次，这是因为第一次执行也算作一次尝试，之后才进行三次重试
这一机制非常适用于网络请求或数据库连接等需要自动重试的场景

3 ）延迟创建：使用 Defer 与 LastValueFrom 获取最终值

第三个示例中，我们将使用 defer 和 lastValueFrom 来演示延迟创建 observable 并获取其最后一个值。

我们定义一个 defer observable：

import { defer } from 'rxjs';const deferredObservable = defer(() => {console.log('observable created');return new Observable((subscriber) => {subscriber.next('hello');subscriber.complete();});
});

接着使用 lastValueFrom 获取 observable 的最后一个值：

import { lastValueFrom } from 'rxjs';async function getDeferredValue() {const result = await lastValueFrom(deferredObservable);console.log('deferred value:', result);
}

运行后，控制台将输出：

observable created
deferred value: hello

lastValueFrom 的作用是将 observable 转换为 promise，并返回其最后一个值
它会自动订阅该 observable 并等待其完成

4 ) 结合 Timer 实现延迟输出

示例让 hello 在两秒后输出，实现方式如下：

subscriber.next('hello three');
subscriber.complete();

修改为使用 timer 延迟执行：

timer(2000).subscribe(() => {subscriber.next('hello three');subscriber.complete();
});

运行后，observable created 会立即输出，而 deferred value: hello three 将在两秒后打印

总结

通过以上三个示例，实现了 RxJS 的几个核心方法：
- 1. Timer：用于定时触发 observable 的响应。
- 1. Retry 与 CatchError：实现错误自动重试及最终错误捕获。
- 1. Defer 与 LastValueFrom：延迟创建 observable 并获取其最终值。
这些方法在实际开发中具有广泛的应用价值
例如在处理网络请求、数据库连接、异步任务调度等场景时
能够极大地提升代码的健壮性与可维护性
建议大家结合官方 RxJS 文档，进一步深入理解这些操作符的原理与使用技巧
并将其应用到实际项目中，例如 freestmas 模型的构建与优化